unzhakov
2788ea0683
- --normalize-to-one renamed to --separate-branches (no alias; the name now describes the output choice: total spectrum with intensity-normalized branches vs separate branches, each normalized to unit area with intensities left to the data user). SpectrumConfig.normalize_to_one -> separate_branches, CSV header word to_one -> separate. - every long option gains a single-letter alias: -N -i -s -m -o -p -L -d -c -V -S -b (only -v/-q existed before) - dry-run drops the top-level Transition: line — the per-branch list below already reports each branch's transition type - version bumped to 0.4.1 (pyproject, __version__, CSV semantics docstring, TODO roadmap entry)
243 lines
9.7 KiB
Python
243 lines
9.7 KiB
Python
"""
|
|
CSV export module — CSV export functionality for BetaSpectrumAnalyzer.
|
|
|
|
Column semantics (v0.4.1 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.
|
|
- ``separate_branches`` 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
|
|
separate = config.separate_branches
|
|
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 separate:
|
|
header_lines.append(
|
|
"# branch normalization: separate "
|
|
"(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 separate:
|
|
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 separate:
|
|
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).
|
|
# Only multi-branch mode carries per-branch factors: in single
|
|
# mode the bare PhaseSpace/Radiative columns below already
|
|
# describe the single spectrum, and prefixed duplicates would
|
|
# silently disagree with them once renamed.
|
|
if analyzer._is_multi:
|
|
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. In single mode the core factors
|
|
# (PhaseSpace, Radiative) also appear here unprefixed.
|
|
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)
|