Compare commits

..
10 Commits
Author SHA1 Message Date
unzhakov 67de4e3baa style: canonicalize black formatting on test files 2026-10-03 01:31:35 +03:00
unzhakov 2788ea0683 feat(cli): separate-branches rename, short aliases, clean dry-run
- --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)
2026-10-03 01:08:51 +03:00
unzhakov 7f47a876ae style: apply black formatting to test files 2026-10-02 14:32:04 +03:00
unzhakov 7e31ac4b4e docs: reflect normalization fixes, detector cleanup and CLI overrides
- README: remove stale detector_response.py tree entry and the
  detector-models highlight (module removed); document per-branch
  measured transition endpoints from paceENSDF and JSON-flow CLI
  override semantics
- TODO: drop the DEFAULT_JSON_SCHEMA detector-keys clause (removed)
- data/custom_input_example.json: strip dead detector_* keys
2026-10-02 12:01:45 +03:00
unzhakov 36174b5d25 refactor: remove stale detector schema keys and single-mode duplicates
- DEFAULT_JSON_SCHEMA: drop the 8 detector_* keys left over from the
  removed detector_response module; CLI dry-run drops the dead
  'Detector: not configured (Phase 4)' line
- CSV export: single-mode no longer writes branch_1_PhaseSpace /
  branch_1_Radiative duplicates next to the bare PhaseSpace/Radiative
  columns; per-branch core factors only in multi-branch mode
- Visualizer: debug panels fall back to the unprefixed core-factor
  columns for single-mode CSVs instead of skipping them
- Spectrum ylabel now states the actual units: normalized spectrum
  [MeV^-1], not 'Normalized Counts / keV'
- tests: defaults assert detector keys gone; column layout pinned for
  single vs multi; panel and ylabel assertions
2026-10-02 10:21:07 +03:00
unzhakov 3590ca91a1 fix(nuclear_data): use measured transition endpoints per branch
paceENSDF stores the measured beta endpoint energy of each transition
(ENSDF column 7). The parser instead derived branch endpoints as
base-endpoint minus level energy with the base taken from an arbitrary
first-listed branch, so Co60 produced negative branch endpoints and a
hard ValueError (bs_pnpi Co60 exited 1 regardless of cutoff).

- BranchInfo carries the measured endpoint_MeV
- overall endpoint = max measured endpoint over all branches
- decay_info_to_config prefers measured endpoints; the subtraction is
  retained only as a legacy fallback for hand-built DecayInfo objects
- Co60/Tc99/Cs137 endpoint tests + CLI dry-run/export tests (D3)
2026-10-02 08:42:51 +03:00
unzhakov 9c3dd80b5c refactor(cli): delegate paceENSDF config building to library
Remove the duplicated inline paceENSDF config builder in the CLI.
It renormalized branch intensities before filtering, diverging from
the library decay_info_to_config path that library callers use
(audit finding D5). CLI paceENSDF configs now come from
create_config_from_source exactly once per run.

Regression test spies on create_config_from_source during a CLI run.
2026-10-02 07:13:14 +03:00
unzhakov d166532779 test: cover to_one/concise/single/plot normalization combinations 2026-10-02 06:31:38 +03:00
unzhakov 18a46ad02f feat(cli): propagate export options through JSON/paceENSDF flows
- main(argv) signature enables in-process CLI invocation and testing
- parser defaults for --e-step/--neutrino-mass/--intensity-cutoff/
  --csv-separator/--normalize-to-one now None = 'not provided'
- JSON flow applies file values by default and CLI flags as overrides
  (previously these five options were silently discarded)
- paceENSDF flow resolves None flags to literal defaults
- new tests/integration/test_cli_flows.py: JSON propagation,
  degenerate-mode non-zero exit with no file (JSON + paceENSDF)
