feat: add paceENSDF integration and CLI interface (Track B1)
- Add nuclear_data.py module for paceENSDF integration and JSON input - Add DecayInfo and BranchInfo dataclasses for parsed nuclear data - Implement get_decay_info_from_paceENSDF() to retrieve decay parameters - Implement load_json_input() and json_to_config() for custom JSON input - Add create_config_from_source() unified interface for both sources - Add bs_pnpi CLI with --nuclide, --input, --output, --plot, and detector options - Add 23 comprehensive tests for nuclear_data module - Add sample JSON input file - Update __init__.py and pyproject.toml with new exports and entry point - Add paceENSDF as optional dependency
This commit is contained in:
@@ -14,6 +14,16 @@ from .components.detector_response import DetectorResponse
|
||||
from .spectrum import BetaSpectrum, SpectrumConfig, BetaSpectrumAnalyzer
|
||||
from .fitter import CurveFitter, FitConfig, FitResult
|
||||
from .cw_extractor import CWExtractor, CWExtractionResult, GVAExtractionResult
|
||||
from .nuclear_data import (
|
||||
DecayInfo,
|
||||
BranchInfo,
|
||||
get_decay_info_from_paceENSDF,
|
||||
decay_info_to_config,
|
||||
create_config_from_source,
|
||||
load_json_input,
|
||||
json_to_config,
|
||||
DEFAULT_JSON_SCHEMA,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"ALPHA",
|
||||
@@ -40,6 +50,14 @@ __all__ = [
|
||||
"CWExtractor",
|
||||
"CWExtractionResult",
|
||||
"GVAExtractionResult",
|
||||
"DecayInfo",
|
||||
"BranchInfo",
|
||||
"get_decay_info_from_paceENSDF",
|
||||
"decay_info_to_config",
|
||||
"create_config_from_source",
|
||||
"load_json_input",
|
||||
"json_to_config",
|
||||
"DEFAULT_JSON_SCHEMA",
|
||||
]
|
||||
|
||||
__version__ = "0.3.0"
|
||||
|
||||
@@ -0,0 +1,215 @@
|
||||
"""
|
||||
Command-line interface for beta-spectrum calculations.
|
||||
|
||||
Usage examples:
|
||||
# From paceENSDF nuclear data
|
||||
bs_pnpi --nuclide Tc99 --output spectrum.csv
|
||||
|
||||
# From custom JSON input
|
||||
bs_pnpi --input custom_input.json --output spectrum.csv
|
||||
|
||||
# Full analysis with plotting
|
||||
bs_pnpi --nuclide Co60 --mode BM --plot analysis.png --output spectrum.csv
|
||||
|
||||
# With detector response
|
||||
bs_pnpi --nuclide Tc99 --detector sigma=1.0 --output convolved.csv
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
from typing import Any, Dict
|
||||
|
||||
|
||||
def _build_parser() -> argparse.ArgumentParser:
|
||||
"""Build the argument parser."""
|
||||
parser = argparse.ArgumentParser(
|
||||
prog="bs_pnpi",
|
||||
description="Beta spectrum calculation toolkit — "
|
||||
"calculate high-precision beta decay energy spectra.",
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
epilog=(
|
||||
"Examples:\n"
|
||||
" bs_pnpi --nuclide Tc99 --output spectrum.csv\n"
|
||||
" bs_pnpi --input custom_input.json --output spectrum.csv\n"
|
||||
" bs_pnpi --nuclide Tc99 --plot spectrum.png --output spectrum.csv\n"
|
||||
" bs_pnpi --nuclide Tc99 --detector sigma=1.0 --output convolved.csv\n"
|
||||
),
|
||||
)
|
||||
|
||||
# Source selection (mutually exclusive)
|
||||
source_group = parser.add_mutually_exclusive_group(required=True)
|
||||
source_group.add_argument(
|
||||
"--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(
|
||||
"--input",
|
||||
type=str,
|
||||
metavar="FILE",
|
||||
help="Path to a JSON input file with custom parameters.",
|
||||
)
|
||||
|
||||
# Common parameters
|
||||
parser.add_argument(
|
||||
"--mode",
|
||||
type=str,
|
||||
default="beta_minus",
|
||||
choices=["beta_minus", "beta_plus", "ec"],
|
||||
help="Decay mode for paceENSDF source (default: beta_minus).",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--decay-index",
|
||||
type=int,
|
||||
default=None,
|
||||
help="Decay index for paceENSDF (0 = ground state, default: auto-select).",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--transition-type",
|
||||
type=str,
|
||||
default=None,
|
||||
choices=["A", "F1", "F1U", "F2", "F2U", "F3", "F3U", "F4"],
|
||||
help="Override transition type (forbiddenness classification).",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--e-step",
|
||||
type=float,
|
||||
default=0.001,
|
||||
help="Energy step size in MeV (default: 0.001).",
|
||||
)
|
||||
|
||||
# Output
|
||||
parser.add_argument(
|
||||
"--output",
|
||||
type=str,
|
||||
metavar="FILE",
|
||||
help="Output CSV file path.",
|
||||
)
|
||||
|
||||
# Plotting
|
||||
parser.add_argument(
|
||||
"--plot",
|
||||
type=str,
|
||||
metavar="FILE",
|
||||
help="Save analysis plot to the specified path.",
|
||||
)
|
||||
|
||||
# Detector response
|
||||
detector_group = parser.add_argument_group("detector response")
|
||||
detector_group.add_argument(
|
||||
"--detector",
|
||||
type=str,
|
||||
default=None,
|
||||
metavar="MODEL",
|
||||
help="Enable detector response convolution. "
|
||||
"Specify model: 'gaussian', 'gaussian_tail', 'tikhonov', 'tabulated'.",
|
||||
)
|
||||
detector_group.add_argument(
|
||||
"--sigma",
|
||||
type=float,
|
||||
default=1.0,
|
||||
metavar="KEV",
|
||||
help="Resolution parameter sigma in keV (default: 1.0).",
|
||||
)
|
||||
detector_group.add_argument(
|
||||
"--tau",
|
||||
type=float,
|
||||
default=5.0,
|
||||
metavar="KEV",
|
||||
help="Tail decay constant tau in keV (default: 5.0).",
|
||||
)
|
||||
detector_group.add_argument(
|
||||
"--tail-fraction",
|
||||
type=float,
|
||||
default=0.0,
|
||||
metavar="FRAC",
|
||||
help="Low-energy tail fraction (0.0–1.0, default: 0.0).",
|
||||
)
|
||||
|
||||
return parser
|
||||
|
||||
|
||||
def _parse_detector_args(args: argparse.Namespace) -> Dict[str, Any]:
|
||||
"""Parse detector-related arguments into a config dict."""
|
||||
if not args.detector:
|
||||
return {}
|
||||
|
||||
return {
|
||||
"use_detector_response": True,
|
||||
"detector_model": args.detector,
|
||||
"detector_sigma_a_keV": args.sigma,
|
||||
"detector_tau_keV": args.tau,
|
||||
"detector_tail_fraction": args.tail_fraction,
|
||||
}
|
||||
|
||||
|
||||
def _run(args: argparse.Namespace) -> None:
|
||||
"""Execute the CLI command."""
|
||||
from beta_spectrum.nuclear_data import (
|
||||
create_config_from_source,
|
||||
json_to_config,
|
||||
load_json_input,
|
||||
)
|
||||
from beta_spectrum.spectrum import BetaSpectrum, BetaSpectrumAnalyzer
|
||||
|
||||
# Build config
|
||||
detector_kwargs = _parse_detector_args(args)
|
||||
|
||||
if args.nuclide:
|
||||
config = create_config_from_source(
|
||||
"paceENSDF",
|
||||
nuclide=args.nuclide,
|
||||
decay_type=args.mode,
|
||||
decay_index=args.decay_index,
|
||||
e_step_MeV=args.e_step,
|
||||
use_detector_response=detector_kwargs.get("use_detector_response", False),
|
||||
detector_model=detector_kwargs.get("detector_model", "gaussian"),
|
||||
detector_sigma_a_keV=detector_kwargs.get("detector_sigma_a_keV", 1.0),
|
||||
detector_tau_keV=detector_kwargs.get("detector_tau_keV", 5.0),
|
||||
detector_tail_fraction=detector_kwargs.get("detector_tail_fraction", 0.0),
|
||||
)
|
||||
elif args.input:
|
||||
data = load_json_input(args.input)
|
||||
if args.transition_type:
|
||||
data["transition_type"] = args.transition_type
|
||||
config = json_to_config(data)
|
||||
|
||||
# Override transition type if explicitly provided
|
||||
if args.transition_type:
|
||||
config.transition_type = args.transition_type
|
||||
|
||||
# Create and calculate spectrum
|
||||
spectrum = BetaSpectrum.from_config(config)
|
||||
_W, _kinetic_MeV = spectrum.get_energy_grid(config)
|
||||
_ = spectrum(_W)
|
||||
|
||||
# Export to CSV
|
||||
if args.output:
|
||||
analyzer = BetaSpectrumAnalyzer(spectrum, config)
|
||||
analyzer.export_to_csv(args.output)
|
||||
print(f"Spectrum exported to {args.output}")
|
||||
|
||||
# Generate plot
|
||||
if args.plot:
|
||||
analyzer = BetaSpectrumAnalyzer(spectrum, config)
|
||||
analyzer.plot_analysis(save_path=args.plot)
|
||||
print(f"Analysis plot saved to {args.plot}")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
"""Main entry point for the CLI."""
|
||||
parser = _build_parser()
|
||||
args = parser.parse_args()
|
||||
|
||||
try:
|
||||
_run(args)
|
||||
except Exception as e:
|
||||
print(f"Error: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,654 @@
|
||||
"""
|
||||
Nuclear data integration module.
|
||||
|
||||
Provides interfaces to retrieve nuclear decay data from paceENSDF
|
||||
and from custom JSON input files. Both sources produce a valid
|
||||
SpectrumConfig for beta spectrum calculation.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
|
||||
from beta_spectrum.spectrum import SpectrumConfig
|
||||
|
||||
# Mapping from paceENSDF forbiddenness codes to our transition_type
|
||||
FORBIDDENNESS_MAP: Dict[str, str] = {
|
||||
"0A": "A",
|
||||
"1A": "A",
|
||||
"1F": "F1",
|
||||
"1UF": "F1U",
|
||||
"2F": "F2",
|
||||
"2UF": "F2U",
|
||||
"3F": "F3",
|
||||
"3UF": "F3U",
|
||||
"4F": "F4",
|
||||
"4UF": "F4",
|
||||
"5F": "F4",
|
||||
"5UF": "F4",
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class DecayInfo:
|
||||
"""
|
||||
Parsed nuclear decay information from paceENSDF or JSON.
|
||||
|
||||
Attributes
|
||||
----------
|
||||
parent_symbol : str
|
||||
Parent isotope symbol (e.g. "Tc99").
|
||||
Z_parent : int
|
||||
Atomic number of parent.
|
||||
Z_daughter : int
|
||||
Atomic number of daughter.
|
||||
A_number : int
|
||||
Mass number (same for parent and daughter in beta decay).
|
||||
endpoint_keV : float
|
||||
Q-value in keV.
|
||||
endpoint_MeV : float
|
||||
Q-value in MeV.
|
||||
transition_type : str
|
||||
Forbiddenness classification (A, F1, F1U, F2, F2U, ...).
|
||||
forbiddenness_code : str
|
||||
Original paceENSDF forbiddenness code (e.g. "2F").
|
||||
parent_J : float
|
||||
Parent spin.
|
||||
parent_pi : int
|
||||
Parent parity (1=+, -1=-).
|
||||
half_life : str
|
||||
Half-life string from ENSDF (e.g. "y", "h").
|
||||
branches : List[BranchInfo]
|
||||
List of decay branches to daughter states.
|
||||
"""
|
||||
|
||||
parent_symbol: str
|
||||
Z_parent: int
|
||||
Z_daughter: int
|
||||
A_number: int
|
||||
endpoint_keV: float
|
||||
endpoint_MeV: float
|
||||
transition_type: str
|
||||
forbiddenness_code: str
|
||||
parent_J: float
|
||||
parent_pi: int
|
||||
half_life: str
|
||||
branches: List[BranchInfo]
|
||||
|
||||
|
||||
@dataclass
|
||||
class BranchInfo:
|
||||
"""
|
||||
A single decay branch to a daughter state.
|
||||
|
||||
Attributes
|
||||
----------
|
||||
level_index : int
|
||||
Level index in daughter nucleus.
|
||||
level_energy_keV : float
|
||||
Excitation energy of the daughter level.
|
||||
intensity : float
|
||||
Branch intensity (fraction of decays).
|
||||
log_ft : float
|
||||
log(ft) value for this branch.
|
||||
"""
|
||||
|
||||
level_index: int
|
||||
level_energy_keV: float
|
||||
intensity: float
|
||||
log_ft: Optional[float]
|
||||
|
||||
|
||||
def _parse_nuclide_symbol(symbol: str) -> Tuple[str, int, int]:
|
||||
"""
|
||||
Parse a nuclide symbol like 'Tc99' into (element, Z, A).
|
||||
|
||||
Parameters
|
||||
----------
|
||||
symbol : str
|
||||
Nuclide symbol, e.g. "Tc99", "Co60", "Y86".
|
||||
|
||||
Returns
|
||||
-------
|
||||
tuple[str, int, int]
|
||||
(element_symbol, Z, A)
|
||||
|
||||
Raises
|
||||
------
|
||||
ValueError
|
||||
If the symbol cannot be parsed.
|
||||
"""
|
||||
# Separate letters from digits
|
||||
i = 0
|
||||
while i < len(symbol) and symbol[i].isalpha():
|
||||
i += 1
|
||||
|
||||
if i == 0 or i == len(symbol):
|
||||
raise ValueError(
|
||||
f"Cannot parse nuclide symbol '{symbol}'. "
|
||||
f"Expected format like 'Tc99', 'Co60', 'Y86'."
|
||||
)
|
||||
|
||||
element = symbol[:i]
|
||||
A = int(symbol[i:])
|
||||
|
||||
# Look up Z from element symbol
|
||||
element_to_Z: Dict[str, int] = {
|
||||
"H": 1,
|
||||
"He": 2,
|
||||
"Li": 3,
|
||||
"Be": 4,
|
||||
"B": 5,
|
||||
"C": 6,
|
||||
"N": 7,
|
||||
"O": 8,
|
||||
"F": 9,
|
||||
"Ne": 10,
|
||||
"Na": 11,
|
||||
"Mg": 12,
|
||||
"Al": 13,
|
||||
"Si": 14,
|
||||
"P": 15,
|
||||
"S": 16,
|
||||
"Cl": 17,
|
||||
"Ar": 18,
|
||||
"K": 19,
|
||||
"Ca": 20,
|
||||
"Sc": 21,
|
||||
"Ti": 22,
|
||||
"V": 23,
|
||||
"Cr": 24,
|
||||
"Mn": 25,
|
||||
"Fe": 26,
|
||||
"Co": 27,
|
||||
"Ni": 28,
|
||||
"Cu": 29,
|
||||
"Zn": 30,
|
||||
"Ga": 31,
|
||||
"Ge": 32,
|
||||
"As": 33,
|
||||
"Se": 34,
|
||||
"Br": 35,
|
||||
"Kr": 36,
|
||||
"Rb": 37,
|
||||
"Sr": 38,
|
||||
"Y": 39,
|
||||
"Zr": 40,
|
||||
"Nb": 41,
|
||||
"Mo": 42,
|
||||
"Tc": 43,
|
||||
"Ru": 44,
|
||||
"Rh": 45,
|
||||
"Pd": 46,
|
||||
"Ag": 47,
|
||||
"Cd": 48,
|
||||
"In": 49,
|
||||
"Sn": 50,
|
||||
"Sb": 51,
|
||||
"Te": 52,
|
||||
"I": 53,
|
||||
"Xe": 54,
|
||||
"Cs": 55,
|
||||
"Ba": 56,
|
||||
"La": 57,
|
||||
"Ce": 58,
|
||||
"Pr": 59,
|
||||
"Nd": 60,
|
||||
"Pm": 61,
|
||||
"Sm": 62,
|
||||
"Eu": 63,
|
||||
"Gd": 64,
|
||||
"Tb": 65,
|
||||
"Dy": 66,
|
||||
"Ho": 67,
|
||||
"Er": 68,
|
||||
"Tm": 69,
|
||||
"Yb": 70,
|
||||
"Lu": 71,
|
||||
"Hf": 72,
|
||||
"Ta": 73,
|
||||
"W": 74,
|
||||
"Re": 75,
|
||||
"Os": 76,
|
||||
"Ir": 77,
|
||||
"Pt": 78,
|
||||
"Au": 79,
|
||||
"Hg": 80,
|
||||
"Tl": 81,
|
||||
"Pb": 82,
|
||||
"Bi": 83,
|
||||
"Po": 84,
|
||||
"At": 85,
|
||||
"Rn": 86,
|
||||
"Fr": 87,
|
||||
"Ra": 88,
|
||||
"Ac": 89,
|
||||
"Th": 90,
|
||||
"Pa": 91,
|
||||
"U": 92,
|
||||
"Np": 93,
|
||||
"Pu": 94,
|
||||
"Am": 95,
|
||||
"Cm": 96,
|
||||
"Bk": 97,
|
||||
"Cf": 98,
|
||||
}
|
||||
|
||||
if element not in element_to_Z:
|
||||
raise ValueError(
|
||||
f"Unknown element symbol '{element}'. "
|
||||
f"Supported: {sorted(element_to_Z.keys())}"
|
||||
)
|
||||
|
||||
return element, element_to_Z[element], A
|
||||
|
||||
|
||||
def _get_decay_mode_symbol(
|
||||
decay_type: str,
|
||||
) -> str:
|
||||
"""
|
||||
Map internal decay type to paceENSDF mode string.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
decay_type : str
|
||||
One of 'beta_minus', 'beta_plus', 'ec'.
|
||||
|
||||
Returns
|
||||
-------
|
||||
str
|
||||
paceENSDF mode string: 'BM', 'ECBP'.
|
||||
"""
|
||||
mapping = {
|
||||
"beta_minus": "BM",
|
||||
"beta_plus": "ECBP",
|
||||
"ec": "ECBP",
|
||||
}
|
||||
if decay_type not in mapping:
|
||||
raise ValueError(
|
||||
f"Unknown decay type '{decay_type}'. " f"Supported: {list(mapping.keys())}"
|
||||
)
|
||||
return mapping[decay_type]
|
||||
|
||||
|
||||
def _resolve_decay_index(
|
||||
edata: list[Any],
|
||||
parent_symbol: str,
|
||||
mode: str,
|
||||
decay_index: Optional[int] = None,
|
||||
) -> int:
|
||||
"""
|
||||
Find the correct decay index for a parent isotope.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
edata : list
|
||||
paceENSDF data from load_ensdf().
|
||||
parent_symbol : str
|
||||
Parent isotope symbol.
|
||||
mode : str
|
||||
paceENSDF mode ('BM', 'ECBP', 'A').
|
||||
decay_index : int, optional
|
||||
Explicit index. If None, selects ground state (0) if available,
|
||||
otherwise the first available decay.
|
||||
|
||||
Returns
|
||||
-------
|
||||
int
|
||||
The decay index to use.
|
||||
|
||||
Raises
|
||||
------
|
||||
ValueError
|
||||
If no matching decay is found.
|
||||
"""
|
||||
import paceENSDF as pe
|
||||
|
||||
e = pe.ENSDF()
|
||||
pairs = e.ensdf_pairs(edata, mode)
|
||||
|
||||
matching = [(k, v) for k, v in pairs.items() if str(k[0]) == parent_symbol]
|
||||
|
||||
if not matching:
|
||||
raise ValueError(
|
||||
f"No {mode} decay data found for {parent_symbol}. "
|
||||
f"Available {mode} parents: {[str(k[0]) for k in pairs.keys()]}"
|
||||
)
|
||||
|
||||
if decay_index is not None:
|
||||
for k, v in matching:
|
||||
if k[3] == decay_index:
|
||||
return decay_index
|
||||
raise ValueError(
|
||||
f"Decay index {decay_index} not found for {parent_symbol}. "
|
||||
f"Available indices: {[k[3] for k, v in matching]}"
|
||||
)
|
||||
|
||||
# Default: ground state (index 0), or first available
|
||||
for k, v in matching:
|
||||
if k[3] == 0:
|
||||
return 0
|
||||
return int(matching[0][0][3])
|
||||
|
||||
|
||||
def get_decay_info_from_paceENSDF(
|
||||
nuclide: str,
|
||||
decay_type: str = "beta_minus",
|
||||
decay_index: Optional[int] = None,
|
||||
) -> DecayInfo:
|
||||
"""
|
||||
Retrieve decay information for a nuclide from paceENSDF.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
nuclide : str
|
||||
Nuclide symbol, e.g. "Tc99", "Co60".
|
||||
decay_type : str
|
||||
Decay mode: 'beta_minus', 'beta_plus', or 'ec'.
|
||||
decay_index : int, optional
|
||||
Decay index (0 = ground state). If None, defaults to ground state.
|
||||
|
||||
Returns
|
||||
-------
|
||||
DecayInfo
|
||||
Parsed decay information.
|
||||
|
||||
Examples
|
||||
--------
|
||||
>>> info = get_decay_info_from_paceENSDF("Tc99", "beta_minus")
|
||||
>>> config = decay_info_to_config(info)
|
||||
"""
|
||||
import paceENSDF as pe
|
||||
|
||||
e = pe.ENSDF()
|
||||
edata = e.load_ensdf()
|
||||
|
||||
mode = _get_decay_mode_symbol(decay_type)
|
||||
_element, Z_parent, A_number = _parse_nuclide_symbol(nuclide)
|
||||
decay_idx = _resolve_decay_index(edata, nuclide, mode, decay_index)
|
||||
|
||||
# Get parent decay info
|
||||
_decay_data = e.get_parent_decay(edata, nuclide, decay_idx, mode=mode)
|
||||
halflife_data = e.get_parent_halflife(
|
||||
edata, nuclide, decay_idx, mode=mode, units="best"
|
||||
)
|
||||
jpi_data = e.get_parent_jpi(edata, nuclide, decay_idx, mode=mode)
|
||||
beta_data = e.get_beta_minus(edata, nuclide, decay_idx, units="best")
|
||||
|
||||
# Extract Q-value and half-life
|
||||
q_value_keV = 0.0
|
||||
half_life_str = "unknown"
|
||||
for k, v in halflife_data.items():
|
||||
q_value_keV = v[0]
|
||||
half_life_str = str(v[2]) if len(v) > 2 else "unknown"
|
||||
break
|
||||
|
||||
# Extract parent spin/parity
|
||||
parent_J = jpi_data[0][0] if jpi_data else 0.0
|
||||
parent_pi = jpi_data[0][1] if jpi_data else 1
|
||||
|
||||
# Extract daughter Z from decay data
|
||||
Z_daughter = Z_parent + 1 if decay_type == "beta_minus" else Z_parent - 1
|
||||
|
||||
# Extract branches and forbiddenness info
|
||||
branches: List[BranchInfo] = []
|
||||
forbidden_code = "0A" # default
|
||||
transition_type = "A" # default
|
||||
|
||||
for k, v in beta_data.items():
|
||||
for state in v:
|
||||
# state[15] contains forbiddenness code (e.g. '2F', '1UF')
|
||||
if len(state) > 15 and state[15] is not None:
|
||||
code = str(state[15])
|
||||
if code in FORBIDDENNESS_MAP:
|
||||
forbidden_code = code
|
||||
transition_type = FORBIDDENNESS_MAP[code]
|
||||
|
||||
branch = BranchInfo(
|
||||
level_index=int(state[0]),
|
||||
level_energy_keV=float(state[1]),
|
||||
intensity=float(state[8]),
|
||||
log_ft=float(state[11]) if state[11] is not None else None,
|
||||
)
|
||||
branches.append(branch)
|
||||
|
||||
return DecayInfo(
|
||||
parent_symbol=nuclide,
|
||||
Z_parent=Z_parent,
|
||||
Z_daughter=Z_daughter,
|
||||
A_number=A_number,
|
||||
endpoint_keV=q_value_keV,
|
||||
endpoint_MeV=q_value_keV / 1000.0,
|
||||
transition_type=transition_type,
|
||||
forbiddenness_code=forbidden_code,
|
||||
parent_J=parent_J,
|
||||
parent_pi=parent_pi,
|
||||
half_life=half_life_str,
|
||||
branches=branches,
|
||||
)
|
||||
|
||||
|
||||
def decay_info_to_config(
|
||||
info: DecayInfo,
|
||||
e_step_MeV: float = 0.001,
|
||||
use_detector_response: bool = False,
|
||||
) -> SpectrumConfig:
|
||||
"""
|
||||
Convert DecayInfo to a SpectrumConfig.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
info : DecayInfo
|
||||
Parsed decay information.
|
||||
e_step_MeV : float
|
||||
Energy step size in MeV.
|
||||
use_detector_response : bool
|
||||
Whether to enable detector response convolution.
|
||||
|
||||
Returns
|
||||
-------
|
||||
SpectrumConfig
|
||||
Configuration ready for BetaSpectrum.from_config().
|
||||
"""
|
||||
return SpectrumConfig(
|
||||
Z_parent=info.Z_parent,
|
||||
Z_daughter=info.Z_daughter,
|
||||
A_number=info.A_number,
|
||||
endpoint_MeV=info.endpoint_MeV,
|
||||
transition_type=info.transition_type,
|
||||
e_step_MeV=e_step_MeV,
|
||||
use_detector_response=use_detector_response,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# JSON input support
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
# Default JSON schema
|
||||
DEFAULT_JSON_SCHEMA = {
|
||||
"parent_symbol": "Tc99",
|
||||
"decay_type": "beta_minus",
|
||||
"endpoint_MeV": 0.2111,
|
||||
"Z_parent": 43,
|
||||
"Z_daughter": 44,
|
||||
"A_number": 99,
|
||||
"transition_type": "F2",
|
||||
"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],
|
||||
}
|
||||
|
||||
|
||||
def validate_json_input(data: Dict[str, Any]) -> None:
|
||||
"""
|
||||
Validate a JSON input dictionary.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
data : dict
|
||||
Parsed JSON data.
|
||||
|
||||
Raises
|
||||
------
|
||||
ValueError
|
||||
If required fields are missing or invalid.
|
||||
"""
|
||||
required = ["endpoint_MeV", "Z_parent", "Z_daughter", "A_number"]
|
||||
for field in required:
|
||||
if field not in data:
|
||||
raise ValueError(f"Missing required field: '{field}'")
|
||||
|
||||
if data["endpoint_MeV"] <= 0:
|
||||
raise ValueError("endpoint_MeV must be positive")
|
||||
|
||||
if data["Z_parent"] <= 0 or data["Z_daughter"] <= 0:
|
||||
raise ValueError("Z_parent and Z_daughter must be positive")
|
||||
|
||||
if data["A_number"] <= 0:
|
||||
raise ValueError("A_number must be positive")
|
||||
|
||||
if data.get("transition_type", "A") not in [
|
||||
"A",
|
||||
"F1",
|
||||
"F1U",
|
||||
"F2",
|
||||
"F2U",
|
||||
"F3",
|
||||
"F3U",
|
||||
"F4",
|
||||
]:
|
||||
raise ValueError(
|
||||
f"Invalid transition_type: '{data['transition_type']}'. "
|
||||
f"Supported: A, F1, F1U, F2, F2U, F3, F3U, F4"
|
||||
)
|
||||
|
||||
|
||||
def load_json_input(filepath: str) -> Dict[str, Any]:
|
||||
"""
|
||||
Load and validate a JSON input file.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
filepath : str
|
||||
Path to the JSON file.
|
||||
|
||||
Returns
|
||||
-------
|
||||
dict
|
||||
Validated input data with defaults filled in.
|
||||
"""
|
||||
path = Path(filepath)
|
||||
if not path.exists():
|
||||
raise FileNotFoundError(f"Input file not found: {filepath}")
|
||||
|
||||
with open(path, "r") as f:
|
||||
data = json.load(f)
|
||||
|
||||
# Fill in defaults for missing optional fields
|
||||
for key, default in DEFAULT_JSON_SCHEMA.items():
|
||||
if key not in data:
|
||||
data[key] = default
|
||||
|
||||
validate_json_input(data)
|
||||
return data
|
||||
|
||||
|
||||
def json_to_config(data: Dict[str, Any]) -> SpectrumConfig:
|
||||
"""
|
||||
Convert a validated JSON input dictionary to SpectrumConfig.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
data : dict
|
||||
Validated JSON input data.
|
||||
|
||||
Returns
|
||||
-------
|
||||
SpectrumConfig
|
||||
Configuration ready for BetaSpectrum.from_config().
|
||||
"""
|
||||
return SpectrumConfig(
|
||||
Z_parent=int(data["Z_parent"]),
|
||||
Z_daughter=int(data["Z_daughter"]),
|
||||
A_number=int(data["A_number"]),
|
||||
endpoint_MeV=float(data["endpoint_MeV"]),
|
||||
transition_type=str(data.get("transition_type", "A")),
|
||||
e_step_MeV=float(data.get("e_step_MeV", 0.001)),
|
||||
use_detector_response=bool(data.get("use_detector_response", False)),
|
||||
detector_model=str(data.get("detector_model", "gaussian")),
|
||||
detector_sigma_a_keV=float(data.get("detector_sigma_a_keV", 1.0)),
|
||||
detector_sigma_b=float(data.get("detector_sigma_b", 0.0)),
|
||||
detector_tail_fraction=float(data.get("detector_tail_fraction", 0.0)),
|
||||
detector_tau_keV=float(data.get("detector_tau_keV", 5.0)),
|
||||
detector_fano_factor=float(data.get("detector_fano_factor", 0.12)),
|
||||
detector_n_channels=int(data.get("detector_n_channels", 4096)),
|
||||
detector_channel_energy_range=tuple(
|
||||
data.get("detector_channel_energy_range", [0.0, 0.35])
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def create_config_from_source(
|
||||
source: str,
|
||||
nuclide: Optional[str] = None,
|
||||
json_path: Optional[str] = None,
|
||||
**kwargs: Any,
|
||||
) -> SpectrumConfig: # type: ignore[return-value]
|
||||
"""
|
||||
Create a SpectrumConfig from either paceENSDF or JSON input.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
source : str
|
||||
Data source: 'paceENSDF' or 'json'.
|
||||
nuclide : str, optional
|
||||
Nuclide symbol for paceENSDF source (e.g. "Tc99").
|
||||
json_path : str, optional
|
||||
Path to JSON input file for JSON source.
|
||||
**kwargs
|
||||
Additional parameters passed to decay_info_to_config or json_to_config.
|
||||
|
||||
Returns
|
||||
-------
|
||||
SpectrumConfig
|
||||
Configuration ready for BetaSpectrum.from_config().
|
||||
|
||||
Raises
|
||||
------
|
||||
ValueError
|
||||
If source is unknown or required arguments are missing.
|
||||
|
||||
Examples
|
||||
--------
|
||||
>>> config = create_config_from_source("paceENSDF", nuclide="Tc99")
|
||||
>>> config = create_config_from_source("json", json_path="custom_input.json")
|
||||
"""
|
||||
if source == "paceENSDF":
|
||||
if not nuclide:
|
||||
raise ValueError("nuclide parameter required for paceENSDF source")
|
||||
decay_type = kwargs.get("decay_type", "beta_minus")
|
||||
decay_index = kwargs.get("decay_index", None)
|
||||
e_step = kwargs.get("e_step_MeV", 0.001)
|
||||
use_det = kwargs.get("use_detector_response", False)
|
||||
info = get_decay_info_from_paceENSDF(nuclide, decay_type, decay_index)
|
||||
return decay_info_to_config(
|
||||
info, e_step_MeV=e_step, use_detector_response=use_det
|
||||
)
|
||||
elif source == "json":
|
||||
if not json_path:
|
||||
raise ValueError("json_path parameter required for JSON source")
|
||||
data = load_json_input(json_path)
|
||||
return json_to_config(data)
|
||||
else:
|
||||
raise ValueError(f"Unknown source: '{source}'. Supported: 'paceENSDF', 'json'")
|
||||
@@ -0,0 +1,18 @@
|
||||
{
|
||||
"parent_symbol": "Tc99",
|
||||
"Z_parent": 43,
|
||||
"Z_daughter": 44,
|
||||
"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]
|
||||
}
|
||||
+8
-4
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "beta-spectrum"
|
||||
version = "0.1.2"
|
||||
version = "0.3.0"
|
||||
description = "High-precision beta spectrum calculation for nuclear physics"
|
||||
authors = [
|
||||
{ name = "Evgeniy Unzhakov" }
|
||||
@@ -18,10 +18,8 @@ dependencies = [
|
||||
"pandas"
|
||||
]
|
||||
|
||||
[tool.setuptools.packages.find]
|
||||
where = ["."]
|
||||
|
||||
[project.optional-dependencies]
|
||||
ensdf = ["paceENSDF"]
|
||||
dev = [
|
||||
"pytest",
|
||||
"nbmake",
|
||||
@@ -34,6 +32,12 @@ dev = [
|
||||
"mdformat-tables"
|
||||
]
|
||||
|
||||
[project.scripts]
|
||||
bs_pnpi = "beta_spectrum.cli:main"
|
||||
|
||||
[tool.setuptools.packages.find]
|
||||
where = ["."]
|
||||
|
||||
[tool.mypy]
|
||||
python_version = "3.10"
|
||||
strict = true
|
||||
|
||||
@@ -0,0 +1,313 @@
|
||||
"""Tests for nuclear_data module (paceENSDF integration and JSON input)."""
|
||||
|
||||
import json
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
# ruff: noqa: E402, ANN201, ANN001
|
||||
from beta_spectrum.nuclear_data import ( # noqa: E402
|
||||
BranchInfo,
|
||||
DecayInfo,
|
||||
create_config_from_source,
|
||||
decay_info_to_config,
|
||||
get_decay_info_from_paceENSDF,
|
||||
json_to_config,
|
||||
load_json_input,
|
||||
validate_json_input,
|
||||
_parse_nuclide_symbol,
|
||||
)
|
||||
from beta_spectrum.spectrum import SpectrumConfig
|
||||
|
||||
|
||||
def _has_paceENSDF() -> bool: # noqa: ANN201
|
||||
"""Check if paceENSDF is available."""
|
||||
try:
|
||||
import paceENSDF # noqa: F401
|
||||
|
||||
return True
|
||||
except ImportError:
|
||||
return False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _parse_nuclide_symbol
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestParseNuclideSymbol:
|
||||
"""Test nuclide symbol parsing."""
|
||||
|
||||
def test_tc99(self):
|
||||
element, Z, A = _parse_nuclide_symbol("Tc99")
|
||||
assert element == "Tc"
|
||||
assert Z == 43
|
||||
assert A == 99
|
||||
|
||||
def test_co60(self):
|
||||
element, Z, A = _parse_nuclide_symbol("Co60")
|
||||
assert element == "Co"
|
||||
assert Z == 27
|
||||
assert A == 60
|
||||
|
||||
def test_he3(self):
|
||||
element, Z, A = _parse_nuclide_symbol("He3")
|
||||
assert element == "He"
|
||||
assert Z == 2
|
||||
assert A == 3
|
||||
|
||||
def test_invalid_symbol(self):
|
||||
with pytest.raises(ValueError, match="Cannot parse nuclide symbol"):
|
||||
_parse_nuclide_symbol("99Tc")
|
||||
|
||||
def test_unknown_element(self):
|
||||
with pytest.raises(ValueError, match="Unknown element symbol"):
|
||||
_parse_nuclide_symbol("Xx99")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# validate_json_input
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestValidateJsonInput:
|
||||
"""Test JSON input validation."""
|
||||
|
||||
def test_valid_input(self):
|
||||
data = {
|
||||
"endpoint_MeV": 0.5,
|
||||
"Z_parent": 43,
|
||||
"Z_daughter": 44,
|
||||
"A_number": 99,
|
||||
}
|
||||
validate_json_input(data) # should not raise
|
||||
|
||||
def test_missing_endpoint(self):
|
||||
data = {"Z_parent": 43, "Z_daughter": 44, "A_number": 99}
|
||||
with pytest.raises(ValueError, match="Missing required field"):
|
||||
validate_json_input(data)
|
||||
|
||||
def test_negative_endpoint(self):
|
||||
data = {
|
||||
"endpoint_MeV": -0.5,
|
||||
"Z_parent": 43,
|
||||
"Z_daughter": 44,
|
||||
"A_number": 99,
|
||||
}
|
||||
with pytest.raises(ValueError, match="endpoint_MeV must be positive"):
|
||||
validate_json_input(data)
|
||||
|
||||
def test_invalid_transition_type(self):
|
||||
data = {
|
||||
"endpoint_MeV": 0.5,
|
||||
"Z_parent": 43,
|
||||
"Z_daughter": 44,
|
||||
"A_number": 99,
|
||||
"transition_type": "invalid",
|
||||
}
|
||||
with pytest.raises(ValueError, match="Invalid transition_type"):
|
||||
validate_json_input(data)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# load_json_input / json_to_config
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestJsonInput:
|
||||
"""Test JSON file loading and conversion."""
|
||||
|
||||
def _write_json(self, data: dict) -> Path:
|
||||
"""Write data to a temp JSON file."""
|
||||
tmp = tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False)
|
||||
json.dump(data, tmp)
|
||||
tmp.close()
|
||||
return Path(tmp.name)
|
||||
|
||||
def test_load_and_convert(self):
|
||||
data = {
|
||||
"parent_symbol": "Tc99",
|
||||
"Z_parent": 43,
|
||||
"Z_daughter": 44,
|
||||
"A_number": 99,
|
||||
"endpoint_MeV": 0.294,
|
||||
"transition_type": "F2",
|
||||
"e_step_MeV": 0.001,
|
||||
}
|
||||
path = self._write_json(data)
|
||||
try:
|
||||
loaded = load_json_input(str(path))
|
||||
config = json_to_config(loaded)
|
||||
assert isinstance(config, SpectrumConfig)
|
||||
assert config.Z_parent == 43
|
||||
assert config.Z_daughter == 44
|
||||
assert config.A_number == 99
|
||||
assert abs(config.endpoint_MeV - 0.294) < 1e-6
|
||||
assert config.transition_type == "F2"
|
||||
assert config.e_step_MeV == 0.001
|
||||
finally:
|
||||
path.unlink()
|
||||
|
||||
def test_defaults_filled(self):
|
||||
data = {
|
||||
"Z_parent": 43,
|
||||
"Z_daughter": 44,
|
||||
"A_number": 99,
|
||||
"endpoint_MeV": 0.294,
|
||||
}
|
||||
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
|
||||
finally:
|
||||
path.unlink()
|
||||
|
||||
def test_file_not_found(self):
|
||||
with pytest.raises(FileNotFoundError):
|
||||
load_json_input("/nonexistent/file.json")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# decay_info_to_config
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestDecayInfoToConfig:
|
||||
"""Test conversion from DecayInfo to SpectrumConfig."""
|
||||
|
||||
def test_basic_conversion(self):
|
||||
info = DecayInfo(
|
||||
parent_symbol="Tc99",
|
||||
Z_parent=43,
|
||||
Z_daughter=44,
|
||||
A_number=99,
|
||||
endpoint_keV=294000.0,
|
||||
endpoint_MeV=0.294,
|
||||
transition_type="A",
|
||||
forbiddenness_code="0A",
|
||||
parent_J=4.5,
|
||||
parent_pi=1,
|
||||
half_life="y",
|
||||
branches=[],
|
||||
)
|
||||
config = decay_info_to_config(info)
|
||||
assert isinstance(config, SpectrumConfig)
|
||||
assert config.Z_parent == 43
|
||||
assert config.Z_daughter == 44
|
||||
assert config.A_number == 99
|
||||
assert abs(config.endpoint_MeV - 0.294) < 1e-6
|
||||
assert config.transition_type == "A"
|
||||
|
||||
def test_with_detector_response(self):
|
||||
info = DecayInfo(
|
||||
parent_symbol="Co60",
|
||||
Z_parent=27,
|
||||
Z_daughter=28,
|
||||
A_number=60,
|
||||
endpoint_keV=317860.0,
|
||||
endpoint_MeV=0.31786,
|
||||
transition_type="A",
|
||||
forbiddenness_code="1A",
|
||||
parent_J=0.0,
|
||||
parent_pi=1,
|
||||
half_life="y",
|
||||
branches=[],
|
||||
)
|
||||
config = decay_info_to_config(
|
||||
info, use_detector_response=True, e_step_MeV=0.005
|
||||
)
|
||||
assert config.use_detector_response is True
|
||||
assert config.e_step_MeV == 0.005
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# get_decay_info_from_paceENSDF
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.skipif(not _has_paceENSDF(), reason="paceENSDF not installed")
|
||||
class TestPaceENSDFIntegration:
|
||||
"""Test paceENSDF data retrieval."""
|
||||
|
||||
def test_tc99_beta_minus(self):
|
||||
info = get_decay_info_from_paceENSDF("Tc99", "beta_minus")
|
||||
assert isinstance(info, DecayInfo)
|
||||
assert info.parent_symbol == "Tc99"
|
||||
assert info.Z_parent == 43
|
||||
assert info.Z_daughter == 44
|
||||
assert info.A_number == 99
|
||||
assert info.endpoint_MeV > 0
|
||||
assert len(info.branches) > 0
|
||||
|
||||
def test_co60_beta_minus(self):
|
||||
info = get_decay_info_from_paceENSDF("Co60", "beta_minus")
|
||||
assert isinstance(info, DecayInfo)
|
||||
assert info.parent_symbol == "Co60"
|
||||
assert info.Z_parent == 27
|
||||
assert info.Z_daughter == 28
|
||||
assert info.A_number == 60
|
||||
assert info.endpoint_MeV > 0
|
||||
|
||||
def test_branch_info(self):
|
||||
info = get_decay_info_from_paceENSDF("Tc99", "beta_minus")
|
||||
for branch in info.branches:
|
||||
assert isinstance(branch, BranchInfo)
|
||||
assert branch.level_index >= 0
|
||||
assert branch.level_energy_keV >= 0
|
||||
assert 0 < branch.intensity <= 100
|
||||
if branch.log_ft is not None:
|
||||
assert branch.log_ft > 0
|
||||
|
||||
def test_invalid_nuclide(self):
|
||||
with pytest.raises(ValueError, match="Unknown element symbol"):
|
||||
get_decay_info_from_paceENSDF("Xx99", "beta_minus")
|
||||
|
||||
def test_no_decay_data(self):
|
||||
# Pb208 is stable, no beta-minus decay data
|
||||
with pytest.raises(ValueError, match="No BM decay data found"):
|
||||
get_decay_info_from_paceENSDF("Pb208", "beta_minus")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# create_config_from_source
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCreateConfigFromSource:
|
||||
"""Test the unified config creation interface."""
|
||||
|
||||
def _write_json(self, data: dict) -> Path:
|
||||
tmp = tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False)
|
||||
json.dump(data, tmp)
|
||||
tmp.close()
|
||||
return Path(tmp.name)
|
||||
|
||||
def test_json_source(self):
|
||||
data = {
|
||||
"Z_parent": 43,
|
||||
"Z_daughter": 44,
|
||||
"A_number": 99,
|
||||
"endpoint_MeV": 0.294,
|
||||
}
|
||||
path = self._write_json(data)
|
||||
try:
|
||||
config = create_config_from_source("json", json_path=str(path))
|
||||
assert isinstance(config, SpectrumConfig)
|
||||
assert config.Z_parent == 43
|
||||
finally:
|
||||
path.unlink()
|
||||
|
||||
def test_unknown_source(self):
|
||||
with pytest.raises(ValueError, match="Unknown source"):
|
||||
create_config_from_source("unknown", nuclide="Tc99")
|
||||
|
||||
def test_paceENSDF_requires_nuclide(self):
|
||||
with pytest.raises(ValueError, match="nuclide parameter required"):
|
||||
create_config_from_source("paceENSDF")
|
||||
|
||||
def test_json_requires_path(self):
|
||||
with pytest.raises(ValueError, match="json_path parameter required"):
|
||||
create_config_from_source("json")
|
||||
Reference in New Issue
Block a user