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
│ │ ├── screening.py ✓ Atomic electron screening (ratio method)
│ │ ├── exchange.py ✓ Hayen-2018 empirical fit coefficients
│ │ ├── radiative.py ✓ Outer radiative corrections, soft-photon resummation
│ │ └── detector_response.py ✓ Analytical detector response models
│ │ └── radiative.py ✓ Outer radiative corrections, soft-photon resummation
├── beta_spectrum/output/ ← output, plotting, visualization, CSV export
│ ├── __init__.py # BetaSpectrumAnalyzer, CSVEmitter, Visualizer
│ ├── 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:
- 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`
- 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`)
- `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:
- **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
- **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
- **CLI**: `bs_pnpi --nuclide Tc99` — full paceENSDF integration (auto-lookup of Z, A, Q-value, transition type, half-life, branches, forbiddenness from ENSDF)
- **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
- **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; 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
│ ├── exchange.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
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] Version bumped to 0.4.0
- [ ] 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
duck-typed optional hook and `DEFAULT_JSON_SCHEMA` keeps the detector keys;
restore from git history (commits up to `5f99566`) if convolution is needed again
`beta_spectrum/components/` together with its tests; `fitter/model.py` keeps a
duck-typed optional hook only (`DEFAULT_JSON_SCHEMA` detector keys and the CLI
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
**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)
@@ -353,6 +354,12 @@ ______________________________________________________________________
- README updated with new architecture diagram and Quick Start examples
- 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
- **Track A: ⁹⁹Tc spectrum-shape analysis** — start real data processing pipeline
+1 -1
View File
@@ -52,4 +52,4 @@ __all__ = [
"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_group = parser.add_mutually_exclusive_group(required=False)
source_group.add_argument(
"-N",
"--nuclide",
type=str,
help="Nuclide symbol for paceENSDF lookup (e.g. 'Tc99', 'Co60'). "
"Retrieves decay parameters from the ENSDF database via paceENSDF.",
)
source_group.add_argument(
"-i",
"--input",
type=str,
metavar="FILE",
@@ -79,20 +81,27 @@ def _build_parser() -> argparse.ArgumentParser:
# Common parameters
parser.add_argument(
"-s",
"--e-step",
type=float,
default=0.001,
help="Energy step size in MeV (default: 0.001). All energies are in MeV throughout the toolkit.",
default=None,
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(
"-m",
"--neutrino-mass",
type=float,
default=0.0,
help="Neutrino mass in MeV (default: 0.0). Set to non-zero value to include neutrino mass effects in phase space calculation.",
default=None,
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
parser.add_argument(
"-o",
"--output",
type=str,
metavar="FILE",
@@ -101,6 +110,7 @@ def _build_parser() -> argparse.ArgumentParser:
# Plotting
parser.add_argument(
"-p",
"--plot",
type=str,
metavar="FILE",
@@ -124,6 +134,7 @@ def _build_parser() -> argparse.ArgumentParser:
help="Suppress all terminal output (only errors shown).",
)
log_group.add_argument(
"-L",
"--log-file",
type=str,
metavar="PATH",
@@ -131,20 +142,23 @@ def _build_parser() -> argparse.ArgumentParser:
"Log file is always opened in append mode at DEBUG level.",
)
log_group.add_argument(
"-d",
"--dry-run",
action="store_true",
help="Validate input and display resolved configuration without computing spectrum.",
)
log_group.add_argument(
"-c",
"--intensity-cutoff",
type=float,
default=0.0,
default=None,
metavar="FRAC",
help="Minimum branch intensity fraction to include in the spectrum "
"(default: 0.0 = include all branches). "
"Branches below this threshold are excluded from the calculation and plots.",
)
log_group.add_argument(
"-V",
"--csv-verbosity",
type=str,
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)",
)
log_group.add_argument(
"-S",
"--csv-separator",
type=str,
default=",",
default=None,
help="CSV column separator (default: ','). Use 'tab' for tab separation.",
)
log_group.add_argument(
"--normalize-to-one",
"-b",
"--separate-branches",
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
@@ -173,15 +192,10 @@ def _run(args: argparse.Namespace) -> None:
from beta_spectrum.logging_utils import get_logger
from beta_spectrum.nuclear_data import (
create_config_from_source,
get_decay_info_from_paceENSDF,
json_to_config,
load_json_input,
)
from beta_spectrum.spectrum import (
BetaSpectrum,
BranchConfig,
SpectrumConfig,
)
from beta_spectrum.spectrum import BetaSpectrum
from beta_spectrum.output import BetaSpectrumAnalyzer
# Determine logging level
@@ -230,66 +244,45 @@ def _run(args: argparse.Namespace) -> None:
else:
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
if source_type == "paceENSDF":
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(
"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:
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,
e_step_MeV=e_step_MeV,
intensity_cutoff=intensity_cutoff,
neutrino_mass_MeV=neutrino_mass_MeV,
csv_separator=csv_separator,
separate_branches=separate_branches,
)
else:
logger.info("Source: JSON file=%s", source_value)
data = load_json_input(source_value)
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
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)}"
)
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"neutrino_mass_MeV: {config.neutrino_mass_MeV:.6f}")
print(f"Corrections: {', '.join(enabled) if enabled else 'none'}")
@@ -393,8 +385,6 @@ def _print_dry_run_output(config, source_type: str) -> None:
else:
print("Mode: single-branch")
# Detector response is a Phase 4 feature — not yet implemented
print("Detector: not configured (Phase 4)")
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)}"
def main() -> None:
"""Main entry point for the CLI."""
def main(argv: list[str] | None = None) -> None:
"""Main entry point for the CLI.
Parameters
----------
argv : list[str], optional
Explicit argument vector (for tests/embedding); None reads sys.argv.
"""
parser = _build_parser()
args = parser.parse_args()
args = parser.parse_args(argv)
try:
_run(args)
+41 -31
View File
@@ -96,6 +96,10 @@ class BranchInfo:
log(ft) value for this branch.
transition_type : str
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
@@ -103,6 +107,7 @@ class BranchInfo:
intensity: float
log_ft: Optional[float]
transition_type: str = "A"
endpoint_MeV: float = 0.0
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)
# and half-life from halflife_data
endpoint_keV = 0.0
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:
for k, v in beta_data.items():
for _k, v in beta_data.items():
for state in v:
if state[0] == 0: # ground state branch
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 len(state) > 7 and state[7] is not None:
endpoint_keV = max(endpoint_keV, float(state[7]))
if halflife_data:
for k, v in halflife_data.items():
@@ -447,6 +447,11 @@ def get_decay_info_from_paceENSDF(
intensity=float(state[9]) / 100.0,
log_ft=float(state[11]) if state[11] is not None else None,
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)
@@ -472,7 +477,7 @@ def decay_info_to_config(
intensity_cutoff: float = 0.0,
neutrino_mass_MeV: float = 0.0,
csv_separator: str = ",",
normalize_to_one: bool = False,
separate_branches: bool = False,
) -> SpectrumConfig:
"""
Convert DecayInfo to a SpectrumConfig.
@@ -489,20 +494,28 @@ def decay_info_to_config(
Neutrino mass in MeV.
csv_separator : str
CSV column separator.
normalize_to_one : bool
If True, normalize each branch to 1, ignoring intensity.
separate_branches : bool
If True, emit separate branch spectra, each normalized to
integrate to 1 (intensity ignored; no total spectrum column).
Returns
-------
SpectrumConfig
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
if info.branches:
branches = [
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,
intensity=b.intensity,
)
@@ -523,7 +536,7 @@ def decay_info_to_config(
branches=branches,
intensity_cutoff=intensity_cutoff,
csv_separator=csv_separator,
normalize_to_one=normalize_to_one,
separate_branches=separate_branches,
)
@@ -542,14 +555,6 @@ DEFAULT_JSON_SCHEMA = {
"transition_type": "F2",
"e_step_MeV": 0.001,
"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():
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)
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
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)
neutrino_mass = kwargs.get("neutrino_mass_MeV", 0.0)
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)
return decay_info_to_config(
info,
@@ -707,7 +717,7 @@ def create_config_from_source(
intensity_cutoff=intensity_cutoff,
neutrino_mass_MeV=neutrino_mass,
csv_separator=csv_separator,
normalize_to_one=normalize_to_one,
separate_branches=separate_branches,
)
elif source == "json":
if not json_path:
+17 -11
View File
@@ -1,7 +1,7 @@
"""
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
is a core spectrum component, never an optional correction. Higher-order
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
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
- ``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, Optional
from typing import TYPE_CHECKING, Dict
import numpy as np
import pandas as pd
@@ -62,7 +62,7 @@ class CSVEmitter:
"""
components = analyzer.components
config = analyzer.config
to_one = config.normalize_to_one
separate = config.separate_branches
n_branches = analyzer._n_branches
# 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"# n_points: {len(analyzer.energies_MeV)}")
if to_one:
if separate:
header_lines.append(
"# branch normalization: to_one "
"# branch normalization: separate "
"(each branch integrates to 1, intensity ignored)"
)
else:
@@ -179,8 +179,8 @@ class CSVEmitter:
"energy_MeV": analyzer.energies_MeV,
}
integral: Optional[float] = None
if not to_one:
# 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:
@@ -189,6 +189,7 @@ class CSVEmitter:
"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):
@@ -198,19 +199,23 @@ class CSVEmitter:
else:
col_name = f"branch_{key_i}_spectrum"
if to_one:
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:
assert integral is not None
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)
@@ -218,7 +223,8 @@ class CSVEmitter:
data[f"branch_{i + 1}_{comp_name}"] = components[key]
# 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():
if not name.startswith("branch_"):
data[name] = values
+8 -4
View File
@@ -235,7 +235,6 @@ def _plot_spectra_view_from_df(
) -> Figure:
"""Plot total spectrum and per-branch spectra from CSV columns."""
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))
@@ -266,7 +265,7 @@ def _plot_spectra_view_from_df(
)
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")
ax.set_yscale("log")
ax.grid(True, alpha=0.3)
@@ -288,7 +287,6 @@ def _plot_debug_view_from_df(
) -> Figure:
"""Debug view (4 panels) built entirely from CSV columns."""
energies = df["energy_MeV"].to_numpy()
e_step_keV = (meta["e_step_MeV"] or 0.0) * 1000.0
active = _active_branches(meta)
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}%)",
)
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.grid(True, alpha=0.3)
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)
for color_i, b in enumerate(active):
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:
continue
ps_vals = df[col].to_numpy()
@@ -374,6 +375,9 @@ def _plot_debug_view_from_df(
ax = axes[3]
for color_i, b in enumerate(active):
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:
continue
ax.plot(
+3 -5
View File
@@ -57,9 +57,6 @@ class SpectrumConfig:
"""
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
provided, the total spectrum is the intensity-weighted sum of individual
branch spectra. Each branch has its own endpoint energy and transition type.
@@ -90,9 +87,10 @@ class SpectrumConfig:
# CSV output options
csv_separator: str = "," # Default separator for CSV output
normalize_to_one: bool = (
False # If True, normalize each branch to 1, ignoring intensity
separate_branches: bool = (
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]:
+1 -10
View File
@@ -5,14 +5,5 @@
"A_number": 99,
"endpoint_MeV": 0.294,
"transition_type": "A",
"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]
"e_step_MeV": 0.001
}
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "beta-spectrum"
version = "0.4.0"
version = "0.4.1"
description = "High-precision beta spectrum calculation for nuclear physics"
authors = [
{ 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."""
return SpectrumConfig(
Z_parent=43,
@@ -75,6 +75,7 @@ def make_degenerate_config() -> SpectrumConfig:
endpoint_MeV=0.294,
e_step_MeV=0.001,
neutrino_mass_MeV=0.35,
**over,
branches=[
BranchConfig(endpoint_MeV=0.294, transition_type="F2", intensity=0.98),
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 "fermi" in core_line[0]
def test_normalize_to_one_per_branch_unit_area(self, tmp_path: Any) -> None:
config = make_multi_config(normalize_to_one=True)
def test_separate_branches_per_branch_unit_area(self, tmp_path: Any) -> None:
config = make_multi_config(separate_branches=True)
spectrum = BetaSpectrum.from_config(config)
analyzer = BetaSpectrumAnalyzer(spectrum, config)
_, df = export_and_read(analyzer, tmp_path)
@@ -329,8 +330,22 @@ class TestCSVExportCoreSemantics:
spectrum = BetaSpectrum.from_config(config)
analyzer = BetaSpectrumAnalyzer(spectrum, config)
path = str(tmp_path / "degenerate.csv")
with pytest.raises(ValueError):
with pytest.raises(ValueError, match="non-positive"):
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:
config = make_multi_config()
@@ -549,3 +564,168 @@ class TestAnalyzerUniversalFactor:
assert np.all(uni > 1.0)
fermi_only = analyzer.components["Fermi"]
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"
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
parser = _build_parser()
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()
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)
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"
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
parser = _build_parser()
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)
try:
loaded = load_json_input(str(path))
# Defaults should be filled in
assert loaded["detector_model"] == "gaussian"
assert loaded["detector_sigma_a_keV"] == 1.0
# Defaults should be filled in. The detector_* schema keys were
# removed together with the detector_response module; JSON
# defaults carry spectrum physics only.
assert "detector_model" not in loaded
assert loaded["neutrino_mass_MeV"] == 0.0
assert loaded["transition_type"] == "F2"
finally:
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.skipif(not _has_paceENSDF(), reason="paceENSDF not installed")
class TestPaceENSDFIntegration: