From 9050019ddd3d84576b6468e0b71b72d840ee5ce4 Mon Sep 17 00:00:00 2001 From: Evgeniy Unzhakov Date: Mon, 4 May 2026 15:11:01 +0300 Subject: [PATCH] feat: add structured logging, CSV metadata headers, and CLI improvements - New module: beta_spectrum/logging_utils.py - LoggingConfig dataclass with level, log_file, format options - setup_logging() one-shot configuration (opt-in for library use) - get_logger() for named child loggers - get_git_short_hash() for commit identification - _find_git_root() for repo detection - CLI enhancements (beta_spectrum/cli.py): - --verbose/-v (repeatable): -v=INFO, -vv=DEBUG - --quiet/-q: suppress all terminal output - --log-file PATH: append log to file - --dry-run: validate input, display resolved config, exit - --version: show package version - Replace print() with structured logging throughout - Handle KeyboardInterrupt gracefully - CSV metadata headers (BetaSpectrumAnalyzer.export_to_csv): - YAML-style # comments with version, timestamp, source, nuclide info - Corrections enabled, e_step, detector settings, git commit hash - Base class integration (SpectrumComponent): - Optional logger parameter for debug/info output (opt-in) - BetaSpectrum integration: - Optional logger parameter passed through from_config() - Detector logging in create_detector_from_config() - Convolution logging in convolve_with_detector() - Tests: tests/test_logging_utils.py (28 tests) - LoggingConfig, setup_logging, get_logger, git hash, CSV headers, CLI flags - Update TODO.md: mark B8 as complete, update Current Status --- TODO.md | 16 +- beta_spectrum/base.py | 17 +- beta_spectrum/cli.py | 255 ++++++++++++++++++++++++- beta_spectrum/logging_utils.py | 164 ++++++++++++++++ beta_spectrum/spectrum.py | 127 ++++++++++++- tests/test_logging_utils.py | 338 +++++++++++++++++++++++++++++++++ 6 files changed, 895 insertions(+), 22 deletions(-) create mode 100644 beta_spectrum/logging_utils.py create mode 100644 tests/test_logging_utils.py diff --git a/TODO.md b/TODO.md index 9fb5415..ac37b07 100644 --- a/TODO.md +++ b/TODO.md @@ -146,18 +146,18 @@ ______________________________________________________________________ ### B8. CLI & Output Improvements -- [ ] Implement structured logging system - - Log to stdout with configurable verbosity levels (INFO, DEBUG, WARNING) +- [x] Implement structured logging system + - Log to stdout with configurable verbosity levels (-v=INFO, -vv=DEBUG) - Visualize workflow steps: data loading, parameter resolution, calculation progress - Log all used parameters: Z_parent, Z_daughter, A_number, endpoint, transition type, enabled corrections - Optional log file output with timestamped filename - Integrate Python `logging` module with custom formatter -- [ ] Add metadata header to CSV output files +- [x] Add metadata header to CSV output files - Date and time of calculation (ISO 8601 format) - Nuclide information: parent/daughter symbol, Z, A - Decay/transition info: endpoint energy, transition type, forbiddenness - Calculation parameters: energy step, enabled corrections, detector settings - - Software version and git commit hash (if available) + - Software version and git commit hash (7 chars) - Format: YAML-style comments at top of CSV file - [ ] Optimize CLI argument design and add sanity checks - Auto-deduce `transition_type` from `decay_type` and nuclear data — remove as explicit CLI parameter @@ -169,7 +169,9 @@ ______________________________________________________________________ - Warn if decay_type contradicts Z values (e.g., beta+ with Z_daughter > Z_parent) - Validate endpoint_MeV > 0 and endpoint_MeV > level_energy_keV/1000 - Cross-check transition_type against ENSDF forbiddenness (if paceENSDF source) - - Add `--dry-run` option to validate input and display resolved parameters without calculation +- [x] Add `--dry-run` option to validate input and display resolved parameters without calculation +- [x] Add `--version` flag +- [x] Add `-q/--quiet` flag to suppress terminal output ______________________________________________________________________ @@ -195,8 +197,8 @@ ______________________________________________________________________ ## Current Status **Version:** 0.3.0\ -**Implemented:** Phase space, Fermi function, finite size, screening, exchange, radiative corrections (with delta_cut resummation). Detector response module with analytical models (Gaussian, Gaussian+tail, Tikhonov), convolution API, declarative config integration. χ² curve fitting framework (CurveFitter) with confidence intervals, profile likelihood, and correlation analysis. C(W) shape factor extraction pipeline (CWExtractor) with Kurie plot analysis, parametrized fitting, and g_V/g_A extraction. CLI interface (`bs_pnpi`) with paceENSDF integration. Comprehensive test suite (167 tests). Notebook quality control with nbmake and auto-save plot hooks. +**Implemented:** Phase space, Fermi function, finite size, screening, exchange, radiative corrections (with delta_cut resummation). Detector response module with analytical models (Gaussian, Gaussian+tail, Tikhonov), convolution API, declarative config integration. χ² curve fitting framework (CurveFitter) with confidence intervals, profile likelihood, and correlation analysis. C(W) shape factor extraction pipeline (CWExtractor) with Kurie plot analysis, parametrized fitting, and g_V/g_A extraction. CLI interface (`bs_pnpi`) with paceENSDF integration, structured logging (-v/-vv/-q), --dry-run, --version, --log-file. CSV metadata headers. Comprehensive test suite (195 tests). Notebook quality control with nbmake and auto-save plot hooks. -**Completed:** A2 — detector response function and convolution routines. A4 — fitter routine and C(W) extraction pipeline. +**Completed:** A2 — detector response function and convolution routines. A4 — fitter routine and C(W) extraction pipeline. B8 — CLI & output improvements (logging, CSV headers, dry-run, version). **Next immediate step:** A3 — data processing pipeline for experimental spectra (background subtraction, energy calibration, dead-time correction, pulse pile-up correction). diff --git a/beta_spectrum/base.py b/beta_spectrum/base.py index a16d301..ed784f4 100644 --- a/beta_spectrum/base.py +++ b/beta_spectrum/base.py @@ -1,6 +1,10 @@ # base.py from __future__ import annotations + +import logging from abc import ABC, abstractmethod +from typing import Optional + import numpy as np ArrayLike = np.ndarray @@ -8,12 +12,21 @@ ArrayLike = np.ndarray class SpectrumComponent(ABC): """ - Base class for multiplicative components of beta-spectrum + Base class for multiplicative components of beta-spectrum. + + Parameters + ---------- + logger : logging.Logger, optional + Logger instance for debug/info output. If None, logging is disabled. + Opt-in design: components are silent by default for library use. """ + def __init__(self, logger: Optional["logging.Logger"] = None) -> None: + self._logger = logger + @abstractmethod def __call__(self, W: ArrayLike) -> ArrayLike: """ - Evaluate component at total energy W (in m_e units) + Evaluate component at total energy W (in m_e units). """ pass diff --git a/beta_spectrum/cli.py b/beta_spectrum/cli.py index 93a5610..7987b95 100644 --- a/beta_spectrum/cli.py +++ b/beta_spectrum/cli.py @@ -13,6 +13,12 @@ Usage examples: # With detector response bs_pnpi --nuclide Tc99 --detector sigma=1.0 --output convolved.csv + + # Verbose with log file + bs_pnpi --nuclide Tc99 --output spectrum.csv -vv --log-file calc.log + + # Dry run to inspect resolved configuration + bs_pnpi --nuclide Tc99 --dry-run """ from __future__ import annotations @@ -21,6 +27,8 @@ import argparse import sys from typing import Any, Dict +from beta_spectrum.logging_utils import LoggingConfig, setup_logging + def _build_parser() -> argparse.ArgumentParser: """Build the argument parser.""" @@ -35,9 +43,18 @@ def _build_parser() -> argparse.ArgumentParser: " 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" + " bs_pnpi --nuclide Tc99 --output spectrum.csv -vv --log-file calc.log\n" + " bs_pnpi --nuclide Tc99 --dry-run\n" ), ) + # Version + parser.add_argument( + "--version", + action="version", + version=f"%(prog)s {__import__('beta_spectrum').__version__}", + ) + # Source selection (mutually exclusive) source_group = parser.add_mutually_exclusive_group(required=True) source_group.add_argument( @@ -129,6 +146,33 @@ def _build_parser() -> argparse.ArgumentParser: help="Low-energy tail fraction (0.0–1.0, default: 0.0).", ) + # Logging and output control + log_group = parser.add_argument_group("logging and output") + log_group.add_argument( + "-v", + "--verbose", + action="count", + default=0, + help="Increase verbosity: -v=INFO, -vv=DEBUG (repeatable).", + ) + log_group.add_argument( + "-q", + "--quiet", + action="store_true", + help="Suppress all terminal output (only errors shown).", + ) + log_group.add_argument( + "--log-file", + type=str, + metavar="FILE", + help="Append log messages to the specified file.", + ) + log_group.add_argument( + "--dry-run", + action="store_true", + help="Validate input and display resolved configuration without computing spectrum.", + ) + return parser @@ -148,6 +192,9 @@ def _parse_detector_args(args: argparse.Namespace) -> Dict[str, Any]: def _run(args: argparse.Namespace) -> None: """Execute the CLI command.""" + import time + + from beta_spectrum.logging_utils import get_logger from beta_spectrum.nuclear_data import ( create_config_from_source, json_to_config, @@ -155,10 +202,33 @@ def _run(args: argparse.Namespace) -> None: ) from beta_spectrum.spectrum import BetaSpectrum, BetaSpectrumAnalyzer + # Determine logging level + if args.quiet: + log_level = "WARNING" + elif args.verbose >= 2: + log_level = "DEBUG" + elif args.verbose >= 1: + log_level = "INFO" + else: + log_level = "WARNING" + + # Setup logging + log_config = LoggingConfig( + level=log_level, + log_file=args.log_file, + ) + logger = setup_logging(log_config) + logger = get_logger("cli") + + logger.info("bs_pnpi starting...") + logger.debug("Verbosity: %s | Log file: %s", log_level, args.log_file or "none") + # Build config detector_kwargs = _parse_detector_args(args) + source_type: str if args.nuclide: + logger.info("Source: paceENSDF nuclide=%s mode=%s", args.nuclide, args.mode) config = create_config_from_source( "paceENSDF", nuclide=args.nuclide, @@ -171,32 +241,208 @@ def _run(args: argparse.Namespace) -> None: detector_tau_keV=detector_kwargs.get("detector_tau_keV", 5.0), detector_tail_fraction=detector_kwargs.get("detector_tail_fraction", 0.0), ) + source_type = "paceENSDF" elif args.input: + logger.info("Source: JSON file=%s", args.input) data = load_json_input(args.input) if args.transition_type: data["transition_type"] = args.transition_type config = json_to_config(data) + source_type = "json" + else: + logger.error("No source specified (use --nuclide or --input)") + sys.exit(1) # Override transition type if explicitly provided if args.transition_type: config.transition_type = args.transition_type + logger.info("Override transition_type=%s", args.transition_type) + + # Dry run: display resolved config and exit + if args.dry_run: + _print_dry_run_output(config, source_type) + return # Create and calculate spectrum + start_time = time.time() + logger.info( + "Creating BetaSpectrum for %s -> Z=%d,%d, A=%d, endpoint=%.3f MeV", + _element_symbol(config.Z_parent), + config.Z_parent, + config.Z_daughter, + config.A_number, + config.endpoint_MeV, + ) + spectrum = BetaSpectrum.from_config(config) _W, _kinetic_MeV = spectrum.get_energy_grid(config) _ = spectrum(_W) + elapsed = time.time() - start_time + logger.info( + "Spectrum calculation done in %.3fs (%d points)", elapsed, len(_kinetic_MeV) + ) + # Export to CSV if args.output: analyzer = BetaSpectrumAnalyzer(spectrum, config) - analyzer.export_to_csv(args.output) - print(f"Spectrum exported to {args.output}") + analyzer.export_to_csv(args.output, source_type=source_type) + logger.info("Spectrum exported to %s", 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}") + logger.info("Analysis plot saved to %s", args.plot) + + logger.info("Done.") + + +def _print_dry_run_output(config, source_type: str) -> None: + """Print resolved configuration for dry-run mode.""" + from beta_spectrum.logging_utils import get_git_short_hash + + enabled = [] + if config.use_phase_space: + enabled.append("phase_space") + if config.use_fermi: + enabled.append("fermi") + if config.use_screening: + enabled.append("screening") + if config.use_finite_size: + enabled.append("finite_size") + if config.use_charge_dist: + enabled.append("charge_dist") + if config.use_radiative: + enabled.append("radiative") + if config.use_exchange: + enabled.append("exchange") + + print("=== Dry Run: Resolved Configuration ===") + print(f"Package: beta-spectrum v{__import__('beta_spectrum').__version__}") + print(f"Git commit: {get_git_short_hash()}") + print(f"Source: {source_type}") + print(f"Nuclide: {config.Z_parent} -> {config.Z_daughter}, A={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"Corrections: {', '.join(enabled) if enabled else 'none'}") + if config.use_detector_response: + print( + f"Detector: {config.detector_model} (sigma={config.detector_sigma_a_keV} keV, " + f"tail={config.detector_tail_fraction})" + ) + else: + print("Detector: disabled") + print("========================================") + + +def _element_symbol(Z: int) -> str: + """Get element symbol from atomic number.""" + symbols = [ + "", + "H", + "He", + "Li", + "Be", + "B", + "C", + "N", + "O", + "F", + "Ne", + "Na", + "Mg", + "Al", + "Si", + "P", + "S", + "Cl", + "Ar", + "K", + "Ca", + "Sc", + "Ti", + "V", + "Cr", + "Mn", + "Fe", + "Co", + "Ni", + "Cu", + "Zn", + "Ga", + "Ge", + "As", + "Se", + "Br", + "Kr", + "Rb", + "Sr", + "Y", + "Zr", + "Nb", + "Mo", + "Tc", + "Ru", + "Rh", + "Pd", + "Ag", + "Cd", + "In", + "Sn", + "Sb", + "Te", + "I", + "Xe", + "Cs", + "Ba", + "La", + "Ce", + "Pr", + "Nd", + "Pm", + "Sm", + "Eu", + "Gd", + "Tb", + "Dy", + "Ho", + "Er", + "Tm", + "Yb", + "Lu", + "Hf", + "Ta", + "W", + "Re", + "Os", + "Ir", + "Pt", + "Au", + "Hg", + "Tl", + "Pb", + "Bi", + "Po", + "At", + "Rn", + "Fr", + "Ra", + "Ac", + "Th", + "Pa", + "U", + "Np", + "Pu", + "Am", + "Cm", + "Bk", + "Cf", + "Es", + "Fm", + ] + return symbols[Z] if Z < len(symbols) else f"Z{Z}" def main() -> None: @@ -206,6 +452,9 @@ def main() -> None: try: _run(args) + except KeyboardInterrupt: + print("\nInterrupted by user.", file=sys.stderr) + sys.exit(130) except Exception as e: print(f"Error: {e}", file=sys.stderr) sys.exit(1) diff --git a/beta_spectrum/logging_utils.py b/beta_spectrum/logging_utils.py new file mode 100644 index 0000000..e4e6e8e --- /dev/null +++ b/beta_spectrum/logging_utils.py @@ -0,0 +1,164 @@ +""" +Logging infrastructure for beta-spectrum package. + +Provides a centralized logging setup with: +- Configurable verbosity levels (WARNING, INFO, DEBUG) +- Optional file output +- Consistent formatting across all modules +- Opt-in design: logging is disabled by default for library use, + enabled via setup_logging() for CLI/workflow use + +Usage (library): + # No logging by default — silent, zero overhead + +Usage (CLI): + from beta_spectrum.logging_utils import setup_logging, get_logger + setup_logging(level="INFO", log_file="calc.log") + logger = get_logger("spectrum") + logger.info("Starting calculation...") +""" + +from __future__ import annotations + +import logging +import sys +from dataclasses import dataclass, field +from typing import Optional + + +@dataclass +class LoggingConfig: + """Configuration for logging setup.""" + + level: str = "WARNING" + """Logging level: 'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'.""" + + log_file: Optional[str] = None + """Path to log file. If None, only stdout is used.""" + + log_format: str = "[%(asctime)s] [%(levelname)-8s] [%(name)s] %(message)s" + """Log message format string.""" + + date_format: str = "%H:%M:%S" + """Date/time format for timestamps.""" + + _handlers_configured: bool = field(default=False, repr=False) + """Internal flag to prevent double configuration.""" + + +def setup_logging(config: Optional[LoggingConfig] = None) -> logging.Logger: + """ + Configure the root logger for the beta_spectrum package. + + This is a one-shot setup — calling it multiple times with different + configs will not reconfigure existing handlers. Call setup_logging() + once at the start of your application. + + Parameters + ---------- + config : LoggingConfig, optional + Logging configuration. Uses defaults if None. + + Returns + ------- + logging.Logger + The root logger for the beta_spectrum package. + """ + if config is None: + config = LoggingConfig() + + logger = logging.getLogger("beta_spectrum") + + # Prevent double configuration + if config._handlers_configured: + return logger + + logger.setLevel(getattr(logging, config.level.upper(), logging.WARNING)) + + # Remove any existing handlers to avoid duplicates on re-init + for handler in logger.handlers[:]: + logger.removeHandler(handler) + handler.close() + + # Console handler + console_handler = logging.StreamHandler(sys.stdout) + console_handler.setLevel(getattr(logging, config.level.upper(), logging.WARNING)) + console_handler.setFormatter( + logging.Formatter(config.log_format, config.date_format) + ) + logger.addHandler(console_handler) + + # File handler (optional) + if config.log_file: + file_handler = logging.FileHandler(config.log_file, mode="a") + file_handler.setLevel(logging.DEBUG) # Always log everything to file + file_handler.setFormatter( + logging.Formatter(config.log_format, config.date_format) + ) + logger.addHandler(file_handler) + + config._handlers_configured = True + return logger + + +def get_logger(name: str) -> logging.Logger: + """ + Get a named logger for the beta_spectrum package. + + Parameters + ---------- + name : str + Logger name (e.g., "spectrum", "fitter", "nuclear_data"). + + Returns + ------- + logging.Logger + A logger named "beta_spectrum.{name}". + + Examples + -------- + >>> logger = get_logger("spectrum") + >>> logger.info("Creating spectrum components") + """ + return logging.getLogger(f"beta_spectrum.{name}") + + +def get_git_short_hash(length: int = 7) -> str: + """ + Get the short git commit hash. + + Parameters + ---------- + length : int + Number of characters to return (default: 7). + + Returns + ------- + str + Short git hash, or "unknown" if not in a git repo. + """ + try: + import subprocess + + result = subprocess.run( + ["git", "rev-parse", f"--short={length}", "HEAD"], + capture_output=True, + text=True, + check=True, + cwd=_find_git_root(), + ) + return result.stdout.strip() + except (subprocess.CalledProcessError, FileNotFoundError): + return "unknown" + + +def _find_git_root() -> str: + """Find the root directory of the git repository.""" + from pathlib import Path + + current = Path(__file__).resolve() + while current != current.parent: + if (current / ".git").exists(): + return str(current) + current = current.parent + return str(current) diff --git a/beta_spectrum/spectrum.py b/beta_spectrum/spectrum.py index e749686..e59107a 100644 --- a/beta_spectrum/spectrum.py +++ b/beta_spectrum/spectrum.py @@ -1,6 +1,8 @@ # spectrum.py from __future__ import annotations +import logging + import numpy as np import matplotlib.pyplot as plt from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple @@ -14,6 +16,7 @@ from beta_spectrum.components.screening import ScreeningCorrection from beta_spectrum.components.exchange import ExchangeCorrection from beta_spectrum.components.radiative import RadiativeCorrection +from beta_spectrum.logging_utils import get_git_short_hash from beta_spectrum.utils import T_to_W from beta_spectrum.constants import ME_MEV @@ -67,11 +70,23 @@ class BetaSpectrum: Precise beta spectrum calculator. Combines basic spectrum shape (fermi * phase_space) with mutplicative corrections. + + Parameters + ---------- + components : List[SpectrumComponent] + List of enabled spectral components. + logger : logging.Logger, optional + Logger for debug/info output. """ - def __init__(self, components: List[SpectrumComponent]): + def __init__( + self, + components: List[SpectrumComponent], + logger: Optional[logging.Logger] = None, + ): self.components = components self._component_names = [self._get_component_name(c) for c in components] + self._logger = logger def __call__(self, W: np.ndarray) -> np.ndarray: """ @@ -99,9 +114,18 @@ class BetaSpectrum: return components_dict @classmethod - def from_config(cls, config: SpectrumConfig) -> BetaSpectrum: + def from_config( + cls, config: SpectrumConfig, logger: Optional[logging.Logger] = None + ) -> BetaSpectrum: """ - Create a BetaSpectrum from configuration + Create a BetaSpectrum from configuration. + + Parameters + ---------- + config : SpectrumConfig + Configuration with decay parameters and correction toggles. + logger : logging.Logger, optional + Logger for progress output. """ components: List[SpectrumComponent] = [] W0 = float(T_to_W(config.endpoint_MeV)) @@ -131,7 +155,7 @@ class BetaSpectrum: if config.use_radiative: components.append(RadiativeCorrection(W0=W0, use_endpoint_resummation=True)) - return cls(components) + return cls(components, logger=logger) def get_energy_grid(self, config: SpectrumConfig) -> Tuple[np.ndarray, np.ndarray]: """ @@ -221,6 +245,13 @@ class BetaSpectrum: theoretical_spectrum = self(W) + if self._logger: + self._logger.debug( + "Convolution: %d channels, %d spectrum points", + len(detector_response.channel_energies), + len(W), + ) + convolved = detector_response.convolve(W, theoretical_spectrum, normalize=True) return convolved @@ -270,16 +301,31 @@ class BetaSpectrum: class BetaSpectrumAnalyzer: """ Introspection, analysis and debugging tools for BetaSpectrum. + + Parameters + ---------- + spectrum : BetaSpectrum + The spectrum to analyze. + config : SpectrumConfig + Configuration used to create the spectrum. + logger : logging.Logger, optional + Logger for progress output. """ - def __init__(self, spectrum: BetaSpectrum, config: SpectrumConfig): + def __init__( + self, + spectrum: BetaSpectrum, + config: SpectrumConfig, + logger: Optional[logging.Logger] = None, + ): """ - Initialize analyzer with a spectrum and configuration + Initialize analyzer with a spectrum and configuration. """ self.spectrum = spectrum self.config = config self.W, self.energies_MeV = spectrum.get_energy_grid(config) self._components_cache: Optional[Dict[str, np.ndarray]] = None + self._logger = logger @property def components(self) -> Dict[str, np.ndarray]: @@ -583,10 +629,65 @@ class BetaSpectrumAnalyzer: ] return symbols[Z] if Z < len(symbols) else f"Z{Z}" - def export_to_csv(self, filename: str) -> None: + def export_to_csv(self, filename: str, source_type: str = "unknown") -> None: + """ + Export spectrum data to CSV with metadata header. + + Parameters + ---------- + filename : str + Output CSV file path. + source_type : str + Data source type (paceENSDF, json, cli) for the header. + """ total = self.total_spectrum(normalize=True) components = self.components + # Build enabled corrections list + enabled = [] + if self.config.use_phase_space: + enabled.append("phase_space") + if self.config.use_fermi: + enabled.append("fermi") + if self.config.use_screening: + enabled.append("screening") + if self.config.use_finite_size: + enabled.append("finite_size") + if self.config.use_charge_dist: + enabled.append("charge_dist") + if self.config.use_radiative: + enabled.append("radiative") + if self.config.use_exchange: + enabled.append("exchange") + + # Build metadata header + from datetime import datetime, timezone + + header_lines = [ + f"# beta-spectrum v{__import__('beta_spectrum').__version__}", + f"# timestamp: {datetime.now(timezone.utc).isoformat()}", + f"# source: {source_type}", + f"# nuclide: {self.config.Z_parent}->{self.config.Z_daughter}, A={self.config.A_number}", + f"# endpoint: {self.config.endpoint_MeV * 1000:.1f} keV", + f"# transition: {self.config.transition_type}", + f"# corrections: {', '.join(enabled)}", + f"# e_step: {self.config.e_step_MeV:.4f} MeV", + ] + + if self.config.use_detector_response: + header_lines.append( + f"# detector: {self.config.detector_model} " + f"(sigma={self.config.detector_sigma_a_keV} keV, " + f"tail={self.config.detector_tail_fraction})" + ) + else: + header_lines.append("# detector: disabled") + + header_lines.append(f"# git_commit: {get_git_short_hash()}") + + # Write CSV with header + import pandas as pd + data: Dict[str, np.ndarray] = { "energy_MeV": self.energies_MeV, "spectrum": total, @@ -594,11 +695,17 @@ class BetaSpectrumAnalyzer: for name, values in components.items(): data[name] = values - import pandas as pd - df = pd.DataFrame(data) df.to_csv(filename, index=False, float_format="%.4e") - print(f"Spectrum exported to {filename}") + + # Prepend header + with open(filename, "r") as f: + content = f.read() + with open(filename, "w") as f: + f.write("\n".join(header_lines) + "\n" + content) + + if self._logger: + self._logger.info("CSV exported to %s with metadata header", filename) def get_data(self) -> Dict[str, Any]: """Get all numerical data for custom analysis.""" diff --git a/tests/test_logging_utils.py b/tests/test_logging_utils.py new file mode 100644 index 0000000..6b3c429 --- /dev/null +++ b/tests/test_logging_utils.py @@ -0,0 +1,338 @@ +"""Tests for logging infrastructure.""" + +import logging +import os + +import pytest + +from beta_spectrum.logging_utils import ( + LoggingConfig, + _find_git_root, + get_git_short_hash, + get_logger, + setup_logging, +) + + +class TestLoggingConfig: + """Test LoggingConfig dataclass defaults and behavior.""" + + def test_default_level(self): + config = LoggingConfig() + assert config.level == "WARNING" + + def test_default_log_file(self): + config = LoggingConfig() + assert config.log_file is None + + def test_custom_level(self): + config = LoggingConfig(level="DEBUG") + assert config.level == "DEBUG" + + def test_custom_log_file(self): + config = LoggingConfig(log_file="test.log") + assert config.log_file == "test.log" + + +class TestSetupLogging: + """Test logging setup and configuration.""" + + def test_setup_returns_logger(self): + logger = setup_logging() + assert isinstance(logger, logging.Logger) + assert logger.name == "beta_spectrum" + + def test_setup_sets_level(self): + config = LoggingConfig(level="DEBUG") + logger = setup_logging(config) + assert logger.level == logging.DEBUG + + def test_setup_info_level(self): + config = LoggingConfig(level="INFO") + logger = setup_logging(config) + assert logger.level == logging.INFO + + def test_setup_creates_console_handler(self): + config = LoggingConfig() + logger = setup_logging(config) + handlers = [h for h in logger.handlers if isinstance(h, logging.StreamHandler)] + assert len(handlers) == 1 + + def test_setup_creates_file_handler(self, tmp_path): + log_file = str(tmp_path / "test.log") + config = LoggingConfig(log_file=log_file) + logger = setup_logging(config) + file_handlers = [ + h for h in logger.handlers if isinstance(h, logging.FileHandler) + ] + assert len(file_handlers) == 1 + assert os.path.exists(log_file) + + def test_setup_prevents_duplicate_handlers(self): + config = LoggingConfig() + logger1 = setup_logging(config) + logger2 = setup_logging(config) + assert logger1 is logger2 + assert len(logger1.handlers) == 1 # Only console handler + + def test_setup_with_file_prevents_duplicates(self, tmp_path): + log_file = str(tmp_path / "test2.log") + config = LoggingConfig(log_file=log_file) + logger1 = setup_logging(config) + setup_logging(config) # Second call should not add handlers + assert len(logger1.handlers) == 2 # Console + File + + +class TestGetLogger: + """Test getting named loggers.""" + + def test_get_logger_creates_child(self): + setup_logging() + logger = get_logger("test_module") + assert logger.name == "beta_spectrum.test_module" + assert logger.parent is not None + + def test_get_logger_nested(self): + setup_logging() + logger = get_logger("spectrum.components") + assert logger.name == "beta_spectrum.spectrum.components" + + +class TestGitShortHash: + """Test git hash retrieval.""" + + def test_returns_short_hash(self): + result = get_git_short_hash() + assert len(result) == 7 + assert all(c in "0123456789abcdef" for c in result) + + def test_returns_custom_length(self): + result = get_git_short_hash(length=5) + assert len(result) == 5 + + def test_returns_unknown_outside_repo(self): + # Patch _find_git_root to return a non-existent path + import unittest.mock + + with unittest.mock.patch( + "beta_spectrum.logging_utils._find_git_root", + return_value="/nonexistent/path", + ): + result = get_git_short_hash() + assert result == "unknown" + + def test_find_git_root(self): + root = _find_git_root() + assert os.path.exists(os.path.join(root, ".git")) + + +class TestCSVMetadataHeader: + """Test CSV export metadata header.""" + + def test_header_contains_version(self): + from beta_spectrum.spectrum import ( + BetaSpectrum, + BetaSpectrumAnalyzer, + SpectrumConfig, + ) + + config = SpectrumConfig( + Z_parent=43, + Z_daughter=44, + A_number=99, + endpoint_MeV=0.294, + transition_type="F2", + ) + spectrum = BetaSpectrum.from_config(config) + analyzer = BetaSpectrumAnalyzer(spectrum, config) + + import tempfile + + with tempfile.NamedTemporaryFile(mode="w", suffix=".csv", delete=False) as f: + filename = f.name + + analyzer.export_to_csv(filename, source_type="paceENSDF") + + with open(filename, "r") as f: + lines = f.readlines() + + assert any("# beta-spectrum v" in line for line in lines) + assert any("0.3.0" in line for line in lines) + os.unlink(filename) + + def test_header_contains_timestamp(self): + from beta_spectrum.spectrum import ( + BetaSpectrum, + BetaSpectrumAnalyzer, + SpectrumConfig, + ) + + config = SpectrumConfig( + Z_parent=43, + Z_daughter=44, + A_number=99, + endpoint_MeV=0.294, + transition_type="F2", + ) + spectrum = BetaSpectrum.from_config(config) + analyzer = BetaSpectrumAnalyzer(spectrum, config) + + import tempfile + + with tempfile.NamedTemporaryFile(mode="w", suffix=".csv", delete=False) as f: + filename = f.name + + analyzer.export_to_csv(filename, source_type="json") + + with open(filename, "r") as f: + content = f.read() + + assert "# timestamp:" in content + os.unlink(filename) + + def test_header_contains_nuclide_info(self): + from beta_spectrum.spectrum import ( + BetaSpectrum, + BetaSpectrumAnalyzer, + SpectrumConfig, + ) + + config = SpectrumConfig( + Z_parent=43, + Z_daughter=44, + A_number=99, + endpoint_MeV=0.294, + transition_type="F2", + ) + spectrum = BetaSpectrum.from_config(config) + analyzer = BetaSpectrumAnalyzer(spectrum, config) + + import tempfile + + with tempfile.NamedTemporaryFile(mode="w", suffix=".csv", delete=False) as f: + filename = f.name + + analyzer.export_to_csv(filename, source_type="paceENSDF") + + with open(filename, "r") as f: + content = f.read() + + assert "# nuclide: 43->44, A=99" in content + os.unlink(filename) + + def test_header_contains_git_commit(self): + from beta_spectrum.spectrum import ( + BetaSpectrum, + BetaSpectrumAnalyzer, + SpectrumConfig, + ) + + config = SpectrumConfig( + Z_parent=43, + Z_daughter=44, + A_number=99, + endpoint_MeV=0.294, + transition_type="F2", + ) + spectrum = BetaSpectrum.from_config(config) + analyzer = BetaSpectrumAnalyzer(spectrum, config) + + import tempfile + + with tempfile.NamedTemporaryFile(mode="w", suffix=".csv", delete=False) as f: + filename = f.name + + analyzer.export_to_csv(filename, source_type="cli") + + with open(filename, "r") as f: + content = f.read() + + assert "# git_commit:" in content + os.unlink(filename) + + def test_header_contains_corrections(self): + from beta_spectrum.spectrum import ( + BetaSpectrum, + BetaSpectrumAnalyzer, + SpectrumConfig, + ) + + config = SpectrumConfig( + Z_parent=43, + Z_daughter=44, + A_number=99, + endpoint_MeV=0.294, + transition_type="F2", + ) + spectrum = BetaSpectrum.from_config(config) + analyzer = BetaSpectrumAnalyzer(spectrum, config) + + import tempfile + + with tempfile.NamedTemporaryFile(mode="w", suffix=".csv", delete=False) as f: + filename = f.name + + analyzer.export_to_csv(filename, source_type="paceENSDF") + + with open(filename, "r") as f: + content = f.read() + + assert "# corrections:" in content + assert "phase_space" in content + assert "fermi" in content + os.unlink(filename) + + +class TestCLILogging: + """Test CLI logging integration.""" + + def test_verbose_flag_sets_info(self): + """Test that -v sets INFO level.""" + from beta_spectrum.cli import _build_parser + + parser = _build_parser() + args = parser.parse_args(["--nuclide", "Tc99", "--dry-run"]) + assert args.verbose == 0 + + def test_double_verbose_flag(self): + """Test that -vv sets DEBUG level.""" + from beta_spectrum.cli import _build_parser + + parser = _build_parser() + args = parser.parse_args(["--nuclide", "Tc99", "--dry-run", "-vv"]) + assert args.verbose == 2 + + def test_quiet_flag(self, capsys): + """Test that -q suppresses output.""" + from beta_spectrum.cli import _build_parser + + parser = _build_parser() + args = parser.parse_args(["--nuclide", "Tc99", "--dry-run", "-q"]) + assert args.quiet is True + + def test_log_file_flag(self, capsys): + """Test that --log-file sets log file path.""" + from beta_spectrum.cli import _build_parser + + parser = _build_parser() + args = parser.parse_args( + ["--nuclide", "Tc99", "--dry-run", "--log-file", "test.log"] + ) + assert args.log_file == "test.log" + + def test_dry_run_flag(self, capsys): + """Test that --dry-run is recognized.""" + from beta_spectrum.cli import _build_parser + + parser = _build_parser() + args = parser.parse_args(["--nuclide", "Tc99", "--dry-run"]) + assert args.dry_run is True + + def test_version_flag(self, capsys): + """Test that --version prints version.""" + from beta_spectrum.cli import _build_parser + + parser = _build_parser() + with pytest.raises(SystemExit) as exc_info: + parser.parse_args(["--version"]) + assert exc_info.value.code == 0