""" CSV export module — CSV export functionality for BetaSpectrumAnalyzer. Column semantics (v0.4.0 onward): - The basic spectrum is ALWAYS core = phase_space x fermi; the Fermi function is a core spectrum component, never an optional correction. Higher-order corrections (finite size, charge distribution, screening, exchange, radiative) are multiplied on top of the core. - ``branch_{i}_spectrum`` columns carry the PHYSICAL branch spectrum (core x all enabled corrections), weighted by intensity and area-normalized so that the sum of branch columns equals total_spectrum (the total column itself is normalized to integrate to 1). In the legacy export branch columns contained only the per-branch phase-space x radiative product WITHOUT the Fermi function — physically wrong. - ``normalize_to_one`` mode: each branch column is normalized to integrate to 1 individually, intensity is ignored, no total column. """ from __future__ import annotations from datetime import datetime, timezone from typing import TYPE_CHECKING, Dict import numpy as np import pandas as pd from beta_spectrum.logging_utils import get_git_short_hash if TYPE_CHECKING: from beta_spectrum.output.core import BetaSpectrumAnalyzer class CSVEmitter: """CSV export functionality for BetaSpectrumAnalyzer.""" @staticmethod def export_to_csv( analyzer: BetaSpectrumAnalyzer, filename: str, source_type: str = "unknown", csv_verbosity: str = "full", ) -> None: """Export spectrum data to CSV with metadata header. Parameters ---------- analyzer : BetaSpectrumAnalyzer The analyzer instance to export data from. filename : str Output CSV file path. source_type : str Source type (paceENSDF, json, etc.). csv_verbosity : str Verbosity level: 'full' (all components) or 'concise' (total spectrum and physical branch spectra only). Raises ------ ValueError If every branch spectrum integrates to a non-positive area (e.g. all branch endpoints lie below the neutrino mass). """ components = analyzer.components config = analyzer.config to_one = config.normalize_to_one n_branches = analyzer._n_branches # Physical branch spectra: core(phase_space x fermi) x corrections. physical = [analyzer.physical_branch_spectrum(i) for i in range(n_branches)] # Only higher-order corrections are toggleable; list them enabled. enabled = [] if config.use_screening: enabled.append("screening") if config.use_finite_size: enabled.append("finite_size") if config.use_charge_dist: enabled.append("charge_dist") if config.use_radiative: enabled.append("radiative") if config.use_exchange: enabled.append("exchange") # The header endpoint must describe the spectrum that is actually # computed. Single mode with exactly one configured branch follows # the branch (the top-level endpoint may be stale from the parent # decay, e.g. after intensity-cutoff filtering). basic_endpoint_MeV = config.endpoint_MeV if ( not analyzer._is_multi and config.branches is not None and len(config.branches) == 1 ): basic_endpoint_MeV = analyzer._unified_branches[0].endpoint_MeV parent_symbol = analyzer._element_symbol(config.Z_parent) daughter_symbol = analyzer._element_symbol(config.Z_daughter) git_commit = get_git_short_hash() if source_type == "paceENSDF": source_line = "# source: paceENSDF (https://pypi.org/project/paceENSDF/)" else: source_line = f"# source: {source_type}" header_lines = [ f"# beta-spectrum v{__import__('beta_spectrum').__version__}" f"(build.{git_commit})", f"# timestamp: {datetime.now(timezone.utc).isoformat()}", source_line, f"# nuclide: {parent_symbol}{config.A_number} " f"-> {daughter_symbol}{config.A_number} " f"(Z={config.Z_parent}->{config.Z_daughter})", f"# basic_endpoint: {basic_endpoint_MeV * 1000:.1f} keV", ] if config.neutrino_mass_MeV > 0: header_lines.append( f"# neutrino_mass_MeV: {config.neutrino_mass_MeV:.6f} " f"(non-zero neutrino mass effects included)" ) else: header_lines.append(f"# neutrino_mass_MeV: {config.neutrino_mass_MeV:.6f}") header_lines.append("# core: phase_space x fermi (basic spectrum)") header_lines.append(f"# corrections: {', '.join(enabled)}") header_lines.append(f"# e_step: {config.e_step_MeV:.4f} MeV") header_lines.append(f"# n_points: {len(analyzer.energies_MeV)}") if to_one: header_lines.append( "# branch normalization: to_one " "(each branch integrates to 1, intensity ignored)" ) else: header_lines.append( "# branch normalization: intensity " "(sum of branch columns == total_spectrum)" ) # Separate branches into active and excluded lists active_branches = [] excluded_branches = [] for i, branch in enumerate(analyzer._unified_branches): branch_endpoint_keV = branch.endpoint_MeV * 1000 neutrino_mass_keV = config.neutrino_mass_MeV * 1000 if branch_endpoint_keV <= neutrino_mass_keV: excluded_branches.append( f" branch_{i + 1}: E₀={branch_endpoint_keV:.1f} keV, " f"transition={branch.transition_type}, " f"intensity={branch.intensity:.4f} - excluded: endpoint < m_nu" ) elif branch.intensity < config.intensity_cutoff: excluded_branches.append( f" branch_{i + 1}: E₀={branch_endpoint_keV:.1f} keV, " f"transition={branch.transition_type}, " f"intensity={branch.intensity:.4f} " "- excluded: intensity < cutoff" ) else: active_branches.append( f" branch_{i + 1}: E₀={branch_endpoint_keV:.1f} keV, " f"transition={branch.transition_type}, " f"intensity={branch.intensity:.4f}" ) if active_branches: header_lines.append("# active branches:") header_lines.extend(f"# {b}" for b in active_branches) if excluded_branches: header_lines.append("# excluded branches:") header_lines.extend(f"# {b}" for b in excluded_branches) # ----- data columns ----- data: Dict[str, np.ndarray] = { "energy_MeV": analyzer.energies_MeV, } # Positive-integral guard applies to BOTH normalization modes: a # degenerate spectrum must never be written (not even as zeros). total_raw = analyzer.spectrum(analyzer.W) integral = float(np.trapezoid(total_raw, analyzer.energies_MeV)) if integral <= 0.0: raise ValueError( "Total spectrum integral is non-positive: every branch " "is degenerate (check neutrino_mass vs branch endpoints " "and intensity_cutoff)" ) if not to_one: data["total_spectrum"] = total_raw / integral for i in range(n_branches): key_i = i + 1 if csv_verbosity == "concise": col_name = f"branch_{key_i}" else: col_name = f"branch_{key_i}_spectrum" if to_one: area = float(np.trapezoid(physical[i], analyzer.energies_MeV)) if area > 0.0: data[col_name] = physical[i] / area else: data[col_name] = np.zeros_like(physical[i]) else: intensity_i = analyzer._unified_branches[i].intensity data[col_name] = physical[i] * intensity_i / integral if csv_verbosity == "full": # Per-branch component columns (factorized ingredients). for i in range(n_branches): for comp_name in ["PhaseSpace", "Radiative"]: key = analyzer._get_component_key(i, comp_name) if key in components: data[f"branch_{i + 1}_{comp_name}"] = components[key] # Universal components (no prefix): Fermi plus universal # higher-order corrections. for name, values in components.items(): if not name.startswith("branch_"): data[name] = values df = pd.DataFrame(data) separator = "\t" if config.csv_separator == "tab" else config.csv_separator try: with open(filename, "w") as f: f.write("\n".join(header_lines) + "\n") df.to_csv(f, index=False, float_format="%.4e", sep=separator) except OSError as exc: raise ValueError(f"Cannot write spectrum CSV {filename}: {exc}") from exc if analyzer._logger: analyzer._logger.info("CSV exported to %s", filename)