haniwers.v1.threshold.serial#
Serial threshold scanning functionality.
This module provides core functions for scanning detector thresholds and collecting cosmic ray event data. It supports single-channel and multi-channel scanning with full reproducibility through audit logging.
Classes: ScanDataPoint: Immutable data point representing a single threshold scan measurement ThresholdOperation: Immutable record of a threshold write operation for audit logging ChannelScanResult: Immutable result from scanning a single detector channel ScanResult: Immutable result from complete serial threshold scanning operation
Functions: run_serial_threshold_scan: Main orchestrator for threshold scanning scan_single_channel: Scan one channel across all threshold values write_scan_results: Save scan data to CSV file write_audit_log: Log all threshold operations for reproducibility
Example: >>> from haniwers.v1.config.loader import ConfigLoader >>> from haniwers.v1.threshold.serial import run_serial_threshold_scan >>> cfg = ConfigLoader(“config.toml”).config >>> device = Device(cfg.device) >>> device.connect() >>> results = run_serial_threshold_scan(cfg, device) >>> device.disconnect()
Module Contents#
Functions#
Run serial threshold scanning for detector characterization. |
|
Scan one channel across all threshold values. |
|
Collect cosmic ray events from device for specified duration. |
|
Save scan data to properly formatted CSV file. |
|
Log all threshold changes for reproducibility and debugging. |
Data#
API#
- haniwers.v1.threshold.serial.__all__#
[‘ScanDataPoint’, ‘ThresholdOperation’, ‘ChannelScanResult’, ‘ScanResult’, 'run_serial_threshold_sca…
- haniwers.v1.threshold.serial.run_serial_threshold_scan(config: haniwers.v1.config.model.HaniwersConfig, device: Any, max_retry: int = 3) haniwers.v1.threshold.model.ScanResult#
Run serial threshold scanning for detector characterization.
Scans each detector channel across a range of threshold values, collecting cosmic ray event data at each threshold level. The device must already be connected before calling this function.
Uses config.sampler.workspace as the output directory for CSV files. The caller (CLI handler) is responsible for setting up the workspace path and creating necessary directories.
Args: config: HaniwersConfig object with sensor and sampler settings. device: Connected Device or RandomMocker instance. max_retry: Number of retries for device communication errors. Defaults to 3.
Returns: ScanResult: Immutable result object with: - success (bool): Whether scan completed without fatal errors - scanned_channels (list[int]): Channel IDs that were scanned - files (list[str]): Full paths to output CSV files created - errors (list[str]): Error messages encountered during scanning
Example: >>> config = HaniwersConfig.from_toml(Path(“config.toml”)) >>> device = Device(config.device) >>> device.connect() >>> results = run_serial_threshold_scan(config, device) >>> device.disconnect() >>> print(results.success) # True if scan completed
- haniwers.v1.threshold.serial.scan_single_channel(sensor_config: haniwers.v1.config.model.SensorConfig, device: Any, project_config: haniwers.v1.config.model.HaniwersConfig, output_dir: pathlib.Path, max_retry: int = 3, operations_log: Optional[List[haniwers.v1.threshold.model.ThresholdOperation]] = None) haniwers.v1.threshold.model.ChannelScanResult#
Scan one channel across all threshold values.
For a single detector channel, sets the channel to each threshold value in the configured range, collects event data for the specified duration, and saves results to CSV. Non-active channels are suppressed (set to 1023).
Args: sensor_config: SensorConfig for the channel being scanned. device: Connected Device or RandomMocker instance. project_config: HaniwersConfig with sampler settings (duration, etc). output_dir: Directory to write output CSV files. max_retry: Number of retry attempts for device communication. Defaults to 3. operations_log: Optional list to accumulate ThresholdOperation records for audit.
Returns: ChannelScanResult: Immutable result with: - channel (int): Channel number scanned - files (list[str]): Full paths to CSV files created - error (str): Error message if scan failed, None otherwise - data_points (list[ScanDataPoint]): Raw data collected as ScanDataPoint objects
Example: >>> sensor = config.sensors[“ch1”] >>> results = scan_single_channel(sensor, device, config, Path(“data/20251027”)) >>> print(results.files[0]) # Path to scan_results_ch1.csv
- haniwers.v1.threshold.serial.collect_events(device: Any, duration: float) int#
Collect cosmic ray events from device for specified duration.
Reads data from the device for the given time period and counts the number of cosmic ray events detected. For mock devices, returns simulated event counts. For real devices, integrates with the device’s event reading interface.
Args: device: Connected Device or RandomMocker instance. duration: Time in seconds to collect data.
Returns: Number of cosmic ray events detected during collection period.
Example: >>> event_count = collect_events(device, duration=2.0) >>> print(event_count) # e.g., 1234
- haniwers.v1.threshold.serial.write_scan_results(channel: int, data_points: List[haniwers.v1.threshold.model.ScanDataPoint], output_dir: pathlib.Path) str#
Save scan data to properly formatted CSV file.
Creates a CSV file with threshold scan results for one channel. Each row contains: timestamp, channel, threshold value, and event count. The file is suitable for import into pandas, Excel, or other analysis tools.
Column Order: timestamp: ISO format timestamp of measurement (YYYY-MM-DDTHH:mm:ss.sssZ) channel: Detector channel number (1-3) threshold: Threshold value used for measurement (1-1023) event_count: Number of cosmic ray events detected
Args: channel: Channel number (1-3) for OSECHI detector. data_points: List of ScanDataPoint objects with measurement data. output_dir: Directory to create CSV file in.
Returns: Full file path (str) to the created CSV file.
Example: >>> data = [ … ScanDataPoint(timestamp=“2025-10-27T10:23:45.123Z”, channel=1, threshold=250, event_count=1234), … ScanDataPoint(timestamp=“2025-10-27T10:23:47.456Z”, channel=1, threshold=240, event_count=1189), … ] >>> path = write_scan_results(1, data, Path(“data/20251027”)) >>> print(path) data/20251027/scan_results_ch1.csv
- haniwers.v1.threshold.serial.write_audit_log(operations: List[haniwers.v1.threshold.model.ThresholdOperation], output_dir: pathlib.Path) str#
Log all threshold changes for reproducibility and debugging.
Creates an audit log CSV file recording every threshold operation executed during scanning. Each row contains: timestamp, channel, old threshold value, new threshold value, and success status.
This enables full traceability of detector configuration changes and helps debug measurement issues.
Args: operations: List of ThresholdOperation objects with: - timestamp (str): ISO format timestamp - channel (int): Detector channel (1-3) - old_threshold (int): Previous threshold value - new_threshold (int): New threshold value - status (str): “success” or error message output_dir: Directory to create audit log in.
Returns: Full file path (str) to the created audit log file.
Example: >>> ops = [ … ThresholdOperation( … timestamp=“2025-10-27T10:23:45.100Z”, … channel=1, … old_threshold=1023, … new_threshold=250, … status=“success”, … ), … ] >>> path = write_audit_log(ops, Path(“data/20251027”)) >>> print(path) data/20251027/threshold_operations.csv