- adapted two neutrino-mass default tests to None-parsing semantics
2026-10-02 05:46:08 +03:00
unzhakov 62bf7fdac2 fix(output): apply positive-integral export guard to to_one mode
CSVEmitter.export_to_csv raised ValueError only in intensity mode;
degenerate exports with --normalize-to-one silently wrote all-zero
CSV files. The guard now runs before the mode split and no file is
written in either mode.
2026-10-01 23:02:33 +03:00
17 changed files with 759 additions and 173 deletions
+5 -6
View File
@@ -60,8 +60,7 @@ beta_shape_pnpi/
│ │ ├── finite_size.py ✓ L0 expansion + charge distribution U term │ │ ├── finite_size.py ✓ L0 expansion + charge distribution U term
│ │ ├── screening.py ✓ Atomic electron screening (ratio method) │ │ ├── screening.py ✓ Atomic electron screening (ratio method)
│ │ ├── exchange.py ✓ Hayen-2018 empirical fit coefficients │ │ ├── exchange.py ✓ Hayen-2018 empirical fit coefficients
│ │ ├── radiative.py ✓ Outer radiative corrections, soft-photon resummation │ │ └── radiative.py ✓ Outer radiative corrections, soft-photon resummation
│ │ └── detector_response.py ✓ Analytical detector response models
├── beta_spectrum/output/ ← output, plotting, visualization, CSV export ├── beta_spectrum/output/ ← output, plotting, visualization, CSV export
│ ├── __init__.py # BetaSpectrumAnalyzer, CSVEmitter, Visualizer │ ├── __init__.py # BetaSpectrumAnalyzer, CSVEmitter, Visualizer
│ ├── core.py # BetaSpectrumAnalyzer base class │ ├── core.py # BetaSpectrumAnalyzer base class
@@ -133,7 +132,7 @@ In `beta_spectrum.output`. Provides debugging and visualization tools:
- `export_to_csv(filename)` — export via pandas DataFrame with YAML-style metadata header: - `export_to_csv(filename)` — export via pandas DataFrame with YAML-style metadata header:
- the basic spectrum is always the core `phase_space × fermi` (the header carries a `# core:` line); finite size, charge distribution, screening, exchange and radiative are higher-order corrections - the basic spectrum is always the core `phase_space × fermi` (the header carries a `# core:` line); finite size, charge distribution, screening, exchange and radiative are higher-order corrections
- `branch_{i}_spectrum` columns contain the *physical* branch spectra weighted by intensity and normalized so that the sum of branch columns equals `total_spectrum` - `branch_{i}_spectrum` columns contain the *physical* branch spectra weighted by intensity and normalized so that the sum of branch columns equals `total_spectrum`
- with `normalize_to_one=True` each branch column integrates to 1 (intensity ignored, no total column) - with `separate_branches=True` (CLI `--separate-branches`) each branch column integrates to 1 (intensity ignored, no total_spectrum column)
- `parse_spectrum_csv(path)` — read an exported CSV and its metadata header (module function in `beta_spectrum.output.visualization`) - `parse_spectrum_csv(path)` — read an exported CSV and its metadata header (module function in `beta_spectrum.output.visualization`)
- `get_data()` — return all numerical data for custom analysis - `get_data()` — return all numerical data for custom analysis
@@ -191,9 +190,9 @@ ______________________________________________________________________
For a complete list of implemented and planned features, see [TODO.md](TODO.md). Key highlights: For a complete list of implemented and planned features, see [TODO.md](TODO.md). Key highlights:
- **Physics corrections**: phase space, Fermi function, finite size, screening, exchange, radiative (with delta_cut resummation) - **Physics corrections**: phase space, Fermi function, finite size, screening, exchange, radiative (with delta_cut resummation)
- **Detector response**: Gaussian, Gaussian+tail, Tikhonov models with tabulated support - **Detector response**: not implemented — `SpectrumModel` keeps a duck-typed optional convolution hook; the analytical models were removed (restore-or-delete decision tracked in TODO.md)
- **Multi-branch decay**: intensity-weighted sum of branch spectra, per-branch calculators with branch-specific endpoint, transition type (from ENSDF), and `--intensity-cutoff` CLI filter - **Multi-branch decay**: intensity-weighted sum of branch spectra, per-branch calculators with branch-specific endpoint, transition type (from ENSDF), and `--intensity-cutoff` CLI filter
- **Analysis tools**: χ² fitter (SpectrumFitter + CurveFitter), C(W) extraction pipeline, Kurie plot analysis, g_V/g_A extraction - **Analysis tools**: χ² fitter (SpectrumFitter + CurveFitter), C(W) extraction pipeline, Kurie plot analysis, g_V/g_A extraction
- **Experimental data**: ExpSpectrum container, energy calibration, peak fitting, dead-time/pile-up correction - **Experimental data**: ExpSpectrum container, energy calibration, peak fitting, dead-time/pile-up correction
- **CLI**: `bs_pnpi --nuclide Tc99` — full paceENSDF integration (auto-lookup of Z, A, Q-value, transition type, half-life, branches, forbiddenness from ENSDF) - **CLI**: `bs_pnpi --nuclide Tc99` — full paceENSDF integration (auto-lookup of Z, A, Q-value, transition type, half-life, branches, forbiddenness from ENSDF); for JSON input the file's values are used and CLI flags (`--e-step`, `--neutrino-mass`, `--intensity-cutoff`, `--csv-separator`, `--separate-branches`) override them only when explicitly given; every long option has a single-letter alias (`-N`, `-i`, `-s`, `-m`, `-o`, `-p`, `-L`, `-d`, `-c`, `-V`, `-S`, `-b`)
- **paceENSDF integration**: `get_decay_info_from_paceENSDF()`, `create_config_from_source("paceENSDF", nuclide="Tc99")`, `_parse_nuclide_symbol()` for Z=1..98, FORBIDDENNESS_MAP, `--nuclide` CLI flag - **paceENSDF integration**: `get_decay_info_from_paceENSDF()`, `create_config_from_source("paceENSDF", nuclide="Tc99")`, `_parse_nuclide_symbol()` for Z=1..98, FORBIDDENNESS_MAP, `--nuclide` CLI flag; per-branch endpoints are the measured transition endpoint energies from ENSDF (max endpoint across branches defines the spectrum endpoint)
+12 -5
View File
@@ -32,7 +32,7 @@ beta_spectrum/ ← theory-only core
│ ├── screening.py │ ├── screening.py
│ ├── exchange.py │ ├── exchange.py
│ ├── radiative.py │ ├── radiative.py
│ └── detector_response.py ← stays here (part of theory) │ └── (detector_response.py — removed; restore-or-delete pending below)
└── nuclear_data.py ← paceENSDF integration └── nuclear_data.py ← paceENSDF integration
exp_data/ ← experimental data handling exp_data/ ← experimental data handling
@@ -102,9 +102,10 @@ output/ ← output, visualization, CSV export
- [x] README updated with new architecture diagram and Quick Start examples - [x] README updated with new architecture diagram and Quick Start examples
- [x] Version bumped to 0.4.0 - [x] Version bumped to 0.4.0
- [ ] Decide: restore or delete the `detector_response` module — it was removed from - [ ] Decide: restore or delete the `detector_response` module — it was removed from
`beta_spectrum/components/` together with its tests, but `fitter/model.py` keeps a `beta_spectrum/components/` together with its tests; `fitter/model.py` keeps a
duck-typed optional hook and `DEFAULT_JSON_SCHEMA` keeps the detector keys; duck-typed optional hook only (`DEFAULT_JSON_SCHEMA` detector keys and the CLI
restore from git history (commits up to `5f99566`) if convolution is needed again dry-run detector line were removed with audit cleanup).
Restore from git history (commits up to `5f99566`) if convolution is needed again
______________________________________________________________________ ______________________________________________________________________
@@ -321,7 +322,7 @@ ______________________________________________________________________
## Current Status ## Current Status
**Version:** 0.3.1 → **In progress: 0.4.0 — Package restructuring** **Version:** 0.4.0 → **v0.4.1 — CLI options refresh & normalization fixes** (released on `dev/core-fermi-csv`)
### Completed (v0.3.x) ### Completed (v0.3.x)
@@ -353,6 +354,12 @@ ______________________________________________________________________
- README updated with new architecture diagram and Quick Start examples - README updated with new architecture diagram and Quick Start examples
- Version bumped to 0.4.0 - Version bumped to 0.4.0
### Completed (v0.4.1)
- Core spectrum made unconditional (`phase_space × fermi`): core toggles removed, physical branch spectra in CSV export, plots built from the exported CSV
- Normalization audit fixes: lone surviving branch defines the spectrum, CLI options propagate through the JSON flow, positive-integral guard for both export modes, paceENSDF per-branch measured endpoints (Co60 fixed), single delegated paceENSDF config builder
- CLI surface refresh: `--normalize-to-one` renamed to `--separate-branches` (no alias), single-letter aliases for every option, dry-run drops the redundant top-level `Transition:` line, stale detector_* JSON schema keys removed
### Next Immediate Step ### Next Immediate Step
- **Track A: ⁹⁹Tc spectrum-shape analysis** — start real data processing pipeline - **Track A: ⁹⁹Tc spectrum-shape analysis** — start real data processing pipeline
+1 -1
View File
@@ -52,4 +52,4 @@ __all__ = [
"DEFAULT_JSON_SCHEMA", "DEFAULT_JSON_SCHEMA",
] ]
__version__ = "0.4.0" __version__ = "0.4.1"
+65 -69
View File
@@ -65,12 +65,14 @@ def _build_parser() -> argparse.ArgumentParser:
# Source selection flags (backward compatible, mutually exclusive) # Source selection flags (backward compatible, mutually exclusive)
source_group = parser.add_mutually_exclusive_group(required=False) source_group = parser.add_mutually_exclusive_group(required=False)
source_group.add_argument( source_group.add_argument(
"-N",
"--nuclide", "--nuclide",
type=str, type=str,
help="Nuclide symbol for paceENSDF lookup (e.g. 'Tc99', 'Co60'). " help="Nuclide symbol for paceENSDF lookup (e.g. 'Tc99', 'Co60'). "
"Retrieves decay parameters from the ENSDF database via paceENSDF.", "Retrieves decay parameters from the ENSDF database via paceENSDF.",
) )
source_group.add_argument( source_group.add_argument(
"-i",
"--input", "--input",
type=str, type=str,
metavar="FILE", metavar="FILE",
@@ -79,20 +81,27 @@ def _build_parser() -> argparse.ArgumentParser:
# Common parameters # Common parameters
parser.add_argument( parser.add_argument(
"-s",
"--e-step", "--e-step",
type=float, type=float,
default=0.001, default=None,
help="Energy step size in MeV (default: 0.001). All energies are in MeV throughout the toolkit.", help="Energy step size in MeV (default: 0.001 for nuclide input; "
"for JSON input the file's e_step_MeV is used when the flag is "
"omitted). All energies are in MeV throughout the toolkit.",
) )
parser.add_argument( parser.add_argument(
"-m",
"--neutrino-mass", "--neutrino-mass",
type=float, type=float,
default=0.0, default=None,
help="Neutrino mass in MeV (default: 0.0). Set to non-zero value to include neutrino mass effects in phase space calculation.", help="Neutrino mass in MeV (default: 0.0, or the JSON file's "
"neutrino_mass_MeV when omitted). Set to non-zero value to include "
"neutrino mass effects in phase space calculation.",
) )
# Output # Output
parser.add_argument( parser.add_argument(
"-o",
"--output", "--output",
type=str, type=str,
metavar="FILE", metavar="FILE",
@@ -101,6 +110,7 @@ def _build_parser() -> argparse.ArgumentParser:
# Plotting # Plotting
parser.add_argument( parser.add_argument(
"-p",
"--plot", "--plot",
type=str, type=str,
metavar="FILE", metavar="FILE",
@@ -124,6 +134,7 @@ def _build_parser() -> argparse.ArgumentParser:
help="Suppress all terminal output (only errors shown).", help="Suppress all terminal output (only errors shown).",
) )
log_group.add_argument( log_group.add_argument(
"-L",
"--log-file", "--log-file",
type=str, type=str,
metavar="PATH", metavar="PATH",
@@ -131,20 +142,23 @@ def _build_parser() -> argparse.ArgumentParser:
"Log file is always opened in append mode at DEBUG level.", "Log file is always opened in append mode at DEBUG level.",
) )
log_group.add_argument( log_group.add_argument(
"-d",
"--dry-run", "--dry-run",
action="store_true", action="store_true",
help="Validate input and display resolved configuration without computing spectrum.", help="Validate input and display resolved configuration without computing spectrum.",
) )
log_group.add_argument( log_group.add_argument(
"-c",
"--intensity-cutoff", "--intensity-cutoff",
type=float, type=float,
default=0.0, default=None,
metavar="FRAC", metavar="FRAC",
help="Minimum branch intensity fraction to include in the spectrum " help="Minimum branch intensity fraction to include in the spectrum "
"(default: 0.0 = include all branches). " "(default: 0.0 = include all branches). "
"Branches below this threshold are excluded from the calculation and plots.", "Branches below this threshold are excluded from the calculation and plots.",
) )
log_group.add_argument( log_group.add_argument(
"-V",
"--csv-verbosity", "--csv-verbosity",
type=str, type=str,
default="full", default="full",
@@ -152,15 +166,20 @@ def _build_parser() -> argparse.ArgumentParser:
help="CSV output verbosity: 'full' (all components) or 'concise' (total spectrum and normalized branch components only) (default: full)", help="CSV output verbosity: 'full' (all components) or 'concise' (total spectrum and normalized branch components only) (default: full)",
) )
log_group.add_argument( log_group.add_argument(
"-S",
"--csv-separator", "--csv-separator",
type=str, type=str,
default=",", default=None,
help="CSV column separator (default: ','). Use 'tab' for tab separation.", help="CSV column separator (default: ','). Use 'tab' for tab separation.",
) )
log_group.add_argument( log_group.add_argument(
"--normalize-to-one", "-b",
"--separate-branches",
action="store_true", action="store_true",
help="Normalize each branch to 1, ignoring intensity. In this case, total spectrum will not be produced (only energy and branch-wise spectra).", default=None,
help="Emit separate branch spectra, each normalized to integrate to "
"1 (intensities ignored; no total_spectrum column). Default: total "
"spectrum with intensity-normalized branch columns.",
) )
return parser return parser
@@ -173,15 +192,10 @@ def _run(args: argparse.Namespace) -> None:
from beta_spectrum.logging_utils import get_logger from beta_spectrum.logging_utils import get_logger
from beta_spectrum.nuclear_data import ( from beta_spectrum.nuclear_data import (
create_config_from_source, create_config_from_source,
get_decay_info_from_paceENSDF,
json_to_config, json_to_config,
load_json_input, load_json_input,
) )
from beta_spectrum.spectrum import ( from beta_spectrum.spectrum import BetaSpectrum
BetaSpectrum,
BranchConfig,
SpectrumConfig,
)
from beta_spectrum.output import BetaSpectrumAnalyzer from beta_spectrum.output import BetaSpectrumAnalyzer
# Determine logging level # Determine logging level
@@ -230,66 +244,45 @@ def _run(args: argparse.Namespace) -> None:
else: else:
source_type = "paceENSDF" source_type = "paceENSDF"
# Build config # Resolve optional CLI flags: None means "not provided". For paceENSDF
# they fall back to literal defaults; for JSON they let the file's own
# values through unless explicitly overridden.
e_step_MeV = 0.001 if args.e_step is None else float(args.e_step)
neutrino_mass_MeV = 0.0 if args.neutrino_mass is None else float(args.neutrino_mass)
intensity_cutoff = (
0.0 if args.intensity_cutoff is None else float(args.intensity_cutoff)
)
csv_separator = "," if args.csv_separator is None else args.csv_separator
separate_branches = bool(args.separate_branches)
# Build config — a single delegated path for every data source.
config: SpectrumConfig config: SpectrumConfig
if source_type == "paceENSDF": if source_type == "paceENSDF":
logger.info("Source: paceENSDF nuclide=%s", source_value) logger.info("Source: paceENSDF nuclide=%s", source_value)
decay_info = get_decay_info_from_paceENSDF(source_value, "beta_minus")
# Normalize branch intensities to sum to 1.0
if decay_info.branches:
total_intensity = sum(b.intensity for b in decay_info.branches)
if total_intensity > 0:
branches = [
BranchConfig(
endpoint_MeV=(
(decay_info.endpoint_MeV - b.level_energy_keV / 1000.0)
if b.level_energy_keV > 0
else decay_info.endpoint_MeV
),
transition_type=b.transition_type,
intensity=b.intensity / total_intensity,
)
for b in decay_info.branches
if b.intensity / total_intensity >= args.intensity_cutoff
]
config = SpectrumConfig(
Z_parent=decay_info.Z_parent,
Z_daughter=decay_info.Z_daughter,
A_number=decay_info.A_number,
endpoint_MeV=decay_info.endpoint_MeV,
transition_type=decay_info.transition_type,
e_step_MeV=args.e_step,
neutrino_mass_MeV=args.neutrino_mass,
branches=branches,
intensity_cutoff=args.intensity_cutoff,
csv_separator=args.csv_separator,
normalize_to_one=args.normalize_to_one,
)
else:
config = create_config_from_source( config = create_config_from_source(
"paceENSDF", "paceENSDF",
nuclide=source_value, nuclide=source_value,
e_step_MeV=args.e_step, e_step_MeV=e_step_MeV,
intensity_cutoff=args.intensity_cutoff, intensity_cutoff=intensity_cutoff,
neutrino_mass_MeV=args.neutrino_mass, neutrino_mass_MeV=neutrino_mass_MeV,
csv_separator=args.csv_separator, csv_separator=csv_separator,
normalize_to_one=args.normalize_to_one, separate_branches=separate_branches,
)
else:
config = create_config_from_source(
"paceENSDF",
nuclide=source_value,
e_step_MeV=args.e_step,
intensity_cutoff=args.intensity_cutoff,
neutrino_mass_MeV=args.neutrino_mass,
csv_separator=args.csv_separator,
normalize_to_one=args.normalize_to_one,
) )
else: else:
logger.info("Source: JSON file=%s", source_value) logger.info("Source: JSON file=%s", source_value)
data = load_json_input(source_value) data = load_json_input(source_value)
config = json_to_config(data) config = json_to_config(data)
# CLI flags override the JSON file values only when explicitly
# provided; unset flags keep the file's own values.
if args.e_step is not None:
config.e_step_MeV = e_step_MeV
if args.neutrino_mass is not None:
config.neutrino_mass_MeV = neutrino_mass_MeV
if args.intensity_cutoff is not None:
config.intensity_cutoff = intensity_cutoff
if args.csv_separator is not None:
config.csv_separator = csv_separator
config.separate_branches = separate_branches
# Dry run: display resolved config and exit # Dry run: display resolved config and exit
if args.dry_run: if args.dry_run:
@@ -373,7 +366,6 @@ def _print_dry_run_output(config, source_type: str) -> None:
f"Nuclide: {_decay_notation(config.Z_parent, config.Z_daughter, config.A_number)}" f"Nuclide: {_decay_notation(config.Z_parent, config.Z_daughter, config.A_number)}"
) )
print(f"Endpoint: {config.endpoint_MeV * 1000:.1f} keV") print(f"Endpoint: {config.endpoint_MeV * 1000:.1f} keV")
print(f"Transition: {config.transition_type}")
print(f"e_step: {config.e_step_MeV:.4f} MeV") print(f"e_step: {config.e_step_MeV:.4f} MeV")
print(f"neutrino_mass_MeV: {config.neutrino_mass_MeV:.6f}") print(f"neutrino_mass_MeV: {config.neutrino_mass_MeV:.6f}")
print(f"Corrections: {', '.join(enabled) if enabled else 'none'}") print(f"Corrections: {', '.join(enabled) if enabled else 'none'}")
@@ -393,8 +385,6 @@ def _print_dry_run_output(config, source_type: str) -> None:
else: else:
print("Mode: single-branch") print("Mode: single-branch")
# Detector response is a Phase 4 feature — not yet implemented
print("Detector: not configured (Phase 4)")
print("========================================") print("========================================")
@@ -516,10 +506,16 @@ def _decay_notation(Z_parent: int, Z_daughter: int, A: int) -> str:
return f"{_nuclide_symbol(Z_parent, A)} -> {_nuclide_symbol(Z_daughter, A)}" return f"{_nuclide_symbol(Z_parent, A)} -> {_nuclide_symbol(Z_daughter, A)}"
def main() -> None: def main(argv: list[str] | None = None) -> None:
"""Main entry point for the CLI.""" """Main entry point for the CLI.
Parameters
----------
argv : list[str], optional
Explicit argument vector (for tests/embedding); None reads sys.argv.
"""
parser = _build_parser() parser = _build_parser()
args = parser.parse_args() args = parser.parse_args(argv)
try: try:
_run(args) _run(args)
+41 -31
View File
@@ -96,6 +96,10 @@ class BranchInfo:
log(ft) value for this branch. log(ft) value for this branch.
transition_type : str transition_type : str
Forbiddenness classification for this branch (A, F1, F1U, F2, ...). Forbiddenness classification for this branch (A, F1, F1U, F2, ...).
endpoint_MeV : float
Measured beta endpoint energy of this transition in MeV (ENSDF
column 7). 0.0 means "not provided": consumers fall back to
base-endpoint minus level-energy subtraction.
""" """
level_index: int level_index: int
@@ -103,6 +107,7 @@ class BranchInfo:
intensity: float intensity: float
log_ft: Optional[float] log_ft: Optional[float]
transition_type: str = "A" transition_type: str = "A"
endpoint_MeV: float = 0.0
def _parse_nuclide_symbol(symbol: str) -> Tuple[str, int, int]: def _parse_nuclide_symbol(symbol: str) -> Tuple[str, int, int]:
@@ -384,24 +389,19 @@ def get_decay_info_from_paceENSDF(
# Extract Q-value from ground state branch (state[7] = endpoint energy in keV) # Extract Q-value from ground state branch (state[7] = endpoint energy in keV)
# and half-life from halflife_data # and half-life from halflife_data
endpoint_keV = 0.0
half_life_str = "unknown" half_life_str = "unknown"
# The spectrum endpoint is the maximum measured transition endpoint
# across all branches (ENSDF column 7, keV). Selecting the ground
# state branch or the first branch is wrong whenever those branches
# are absent or non-maximal (e.g. Co60, whose first listed branch
# endpoint is below higher-level transition endpoints).
endpoint_keV = 0.0
if beta_data: if beta_data:
for k, v in beta_data.items(): for _k, v in beta_data.items():
for state in v: for state in v:
if state[0] == 0: # ground state branch if len(state) > 7 and state[7] is not None:
endpoint_keV = float(state[7]) endpoint_keV = max(endpoint_keV, float(state[7]))
break
if endpoint_keV > 0:
break
# No ground state branch — use first available branch
if endpoint_keV == 0.0:
for k, v in beta_data.items():
if v and len(v) > 0:
endpoint_keV = float(v[0][7])
break
if halflife_data: if halflife_data:
for k, v in halflife_data.items(): for k, v in halflife_data.items():
@@ -447,6 +447,11 @@ def get_decay_info_from_paceENSDF(
intensity=float(state[9]) / 100.0, intensity=float(state[9]) / 100.0,
log_ft=float(state[11]) if state[11] is not None else None, log_ft=float(state[11]) if state[11] is not None else None,
transition_type=branch_transition_type, transition_type=branch_transition_type,
endpoint_MeV=(
float(state[7]) / 1000.0
if len(state) > 7 and state[7] is not None
else 0.0
),
) )
branches.append(branch) branches.append(branch)
@@ -472,7 +477,7 @@ def decay_info_to_config(
intensity_cutoff: float = 0.0, intensity_cutoff: float = 0.0,
neutrino_mass_MeV: float = 0.0, neutrino_mass_MeV: float = 0.0,
csv_separator: str = ",", csv_separator: str = ",",
normalize_to_one: bool = False, separate_branches: bool = False,
) -> SpectrumConfig: ) -> SpectrumConfig:
""" """
Convert DecayInfo to a SpectrumConfig. Convert DecayInfo to a SpectrumConfig.
@@ -489,20 +494,28 @@ def decay_info_to_config(
Neutrino mass in MeV. Neutrino mass in MeV.
csv_separator : str csv_separator : str
CSV column separator. CSV column separator.
normalize_to_one : bool separate_branches : bool
If True, normalize each branch to 1, ignoring intensity. If True, emit separate branch spectra, each normalized to
integrate to 1 (intensity ignored; no total spectrum column).
Returns Returns
------- -------
SpectrumConfig SpectrumConfig
Configuration ready for BetaSpectrum.from_config(). Configuration ready for BetaSpectrum.from_config().
""" """
# Convert BranchInfo -> BranchConfig for SpectrumConfig # Convert BranchInfo -> BranchConfig for SpectrumConfig. Per-branch
# endpoints are the measured transition endpoint energies when
# available; the base-endpoint minus level-energy subtraction is only
# a legacy fallback for hand-built DecayInfo objects without them.
branches: Optional[List[BranchConfig]] = None branches: Optional[List[BranchConfig]] = None
if info.branches: if info.branches:
branches = [ branches = [
BranchConfig( BranchConfig(
endpoint_MeV=info.endpoint_MeV - b.level_energy_keV / 1000.0, endpoint_MeV=(
b.endpoint_MeV
if b.endpoint_MeV > 0.0
else info.endpoint_MeV - b.level_energy_keV / 1000.0
),
transition_type=b.transition_type, transition_type=b.transition_type,
intensity=b.intensity, intensity=b.intensity,
) )
@@ -523,7 +536,7 @@ def decay_info_to_config(
branches=branches, branches=branches,
intensity_cutoff=intensity_cutoff, intensity_cutoff=intensity_cutoff,
csv_separator=csv_separator, csv_separator=csv_separator,
normalize_to_one=normalize_to_one, separate_branches=separate_branches,
) )
@@ -542,14 +555,6 @@ DEFAULT_JSON_SCHEMA = {
"transition_type": "F2", "transition_type": "F2",
"e_step_MeV": 0.001, "e_step_MeV": 0.001,
"neutrino_mass_MeV": 0.0, "neutrino_mass_MeV": 0.0,
"detector_model": "gaussian",
"detector_sigma_a_keV": 1.0,
"detector_sigma_b": 0.0,
"detector_tail_fraction": 0.0,
"detector_tau_keV": 5.0,
"detector_fano_factor": 0.12,
"detector_n_channels": 4096,
"detector_channel_energy_range": [0.0, 0.35],
} }
@@ -615,8 +620,13 @@ def load_json_input(filepath: str) -> Dict[str, Any]:
if not path.exists(): if not path.exists():
raise FileNotFoundError(f"Input file not found: {filepath}") raise FileNotFoundError(f"Input file not found: {filepath}")
with open(path, "r") as f: try:
with open(path, "r", encoding="utf-8") as f:
data = json.load(f) data = json.load(f)
except OSError as exc:
raise FileNotFoundError(f"Input file not readable: {filepath}") from exc
except json.JSONDecodeError as exc:
raise ValueError(f"Invalid JSON in input file {filepath}: {exc}") from exc
# Fill in defaults for missing optional fields # Fill in defaults for missing optional fields
for key, default in DEFAULT_JSON_SCHEMA.items(): for key, default in DEFAULT_JSON_SCHEMA.items():
@@ -699,7 +709,7 @@ def create_config_from_source(
intensity_cutoff = kwargs.get("intensity_cutoff", 0.0) intensity_cutoff = kwargs.get("intensity_cutoff", 0.0)
neutrino_mass = kwargs.get("neutrino_mass_MeV", 0.0) neutrino_mass = kwargs.get("neutrino_mass_MeV", 0.0)
csv_separator = kwargs.get("csv_separator", ",") csv_separator = kwargs.get("csv_separator", ",")
normalize_to_one = kwargs.get("normalize_to_one", False) separate_branches = kwargs.get("separate_branches", False)
info = get_decay_info_from_paceENSDF(nuclide, decay_type, decay_index) info = get_decay_info_from_paceENSDF(nuclide, decay_type, decay_index)
return decay_info_to_config( return decay_info_to_config(
info, info,
@@ -707,7 +717,7 @@ def create_config_from_source(
intensity_cutoff=intensity_cutoff, intensity_cutoff=intensity_cutoff,
neutrino_mass_MeV=neutrino_mass, neutrino_mass_MeV=neutrino_mass,
csv_separator=csv_separator, csv_separator=csv_separator,
normalize_to_one=normalize_to_one, separate_branches=separate_branches,
) )
elif source == "json": elif source == "json":
if not json_path: if not json_path:
+17 -11
View File
@@ -1,7 +1,7 @@
""" """
CSV export module — CSV export functionality for BetaSpectrumAnalyzer. CSV export module — CSV export functionality for BetaSpectrumAnalyzer.
Column semantics (v0.4.0 onward): Column semantics (v0.4.1 onward):
- The basic spectrum is ALWAYS core = phase_space x fermi; the Fermi function - The basic spectrum is ALWAYS core = phase_space x fermi; the Fermi function
is a core spectrum component, never an optional correction. Higher-order is a core spectrum component, never an optional correction. Higher-order
corrections (finite size, charge distribution, screening, exchange, corrections (finite size, charge distribution, screening, exchange,
@@ -12,14 +12,14 @@ Column semantics (v0.4.0 onward):
(the total column itself is normalized to integrate to 1). In the legacy (the total column itself is normalized to integrate to 1). In the legacy
export branch columns contained only the per-branch phase-space x export branch columns contained only the per-branch phase-space x
radiative product WITHOUT the Fermi function — physically wrong. radiative product WITHOUT the Fermi function — physically wrong.
- ``normalize_to_one`` mode: each branch column is normalized to - ``separate_branches`` mode: each branch column is normalized to
integrate to 1 individually, intensity is ignored, no total column. integrate to 1 individually, intensity is ignored, no total column.
""" """
from __future__ import annotations from __future__ import annotations
from datetime import datetime, timezone from datetime import datetime, timezone
from typing import TYPE_CHECKING, Dict, Optional from typing import TYPE_CHECKING, Dict
import numpy as np import numpy as np
import pandas as pd import pandas as pd
@@ -62,7 +62,7 @@ class CSVEmitter:
""" """
components = analyzer.components components = analyzer.components
config = analyzer.config config = analyzer.config
to_one = config.normalize_to_one separate = config.separate_branches
n_branches = analyzer._n_branches n_branches = analyzer._n_branches
# Physical branch spectra: core(phase_space x fermi) x corrections. # Physical branch spectra: core(phase_space x fermi) x corrections.
@@ -127,9 +127,9 @@ class CSVEmitter:
header_lines.append(f"# e_step: {config.e_step_MeV:.4f} MeV") header_lines.append(f"# e_step: {config.e_step_MeV:.4f} MeV")
header_lines.append(f"# n_points: {len(analyzer.energies_MeV)}") header_lines.append(f"# n_points: {len(analyzer.energies_MeV)}")
if to_one: if separate:
header_lines.append( header_lines.append(
"# branch normalization: to_one " "# branch normalization: separate "
"(each branch integrates to 1, intensity ignored)" "(each branch integrates to 1, intensity ignored)"
) )
else: else:
@@ -179,8 +179,8 @@ class CSVEmitter:
"energy_MeV": analyzer.energies_MeV, "energy_MeV": analyzer.energies_MeV,
} }
integral: Optional[float] = None # Positive-integral guard applies to BOTH normalization modes: a
if not to_one: # degenerate spectrum must never be written (not even as zeros).
total_raw = analyzer.spectrum(analyzer.W) total_raw = analyzer.spectrum(analyzer.W)
integral = float(np.trapezoid(total_raw, analyzer.energies_MeV)) integral = float(np.trapezoid(total_raw, analyzer.energies_MeV))
if integral <= 0.0: if integral <= 0.0:
@@ -189,6 +189,7 @@ class CSVEmitter:
"is degenerate (check neutrino_mass vs branch endpoints " "is degenerate (check neutrino_mass vs branch endpoints "
"and intensity_cutoff)" "and intensity_cutoff)"
) )
if not separate:
data["total_spectrum"] = total_raw / integral data["total_spectrum"] = total_raw / integral
for i in range(n_branches): for i in range(n_branches):
@@ -198,19 +199,23 @@ class CSVEmitter:
else: else:
col_name = f"branch_{key_i}_spectrum" col_name = f"branch_{key_i}_spectrum"
if to_one: if separate:
area = float(np.trapezoid(physical[i], analyzer.energies_MeV)) area = float(np.trapezoid(physical[i], analyzer.energies_MeV))
if area > 0.0: if area > 0.0:
data[col_name] = physical[i] / area data[col_name] = physical[i] / area
else: else:
data[col_name] = np.zeros_like(physical[i]) data[col_name] = np.zeros_like(physical[i])
else: else:
assert integral is not None
intensity_i = analyzer._unified_branches[i].intensity intensity_i = analyzer._unified_branches[i].intensity
data[col_name] = physical[i] * intensity_i / integral data[col_name] = physical[i] * intensity_i / integral
if csv_verbosity == "full": if csv_verbosity == "full":
# Per-branch component columns (factorized ingredients). # 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 i in range(n_branches):
for comp_name in ["PhaseSpace", "Radiative"]: for comp_name in ["PhaseSpace", "Radiative"]:
key = analyzer._get_component_key(i, comp_name) key = analyzer._get_component_key(i, comp_name)
@@ -218,7 +223,8 @@ class CSVEmitter:
data[f"branch_{i + 1}_{comp_name}"] = components[key] data[f"branch_{i + 1}_{comp_name}"] = components[key]
# Universal components (no prefix): Fermi plus universal # Universal components (no prefix): Fermi plus universal
# higher-order corrections. # higher-order corrections. In single mode the core factors
# (PhaseSpace, Radiative) also appear here unprefixed.
for name, values in components.items(): for name, values in components.items():
if not name.startswith("branch_"): if not name.startswith("branch_"):
data[name] = values data[name] = values
+8 -4
View File
@@ -235,7 +235,6 @@ def _plot_spectra_view_from_df(
) -> Figure: ) -> Figure:
"""Plot total spectrum and per-branch spectra from CSV columns.""" """Plot total spectrum and per-branch spectra from CSV columns."""
energies = df["energy_MeV"].to_numpy() energies = df["energy_MeV"].to_numpy()
e_step_keV = (meta["e_step_MeV"] or 0.0) * 1000.0
fig, ax = plt.subplots(figsize=(12, 7)) fig, ax = plt.subplots(figsize=(12, 7))
@@ -266,7 +265,7 @@ def _plot_spectra_view_from_df(
) )
ax.set_xlabel(r"Electron kinetic energy $E$ [MeV]", fontsize=12) ax.set_xlabel(r"Electron kinetic energy $E$ [MeV]", fontsize=12)
ax.set_ylabel(f"Normalized Counts / {e_step_keV:.1f} keV", fontsize=12) ax.set_ylabel(r"Normalized spectrum [MeV$^{-1}$]", fontsize=12)
fig.suptitle(_spectrum_title(meta), fontsize=14, fontweight="bold") fig.suptitle(_spectrum_title(meta), fontsize=14, fontweight="bold")
ax.set_yscale("log") ax.set_yscale("log")
ax.grid(True, alpha=0.3) ax.grid(True, alpha=0.3)
@@ -288,7 +287,6 @@ def _plot_debug_view_from_df(
) -> Figure: ) -> Figure:
"""Debug view (4 panels) built entirely from CSV columns.""" """Debug view (4 panels) built entirely from CSV columns."""
energies = df["energy_MeV"].to_numpy() energies = df["energy_MeV"].to_numpy()
e_step_keV = (meta["e_step_MeV"] or 0.0) * 1000.0
active = _active_branches(meta) active = _active_branches(meta)
fig, axes = plt.subplots(4, 1, figsize=(12, 16), sharex=True) fig, axes = plt.subplots(4, 1, figsize=(12, 16), sharex=True)
@@ -318,7 +316,7 @@ def _plot_debug_view_from_df(
label=f"Br. {b['index']} ({b['intensity'] * 100:.1f}%)", label=f"Br. {b['index']} ({b['intensity'] * 100:.1f}%)",
) )
ax.set_yscale("log") ax.set_yscale("log")
ax.set_ylabel(f"Normalized Counts / {e_step_keV:.1f} keV", fontsize=10) ax.set_ylabel(r"Normalized spectrum [MeV$^{-1}$]", fontsize=10)
ax.set_title("Spectrum", fontsize=11) ax.set_title("Spectrum", fontsize=11)
ax.grid(True, alpha=0.3) ax.grid(True, alpha=0.3)
ax.legend(fontsize=8, loc="best", ncol=2) ax.legend(fontsize=8, loc="best", ncol=2)
@@ -331,6 +329,9 @@ def _plot_debug_view_from_df(
ax.plot(energies, fermi_scaled, "r-", lw=1.5, label="Fermi", alpha=0.8) ax.plot(energies, fermi_scaled, "r-", lw=1.5, label="Fermi", alpha=0.8)
for color_i, b in enumerate(active): for color_i, b in enumerate(active):
col = f"branch_{b['index']}_PhaseSpace" col = f"branch_{b['index']}_PhaseSpace"
if col not in df.columns:
# Single-branch CSVs carry the core factors unprefixed.
col = "PhaseSpace"
if col not in df.columns: if col not in df.columns:
continue continue
ps_vals = df[col].to_numpy() ps_vals = df[col].to_numpy()
@@ -374,6 +375,9 @@ def _plot_debug_view_from_df(
ax = axes[3] ax = axes[3]
for color_i, b in enumerate(active): for color_i, b in enumerate(active):
col = f"branch_{b['index']}_Radiative" col = f"branch_{b['index']}_Radiative"
if col not in df.columns:
# Single-branch CSVs carry the core factors unprefixed.
col = "Radiative"
if col not in df.columns: if col not in df.columns:
continue continue
ax.plot( ax.plot(
+3 -5
View File
@@ -57,9 +57,6 @@ class SpectrumConfig:
""" """
Configuration for beta spectrum calculation. Configuration for beta spectrum calculation.
Supports declarative detector response specification:
to enable automatic detector smearing via convolve_with_detector().
Supports multi-branch decay via the `branches` field. When branches are Supports multi-branch decay via the `branches` field. When branches are
provided, the total spectrum is the intensity-weighted sum of individual provided, the total spectrum is the intensity-weighted sum of individual
branch spectra. Each branch has its own endpoint energy and transition type. branch spectra. Each branch has its own endpoint energy and transition type.
@@ -90,9 +87,10 @@ class SpectrumConfig:
# CSV output options # CSV output options
csv_separator: str = "," # Default separator for CSV output csv_separator: str = "," # Default separator for CSV output
normalize_to_one: bool = ( separate_branches: bool = (
False # If True, normalize each branch to 1, ignoring intensity False # If True: emit separate branch spectra, each normalized to
) )
# integrate to 1 (intensity ignored); no total spectrum column
def _effective_single_fields(config: SpectrumConfig) -> Tuple[float, str]: def _effective_single_fields(config: SpectrumConfig) -> Tuple[float, str]:
+1 -10
View File
@@ -5,14 +5,5 @@
"A_number": 99, "A_number": 99,
"endpoint_MeV": 0.294, "endpoint_MeV": 0.294,
"transition_type": "A", "transition_type": "A",
"e_step_MeV": 0.001, "e_step_MeV": 0.001
"use_detector_response": false,
"detector_model": "gaussian",
"detector_sigma_a_keV": 1.0,
"detector_sigma_b": 0.0,
"detector_tail_fraction": 0.0,
"detector_tau_keV": 5.0,
"detector_fano_factor": 0.12,
"detector_n_channels": 4096,
"detector_channel_energy_range": [0.0, 0.35]
} }
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project] [project]
name = "beta-spectrum" name = "beta-spectrum"
version = "0.4.0" version = "0.4.1"
description = "High-precision beta spectrum calculation for nuclear physics" description = "High-precision beta spectrum calculation for nuclear physics"
authors = [ authors = [
{ name = "Evgeniy Unzhakov" } { name = "Evgeniy Unzhakov" }
+208
View File
@@ -0,0 +1,208 @@
"""CLI flow tests: command-line options must reach the config in every
data-source flow (JSON file and paceENSDF), and degenerate exports must
exit non-zero without writing a file (regression D2/D4).
The CLI is invoked in-process via beta_spectrum.cli.main(argv) so the
paceENSDF dataset cache (loaded once per process) is shared with the
rest of the test session.
"""
from __future__ import annotations
import json
from typing import Any, Dict, List, Optional
import numpy as np
import pandas as pd
import pytest
from beta_spectrum.cli import main
pytestmark = pytest.mark.integration
def _base_json() -> Dict[str, Any]:
return {
"parent_symbol": "Tc99",
"Z_parent": 43,
"Z_daughter": 44,
"A_number": 99,
"endpoint_MeV": 0.294,
"transition_type": "A",
}
def write_json(tmp_path: Any, extra: Optional[Dict[str, Any]] = None) -> str:
data = _base_json()
if extra:
data.update(extra)
path = tmp_path / "input.json"
path.write_text(json.dumps(data))
return str(path)
def header_lines_of(path: str) -> List[str]:
lines: List[str] = []
with open(path, "r") as f:
for line in f:
if line.startswith("#"):
lines.append(line.rstrip("\n"))
else:
break
return lines
def read_export(path: str) -> pd.DataFrame:
return pd.read_csv(path, sep=None, engine="python", comment="#")
def n_points_of(header: List[str]) -> int:
for h in header:
if h.startswith("# n_points:"):
return int(h.split(":")[1])
raise AssertionError(f"n_points header missing: {header}")
class TestCLIJsonFlows:
"""Options must propagate through the JSON config flow (D2)."""
def test_json_honors_separate_branches_and_tab_separator(
self, tmp_path: Any
) -> None:
src = write_json(tmp_path)
out = str(tmp_path / "to_one.csv")
main([src, "--output", out, "--separate-branches", "--csv-separator", "tab"])
with open(out) as f:
body = "".join(ln for ln in f if not ln.startswith("#"))
assert "\t" in body, "CSV must be tab-separated"
df = read_export(out)
assert "total_spectrum" not in df.columns
area = np.trapezoid(df["branch_1_spectrum"], df["energy_MeV"])
assert np.isclose(area, 1.0, rtol=1e-3)
def test_json_defaults_keep_total_column(self, tmp_path: Any) -> None:
src = write_json(tmp_path)
out = str(tmp_path / "default.csv")
main([src, "--output", out])
df = read_export(out)
assert "total_spectrum" in df.columns
area = np.trapezoid(df["total_spectrum"], df["energy_MeV"])
assert np.isclose(area, 1.0, rtol=1e-3)
def test_json_cli_overrides_file_physics_values(self, tmp_path: Any) -> None:
src = write_json(tmp_path, {"e_step_MeV": 0.001, "neutrino_mass_MeV": 0.0})
out = str(tmp_path / "overrides.csv")
main([src, "--output", out, "--e-step", "0.02", "--neutrino-mass", "0.15"])
header = header_lines_of(out)
assert any("neutrino_mass_MeV: 0.150000" in h for h in header)
expected_points = int(np.arange(0.02, 0.294, 0.02).size)
assert n_points_of(header) == expected_points
def test_json_file_values_used_when_flags_absent(self, tmp_path: Any) -> None:
src = write_json(tmp_path, {"e_step_MeV": 0.005})
out = str(tmp_path / "file_values.csv")
main([src, "--output", out])
header = header_lines_of(out)
expected_points = int(np.arange(0.005, 0.294, 0.005).size)
assert n_points_of(header) == expected_points
class TestCLIDegenerateFlows:
"""Degenerate exports (m_nu above every endpoint) must fail cleanly."""
def test_degenerate_intensity_mode_exits_nonzero_no_file(
self, tmp_path: Any
) -> None:
src = write_json(tmp_path)
out = str(tmp_path / "deg.csv")
with pytest.raises(SystemExit) as exc:
main([src, "--output", out, "--neutrino-mass", "0.35"])
assert exc.value.code == 1
assert not (tmp_path / "deg.csv").exists()
def test_degenerate_to_one_mode_exits_nonzero_no_file(self, tmp_path: Any) -> None:
src = write_json(tmp_path)
out = str(tmp_path / "deg_to_one.csv")
with pytest.raises(SystemExit) as exc:
main(
[
src,
"--output",
out,
"--neutrino-mass",
"0.35",
"--separate-branches",
]
)
assert exc.value.code == 1
assert not (tmp_path / "deg_to_one.csv").exists()
@pytest.mark.requires_ensdf
class TestCLIPaceENSDF:
"""paceENSDF flow honors the same export options."""
def test_paceensdf_honors_separate_branches(self, tmp_path: Any) -> None:
out = str(tmp_path / "p32.csv")
main(["P32", "--output", out, "--separate-branches"])
df = read_export(out)
assert "total_spectrum" not in df.columns
area = np.trapezoid(df["branch_1_spectrum"], df["energy_MeV"])
assert np.isclose(area, 1.0, rtol=1e-3)
def test_paceensdf_degenerate_exits_nonzero_no_file(self, tmp_path: Any) -> None:
out = str(tmp_path / "tc99deg.csv")
with pytest.raises(SystemExit) as exc:
main(["Tc99", "--output", out, "--neutrino-mass", "0.35"])
assert exc.value.code == 1
assert not (tmp_path / "tc99deg.csv").exists()
def test_paceensdf_co60_dry_run_multi_branch(self, capsys: Any) -> None:
"""Co60 parses into a healthy multi-branch config (D3): no crash,
all branch endpoints positive."""
main(["Co60", "--dry-run"])
out = capsys.readouterr().out
assert "Multi-branch mode: 3 branches" in out
def test_paceensdf_co60_export(self, tmp_path: Any) -> None:
out = str(tmp_path / "co60.csv")
main(["Co60", "--output", out])
df = read_export(out)
area = np.trapezoid(df["total_spectrum"], df["energy_MeV"])
assert np.isclose(area, 1.0, rtol=1e-3)
eps = header_lines_of(out)
assert any("E₀=1492.0" in h for h in eps)
assert not any("excluded" in h for h in eps)
def test_cli_builds_paceensdf_config_via_library(
self, monkeypatch: Any, capsys: Any
) -> None:
"""CLI must delegate paceENSDF config building to the library
(create_config_from_source), not duplicate it inline (D5)."""
import beta_spectrum.nuclear_data as nuclear_data
real_builder = nuclear_data.create_config_from_source
calls: List[Dict[str, Any]] = []
def spy(source: str, **kwargs: Any) -> Any:
calls.append({"source": source, **kwargs})
return real_builder(source, **kwargs)
monkeypatch.setattr(nuclear_data, "create_config_from_source", spy)
main(["Cs137", "--dry-run", "--intensity-cutoff", "0.9"])
capsys.readouterr()
assert len(calls) == 1, (
"CLI must delegate paceENSDF config building to the library "
"builder exactly once"
)
assert calls[0]["source"] == "paceENSDF"
assert calls[0]["nuclide"] == "Cs137"
assert calls[0]["intensity_cutoff"] == 0.9
+184 -4
View File
@@ -66,7 +66,7 @@ def make_single_config(**over: Any) -> SpectrumConfig:
) )
def make_degenerate_config() -> SpectrumConfig: def make_degenerate_config(**over: Any) -> SpectrumConfig:
"""Multi-branch config where every endpoint lies below m_nu.""" """Multi-branch config where every endpoint lies below m_nu."""
return SpectrumConfig( return SpectrumConfig(
Z_parent=43, Z_parent=43,
@@ -75,6 +75,7 @@ def make_degenerate_config() -> SpectrumConfig:
endpoint_MeV=0.294, endpoint_MeV=0.294,
e_step_MeV=0.001, e_step_MeV=0.001,
neutrino_mass_MeV=0.35, neutrino_mass_MeV=0.35,
**over,
branches=[ branches=[
BranchConfig(endpoint_MeV=0.294, transition_type="F2", intensity=0.98), BranchConfig(endpoint_MeV=0.294, transition_type="F2", intensity=0.98),
BranchConfig(endpoint_MeV=0.200, transition_type="F2U", intensity=0.02), BranchConfig(endpoint_MeV=0.200, transition_type="F2U", intensity=0.02),
@@ -312,8 +313,8 @@ class TestCSVExportCoreSemantics:
assert "phase_space" in core_line[0] assert "phase_space" in core_line[0]
assert "fermi" in core_line[0] assert "fermi" in core_line[0]
def test_normalize_to_one_per_branch_unit_area(self, tmp_path: Any) -> None: def test_separate_branches_per_branch_unit_area(self, tmp_path: Any) -> None:
config = make_multi_config(normalize_to_one=True) config = make_multi_config(separate_branches=True)
spectrum = BetaSpectrum.from_config(config) spectrum = BetaSpectrum.from_config(config)
analyzer = BetaSpectrumAnalyzer(spectrum, config) analyzer = BetaSpectrumAnalyzer(spectrum, config)
_, df = export_and_read(analyzer, tmp_path) _, df = export_and_read(analyzer, tmp_path)
@@ -329,8 +330,22 @@ class TestCSVExportCoreSemantics:
spectrum = BetaSpectrum.from_config(config) spectrum = BetaSpectrum.from_config(config)
analyzer = BetaSpectrumAnalyzer(spectrum, config) analyzer = BetaSpectrumAnalyzer(spectrum, config)
path = str(tmp_path / "degenerate.csv") path = str(tmp_path / "degenerate.csv")
with pytest.raises(ValueError): with pytest.raises(ValueError, match="non-positive"):
analyzer.export_to_csv(path) analyzer.export_to_csv(path)
assert not (tmp_path / "degenerate.csv").exists()
def test_degenerate_export_raises_in_separate_branches_mode(
self, tmp_path: Any
) -> None:
"""The positive-integral guard must apply in separate-branches mode too:
no all-zero CSV may be written for a degenerate spectrum."""
config = make_degenerate_config(separate_branches=True)
spectrum = BetaSpectrum.from_config(config)
analyzer = BetaSpectrumAnalyzer(spectrum, config)
path = str(tmp_path / "degenerate_to_one.csv")
with pytest.raises(ValueError, match="non-positive"):
analyzer.export_to_csv(path)
assert not (tmp_path / "degenerate_to_one.csv").exists()
def test_concise_values_match_full(self, tmp_path: Any) -> None: def test_concise_values_match_full(self, tmp_path: Any) -> None:
config = make_multi_config() config = make_multi_config()
@@ -549,3 +564,168 @@ class TestAnalyzerUniversalFactor:
assert np.all(uni > 1.0) assert np.all(uni > 1.0)
fermi_only = analyzer.components["Fermi"] fermi_only = analyzer.components["Fermi"]
assert np.allclose(uni, fermi_only, rtol=1e-12) assert np.allclose(uni, fermi_only, rtol=1e-12)
class TestNormalizationModeMatrix:
"""separate_branches and intensity modes must behave consistently across CSV
verbosity, branch count and the CSV-driven visualizer."""
def test_separate_branches_concise_multi_unit_area_no_total_no_components(
self, tmp_path: Any
) -> None:
config = make_multi_config(separate_branches=True)
spectrum = BetaSpectrum.from_config(config)
analyzer = BetaSpectrumAnalyzer(spectrum, config)
path, df = export_and_read(
analyzer, tmp_path, "to_one_concise.csv", csv_verbosity="concise"
)
assert set(df.columns) == {
"energy_MeV",
"branch_1",
"branch_2",
"branch_3",
}
header = header_lines_of(path)
assert any("branch normalization: separate" in h for h in header)
energies = df["energy_MeV"].to_numpy()
for col in ("branch_1", "branch_2", "branch_3"):
area = np.trapezoid(df[col].to_numpy(), energies)
assert np.isclose(area, 1.0, rtol=5e-4)
def test_separate_branches_single_mode_unit_area(self, tmp_path: Any) -> None:
config = make_single_config(separate_branches=True)
spectrum = BetaSpectrum.from_config(config)
analyzer = BetaSpectrumAnalyzer(spectrum, config)
_path, df = export_and_read(analyzer, tmp_path, "to_one_single.csv")
assert "total_spectrum" not in df.columns
area = np.trapezoid(
df["branch_1_spectrum"].to_numpy(), df["energy_MeV"].to_numpy()
)
assert np.isclose(area, 1.0, rtol=5e-4)
def test_intensity_concise_branch_columns_sum_to_total(self, tmp_path: Any) -> None:
config = make_multi_config()
spectrum = BetaSpectrum.from_config(config)
analyzer = BetaSpectrumAnalyzer(spectrum, config)
_path, df = export_and_read(
analyzer, tmp_path, "intensity_concise.csv", csv_verbosity="concise"
)
summed = (
df["branch_1"].to_numpy()
+ df["branch_2"].to_numpy()
+ df["branch_3"].to_numpy()
)
total = df["total_spectrum"].to_numpy()
nonzero = total > 1e-12
assert np.all(np.abs(summed[nonzero] - total[nonzero]) <= total[nonzero] * 5e-4)
def test_plot_from_csv_separate_branches_plots_branches_without_total(
self, tmp_path: Any
) -> None:
config = make_multi_config(separate_branches=True)
spectrum = BetaSpectrum.from_config(config)
analyzer = BetaSpectrumAnalyzer(spectrum, config)
path, df = export_and_read(analyzer, tmp_path, "to_one_plot.csv")
fig = Visualizer.plot_from_csv(path, save_path=None, show_components=True)
assert len(fig.axes) == 4
# Panel 1 has exactly the 3 branch curves — no total spectrum
# exists in separate-branches mode.
panel = fig.axes[0]
curves = [np.asarray(ln.get_xydata()) for ln in panel.lines]
assert len(curves) == 3
energies = df["energy_MeV"].to_numpy()
assert np.allclose(curves[0][:, 0], energies)
assert np.allclose(curves[0][:, 1], df["branch_1_spectrum"].to_numpy())
def test_plot_analysis_separate_branches_returns_figure(
self, tmp_path: Any
) -> None:
config = make_multi_config(separate_branches=True)
spectrum = BetaSpectrum.from_config(config)
analyzer = BetaSpectrumAnalyzer(spectrum, config)
save = str(tmp_path / "analysis.png")
fig = Visualizer.plot_analysis(analyzer, save_path=save)
assert len(fig.axes) == 4
assert (tmp_path / "analysis.png").exists()
def test_plot_from_csv_separate_branches_concise_warns_and_falls_back(
self, tmp_path: Any, capsys: Any
) -> None:
config = make_multi_config(separate_branches=True)
spectrum = BetaSpectrum.from_config(config)
analyzer = BetaSpectrumAnalyzer(spectrum, config)
path, df = export_and_read(
analyzer, tmp_path, "to_one_concise2.csv", csv_verbosity="concise"
)
fig = Visualizer.plot_from_csv(path, save_path=None, show_components=True)
captured = capsys.readouterr()
assert "Warning" in captured.out
# Simple spectra view: one axis, 3 branch curves, no total curve.
assert len(fig.axes) == 1
assert len(fig.axes[0].lines) == 3
class TestSingleModeColumnCleanup:
"""Single-mode CSVs carry the factorized components once, unprefixed;
multi-mode CSVs carry them per branch; the visualizer reads whichever
naming the file actually uses."""
def test_single_mode_full_export_has_unprefixed_components_only(
self, tmp_path: Any
) -> None:
config = make_single_config(use_radiative=True)
spectrum = BetaSpectrum.from_config(config)
analyzer = BetaSpectrumAnalyzer(spectrum, config)
path, df = export_and_read(analyzer, tmp_path, "single_components.csv")
assert "PhaseSpace" in df.columns
assert "Radiative" in df.columns
assert "Fermi" in df.columns
assert "branch_1_spectrum" in df.columns
assert "branch_1_PhaseSpace" not in df.columns
assert "branch_1_Radiative" not in df.columns
def test_multi_mode_full_export_has_prefixed_core_factors(
self, tmp_path: Any
) -> None:
config = make_multi_config(use_radiative=True)
spectrum = BetaSpectrum.from_config(config)
analyzer = BetaSpectrumAnalyzer(spectrum, config)
path, df = export_and_read(analyzer, tmp_path, "multi_components.csv")
assert "branch_1_PhaseSpace" in df.columns
assert "branch_2_Radiative" in df.columns
assert "Fermi" in df.columns
# The bare core-factor names belong to single mode only.
assert "PhaseSpace" not in df.columns
assert "Radiative" not in df.columns
def test_debug_view_single_csv_includes_core_panels(self, tmp_path: Any) -> None:
config = make_single_config(use_radiative=True)
spectrum = BetaSpectrum.from_config(config)
analyzer = BetaSpectrumAnalyzer(spectrum, config)
path, _df = export_and_read(analyzer, tmp_path, "single_plot.csv")
fig = Visualizer.plot_from_csv(path, None, show_components=True)
assert len(fig.axes) == 4
# Panel 2: Fermi + the single branch's phase space.
assert len(fig.axes[1].lines) == 2
labels = [ln.get_label() for ln in fig.axes[1].lines]
assert "Fermi" in labels
assert any("PS Br" in str(lbl) for lbl in labels)
# Panel 4: the radiative correction for the single branch.
assert len(fig.axes[3].lines) == 1
def test_ylabel_uses_per_MeV_units(self, tmp_path: Any) -> None:
config = make_multi_config()
spectrum = BetaSpectrum.from_config(config)
analyzer = BetaSpectrumAnalyzer(spectrum, config)
path, _df = export_and_read(analyzer, tmp_path, "ylabel.csv")
fig = Visualizer.plot_from_csv(path, None, show_components=True)
assert "MeV$^{-1}$" in fig.axes[0].get_ylabel()
+7 -2
View File
@@ -118,10 +118,15 @@ class TestNeutrinoMassCLIIntegration:
assert args.neutrino_mass == 0.01, "CLI should parse neutrino_mass correctly" assert args.neutrino_mass == 0.01, "CLI should parse neutrino_mass correctly"
def test_cli_neutrino_mass_default_value(self): def test_cli_neutrino_mass_default_value(self):
"""Test that CLI neutrino mass parameter defaults to 0.0.""" """Unset flag parses as None ("not provided"); the effective
default (0.0 MeV) is applied when the config is built."""
from beta_spectrum.cli import _build_parser from beta_spectrum.cli import _build_parser
parser = _build_parser() parser = _build_parser()
args = parser.parse_args(["--nuclide", "Tc99"]) args = parser.parse_args(["--nuclide", "Tc99"])
assert args.neutrino_mass == 0.0, "CLI neutrino_mass default should be 0.0" assert args.neutrino_mass is None, (
"unset --neutrino-mass must parse as None so JSON file values "
"are not overwritten"
)
assert hasattr(args, "neutrino_mass")
+137
View File
@@ -0,0 +1,137 @@
"""CLI option surface tests: renamed branch-separation flag, short
aliases for every long option, and dry-run output shape (audit follow-up
by the maintainer: rename --normalize-to-one, add short options, drop the
redundant top-level Transition line)."""
from __future__ import annotations
import dataclasses
from typing import Any, Optional
import pytest
from beta_spectrum import BranchConfig, SpectrumConfig
from beta_spectrum.cli import _build_parser, _print_dry_run_output
pytestmark = pytest.mark.quality
def make_config() -> SpectrumConfig:
return SpectrumConfig(
Z_parent=59,
Z_daughter=60,
A_number=144,
endpoint_MeV=2.98,
transition_type="F1",
e_step_MeV=0.001,
branches=[
BranchConfig(endpoint_MeV=2.98, transition_type="F1", intensity=0.9),
BranchConfig(endpoint_MeV=2.1, transition_type="A", intensity=0.1),
],
)
class TestSeparateBranchesRename:
"""--normalize-to-one was renamed to --separate-branches (no alias)."""
def test_flag_parses_true(self) -> None:
parser = _build_parser()
args = parser.parse_args(["input.json", "--separate-branches"])
assert args.separate_branches is True
def test_flag_default_is_none(self) -> None:
parser = _build_parser()
args = parser.parse_args(["input.json"])
assert args.separate_branches is None
def test_old_flag_is_rejected(self) -> None:
parser = _build_parser()
with pytest.raises(SystemExit):
parser.parse_args(["input.json", "--normalize-to-one"])
def test_config_field_renamed(self) -> None:
names = {f.name for f in dataclasses.fields(SpectrumConfig)}
assert "separate_branches" in names
assert "normalize_to_one" not in names
def test_config_old_kwarg_rejected(self) -> None:
kwargs: dict[str, Any] = {
"Z_parent": 15,
"Z_daughter": 16,
"A_number": 32,
"endpoint_MeV": 1.7103,
"transition_type": "A",
"normalize_to_one": True,
}
with pytest.raises(TypeError):
SpectrumConfig(**kwargs)
# (short, long, value-or-None-for-store_true)
_SHORT_EQUIV = [
("-N", "--nuclide", "Cs137"),
("-i", "--input", "in.json"),
("-s", "--e-step", "0.01"),
("-m", "--neutrino-mass", "0.15"),
("-o", "--output", "out.csv"),
("-p", "--plot", "plot.png"),
("-L", "--log-file", "calc.log"),
("-d", "--dry-run", None),
("-c", "--intensity-cutoff", "0.5"),
("-V", "--csv-verbosity", "concise"),
("-S", "--csv-separator", "tab"),
("-b", "--separate-branches", None),
]
class TestShortOptions:
"""Every long option gains a single-letter alias (only -v/-q existed)."""
@pytest.mark.parametrize(
"short,long,value",
_SHORT_EQUIV,
ids=[s.lstrip("-") for s, _, _ in _SHORT_EQUIV],
)
def test_short_matches_long(
self, short: str, long: str, value: Optional[str]
) -> None:
parser = _build_parser()
argv_long = [long] if value is None else [long, value]
argv_short = [short] if value is None else [short, value]
args_long = parser.parse_args(["Tc99"] + argv_long)
args_short = parser.parse_args(["Tc99"] + argv_short)
dest = long.lstrip("-").replace("-", "_")
assert getattr(args_short, dest) == getattr(args_long, dest)
assert getattr(args_long, dest) is not None or value is None
def test_existing_shorts_unchanged(self) -> None:
parser = _build_parser()
args = parser.parse_args(["Tc99", "-v", "-q"])
assert args.verbose == 1
assert args.quiet is True
class TestDryRunOutputShape:
"""The top-level Transition line is redundant with the per-branch
list (for multi-branch runs) and was removed."""
def test_multi_branch_dry_run_no_top_level_transition(self, capsys: Any) -> None:
_print_dry_run_output(make_config(), "paceENSDF")
out = capsys.readouterr().out
assert "Transition:" not in out
assert "transition=F1" in out
assert "transition=A" in out
assert "Multi-branch mode: 2 branches" in out
def test_single_branch_dry_run_reports_endpoint(self, capsys: Any) -> None:
config = SpectrumConfig(
Z_parent=15,
Z_daughter=16,
A_number=32,
endpoint_MeV=1.7103,
transition_type="A",
)
_print_dry_run_output(config, "json")
out = capsys.readouterr().out
assert "Transition:" not in out
assert "Mode: single-branch" in out
+1 -1
View File
@@ -154,7 +154,7 @@ class TestCSVMetadataHeader:
lines = f.readlines() lines = f.readlines()
assert any("# beta-spectrum v" in line for line in lines) assert any("# beta-spectrum v" in line for line in lines)
assert any("0.4.0" in line for line in lines) assert any("0.4.1" in line for line in lines)
os.unlink(filename) os.unlink(filename)
def test_header_contains_timestamp(self): def test_header_contains_timestamp(self):
+6 -2
View File
@@ -76,10 +76,14 @@ class TestCLIArguments:
assert args.neutrino_mass == 0.01, "CLI neutrino_mass should be 0.01" assert args.neutrino_mass == 0.01, "CLI neutrino_mass should be 0.01"
def test_neutrino_mass_cli_default_value(self): def test_neutrino_mass_cli_default_value(self):
"""Test that --neutrino-mass defaults to 0.0 in CLI.""" """Unset flag parses as None ("not provided") so JSON file
values are not overwritten; the effective default (0.0 MeV) is
applied when the config is built."""
from beta_spectrum.cli import _build_parser from beta_spectrum.cli import _build_parser
parser = _build_parser() parser = _build_parser()
args = parser.parse_args(["--nuclide", "Tc99"]) args = parser.parse_args(["--nuclide", "Tc99"])
assert args.neutrino_mass == 0.0, "CLI neutrino_mass default should be 0.0" assert (
args.neutrino_mass is None
), "unset --neutrino-mass must parse as None (not a value)"
+44 -3
View File
@@ -161,9 +161,12 @@ class TestJsonInput:
path = self._write_json(data) path = self._write_json(data)
try: try:
loaded = load_json_input(str(path)) loaded = load_json_input(str(path))
# Defaults should be filled in # Defaults should be filled in. The detector_* schema keys were
assert loaded["detector_model"] == "gaussian" # removed together with the detector_response module; JSON
assert loaded["detector_sigma_a_keV"] == 1.0 # defaults carry spectrum physics only.
assert "detector_model" not in loaded
assert loaded["neutrino_mass_MeV"] == 0.0
assert loaded["transition_type"] == "F2"
finally: finally:
path.unlink() path.unlink()
@@ -294,6 +297,44 @@ class TestDecayInfoToConfig:
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@pytest.mark.slow
@pytest.mark.requires_ensdf
@pytest.mark.skipif(not _has_paceENSDF(), reason="paceENSDF not installed")
class TestPaceENSDFBranchEndpoints:
"""Per-branch endpoints are the measured transition endpoint energies
(ENSDF column 7), not base-endpoint minus level energy subtractions
with a base taken from an arbitrary first branch (D3)."""
def _branch_endpoints(self, nuclide: str) -> list[float]:
info = get_decay_info_from_paceENSDF(nuclide, "beta_minus")
cfg = decay_info_to_config(info)
assert cfg.branches is not None
return sorted(
(b.endpoint_MeV for b in cfg.branches),
reverse=True,
)
def test_co60_branch_endpoints_positive(self) -> None:
# Measured endpoints: 1492.0, 670.0, 317.88 keV (levels above the
# first branch's endpoint previously produced negative values and
# a hard ValueError from BranchConfig validation).
eps = self._branch_endpoints("Co60")
assert eps == pytest.approx([1.492, 0.670, 0.31788], abs=1e-6)
assert all(e > 0 for e in eps)
def test_tc99_branch_endpoints(self) -> None:
eps = self._branch_endpoints("Tc99")
assert eps == pytest.approx([0.2975, 0.2080], abs=1e-6)
def test_cs137_branch_endpoints(self) -> None:
eps = self._branch_endpoints("Cs137")
assert eps == pytest.approx([1.176, 0.89213, 0.51403], abs=1e-6)
def test_overall_endpoint_is_max_branch_endpoint(self) -> None:
info = get_decay_info_from_paceENSDF("Co60", "beta_minus")
assert info.endpoint_MeV == pytest.approx(1.492, abs=1e-6)
@pytest.mark.requires_ensdf @pytest.mark.requires_ensdf
@pytest.mark.skipif(not _has_paceENSDF(), reason="paceENSDF not installed") @pytest.mark.skipif(not _has_paceENSDF(), reason="paceENSDF not installed")
class TestPaceENSDFIntegration: class TestPaceENSDFIntegration: