From f5b79000a4f751afed074b2e94246a17cccb250a Mon Sep 17 00:00:00 2001 From: Evgeniy Unzhakov Date: Wed, 30 Sep 2026 03:11:05 +0300 Subject: [PATCH] fix(scripts): list min/max calls, typed mains, drop unused header reads --- notebooks/01-beta-spectrum-calculator.ipynb | 19 ++++--- notebooks/exp_calibration.ipynb | 12 +--- scripts/analyze_raw.py | 63 ++++++++++++++------- scripts/docs_check.py | 2 +- 4 files changed, 55 insertions(+), 41 deletions(-) diff --git a/notebooks/01-beta-spectrum-calculator.ipynb b/notebooks/01-beta-spectrum-calculator.ipynb index 3a4f34c..bf64cb9 100644 --- a/notebooks/01-beta-spectrum-calculator.ipynb +++ b/notebooks/01-beta-spectrum-calculator.ipynb @@ -94,8 +94,6 @@ " endpoint_MeV=0.294,\n", " transition_type=\"F2\",\n", " e_step_MeV=0.001,\n", - " use_phase_space=True,\n", - " use_fermi=True,\n", " use_screening=True,\n", " use_finite_size=True,\n", " use_charge_dist=True,\n", @@ -410,13 +408,16 @@ } ], "source": [ + "multi_analyzer = BetaSpectrumAnalyzer(multi_spectrum, multi_config)\n", + "\n", "fig, ax = plt.subplots(figsize=(12, 6))\n", "ax.plot(T_keV, total, \"k-\", linewidth=2.5, label=\"Total spectrum\")\n", "\n", - "branch_spectra = multi_spectrum.get_branch_normalized_spectra(W)\n", - "for i, bs in enumerate(branch_spectra):\n", - " ax.plot(T_keV, bs, alpha=0.6, linewidth=1.5,\n", - " label=\"Branch {} (E0={:.0f} keV)\".format(i+1, branches[i].endpoint_MeV*1000))\n", + "# Physical branch spectra: core (phase_space x fermi) x all enabled corrections\n", + "for i, b in enumerate(branches):\n", + " physical = multi_analyzer.physical_branch_spectrum(i)\n", + " ax.plot(T_keV, physical, alpha=0.6, linewidth=1.5,\n", + " label=\"Branch {} (E0={:.0f} keV)\".format(i+1, b.endpoint_MeV*1000))\n", "\n", "for i, b in enumerate(branches):\n", " br_end_keV = W_to_T(T_to_W(b.endpoint_MeV))\n", @@ -430,7 +431,7 @@ "ax.set_xlim(0, W0_max_keV)\n", "ax.grid(True, alpha=0.3)\n", "plt.tight_layout()\n", - "plt.show()" + "plt.show()\n" ] }, { @@ -541,8 +542,8 @@ } ], "source": [ - "analyzer = BetaSpectrumAnalyzer(multi_spectrum, multi_config)\n", - "analyzer.plot_analysis(show_components=True)" + "analyzer = multi_analyzer\n", + "analyzer.plot_analysis(show_components=True)\n" ] }, { diff --git a/notebooks/exp_calibration.ipynb b/notebooks/exp_calibration.ipynb index 9747328..a87c65a 100644 --- a/notebooks/exp_calibration.ipynb +++ b/notebooks/exp_calibration.ipynb @@ -81,11 +81,7 @@ "\n", "# Detect header size from file\n", "def detect_header_size(filepath):\n", - " \"\"\"Detect MCA header size by checking for binary pattern.\"\"\"\n", - " with open(filepath, 'rb') as f:\n", - " header = f.read(128)\n", - " # Check if first 32 bytes look like a header (small ints)\n", - " # and bytes 32-128 also look like header\n", + " \"\"\"Detect MCA header size from the file size.\"\"\"\n", " # Simple heuristic: if file is exactly 32 + 4096*4 = 16412, header=32\n", " # If file is 128 + 4096*4 = 16512, header=128\n", " import os\n", @@ -97,11 +93,7 @@ " elif size == expected_128:\n", " return 128\n", " else:\n", - " # Try to detect by checking if byte 32 looks like an int\n", - " with open(filepath, 'rb') as f:\n", - " f.seek(32)\n", - " test = f.read(4)\n", - " return 32 # default\n", + " return 32 # default\n", "\n", "# --- Raw run spectrum ---\n", "header_size = detect_header_size(RAW_FILE)\n", diff --git a/scripts/analyze_raw.py b/scripts/analyze_raw.py index 212bc70..d027b29 100644 --- a/scripts/analyze_raw.py +++ b/scripts/analyze_raw.py @@ -17,6 +17,7 @@ from __future__ import annotations import sys from pathlib import Path +from typing import cast import numpy as np from scipy.stats import linregress @@ -101,7 +102,10 @@ def estimate_endpoint_from_counts( # Linear fit: K = a + b * E, endpoint = -a/b try: - slope, intercept, r_val, _, _ = linregress(E_sel, K)[:5] + lin = cast("tuple[float, float, float, float, float]", linregress(E_sel, K)) + slope = lin[0] + intercept = lin[1] + r_val = lin[2] if slope < 0 and abs(r_val) > 0.3: endpoint = -intercept / slope # Clamp to reasonable range (Q ± 100 keV) @@ -129,7 +133,7 @@ def compute_regional_rates(counts: np.ndarray, energies_keV: np.ndarray) -> dict } -def main(): +def main() -> None: """Main analysis pipeline.""" data_dir = Path(__file__).parent.parent / "data" / "raw" output_dir = Path(__file__).parent.parent / "output" @@ -168,8 +172,6 @@ def main(): print("\nSubtracting consecutive runs...") individual_runs = [] for i in range(1, len(cumulative_specs)): - parent_a = cumulative_specs[i].run_id - parent_b = cumulative_specs[i - 1].run_id diff = subtract_spectra( cumulative_a=cumulative_specs[i], cumulative_b=cumulative_specs[i - 1], @@ -215,33 +217,52 @@ def main(): print(f"\n{'=' * 60}") print(f"Analysis of {len(run_ids)} individual Tc-99 runs") print(f"{'=' * 60}") - print(f"Run range: {run_ids.min()} - {run_ids.max()}") - print(f"Total counts: {total_rates.min():.0f} - {total_rates.max():.0f}") - print(f"Low-energy fraction: {low_fractions.min():.3f} - {low_fractions.max():.3f}") - print(f"Endpoint estimate: {endpoints.min():.1f} - {endpoints.max():.1f} keV") + print(f"Run range: {min(run_ids)} - {max(run_ids)}") + print(f"Total counts: {min(total_rates):.0f} - {max(total_rates):.0f}") + print( + f"Low-energy fraction: {min(low_fractions):.3f} " f"- {max(low_fractions):.3f}" + ) + print(f"Endpoint estimate: {min(endpoints):.1f} - {max(endpoints):.1f} keV") # Endpoint drift if len(run_ids) > 10: - slope, intercept, r_val, _, _ = linregress(run_ids, endpoints) - print(f"\nEndpoint drift: {slope:.3f} keV/run (R²={r_val**2:.4f})") - print(f" Start (run 1): {intercept + slope * run_ids[0]:.1f} keV") - print(f" End (run {run_ids[-1]}): {intercept + slope * run_ids[-1]:.1f} keV") + lin_e = cast( + "tuple[float, float, float, float, float]", + linregress(run_ids, endpoints), + ) + slope_e = lin_e[0] + intercept_e = lin_e[1] + r_e = lin_e[2] + print(f"\nEndpoint drift: {slope_e:.3f} keV/run (R²={r_e ** 2:.4f})") + print(f" Start (run 1): {intercept_e + slope_e * run_ids[0]:.1f} keV") + print( + f" End (run {run_ids[-1]}): {intercept_e + slope_e * run_ids[-1]:.1f} keV" + ) # Rate drift - slope, intercept, r_val, _, _ = linregress(run_ids, total_rates) - print(f"\nTotal count rate drift: {slope:.0f} counts/run (R²={r_val**2:.4f})") + lin_r = cast( + "tuple[float, float, float, float, float]", linregress(run_ids, total_rates) + ) + slope_r = lin_r[0] + r_r = lin_r[2] + print(f"\nTotal count rate drift: {slope_r:.0f} counts/run (R²={r_r ** 2:.4f})") # Save results results_file = output_dir / "rate_analysis.csv" - with open(results_file, "w") as f: - f.write( - "run,total_counts,low_energy_counts,high_energy_counts,low_fraction,endpoint_keV\n" - ) - for i in range(len(run_ids)): + try: + with open(results_file, "w") as f: f.write( - f"{run_ids[i]},{total_rates[i]:.0f},{low_rates[i]:.0f}," - f"{high_rates[i]:.0f},{low_fractions[i]:.4f},{endpoints[i]:.1f}\n" + "run,total_counts,low_energy_counts,high_energy_counts," + "low_fraction,endpoint_keV\n" ) + for i in range(len(run_ids)): + f.write( + f"{run_ids[i]},{total_rates[i]:.0f},{low_rates[i]:.0f}," + f"{high_rates[i]:.0f},{low_fractions[i]:.4f},{endpoints[i]:.1f}\n" + ) + except OSError as exc: + print(f"Cannot write {results_file}: {exc}") + raise SystemExit(1) from exc print(f"\nResults saved to {results_file}") # Plot diff --git a/scripts/docs_check.py b/scripts/docs_check.py index 71f7eba..b90ba31 100644 --- a/scripts/docs_check.py +++ b/scripts/docs_check.py @@ -471,7 +471,7 @@ CHECKS = [ ] -def main(): +def main() -> int: parser = argparse.ArgumentParser(description="Check documentation consistency") parser.add_argument( "--strict", action="store_true", help="Treat warnings as errors (exit 1)"