haniwers.v1.threshold.parallel#
Parallel threshold scanning functionality.
This module provides core functions for parallel threshold scanning of multiple detector channels with coordinated stepping. All channels step together at each measurement point, with all active during collection (unlike serial mode).
Functions: run_parallel_threshold_scan: Main orchestrator for parallel threshold scanning _generate_threshold_values: Compute threshold for each channel at current step _apply_threshold_with_retry: Write threshold with retry logic
Example: >>> from haniwers.v1.config.loader import ConfigLoader >>> from haniwers.v1.threshold.parallel import run_parallel_threshold_scan >>> cfg = ConfigLoader(“config.toml”).config >>> device = Device(cfg.device) >>> device.connect() >>> results = run_parallel_threshold_scan(cfg, device) >>> device.disconnect()
Module Contents#
Functions#
Append scan data to CSV file in append mode. |
|
Log all threshold changes for reproducibility and debugging. |
|
Compute threshold values for all channels at current step. |
|
Write threshold to device with retry logic. |
|
Run parallel threshold scanning for detector characterization. |
Data#
API#
- haniwers.v1.threshold.parallel.__all__#
[‘ParallelScanResult’, ‘run_parallel_threshold_scan’]
- haniwers.v1.threshold.parallel.append_scan_results(channel: int, data_points: List[haniwers.v1.threshold.model.ScanDataPoint], output_dir: pathlib.Path) str#
Append scan data to CSV file in append mode.
Appends threshold scan results for one channel to an existing CSV file, or creates a new file if it doesn’t exist. This preserves historical data across multiple scan runs.
The file uses append mode (‘a’) to add new rows without removing existing data. The header is written only if the file doesn’t exist or is empty.
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 or append to CSV file in.
Returns: Full file path (str) to the CSV file.
Example: >>> data = [ … ScanDataPoint(timestamp=“2025-10-27T10:23:45.123Z”, channel=1, threshold=250, event_count=1234), … ] >>> path = append_scan_results(1, data, Path(“data/20251027”)) >>> print(path) data/20251027/scan_results_ch1.csv
- haniwers.v1.threshold.parallel.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
- haniwers.v1.threshold.parallel._generate_threshold_values(sensor_configs: Dict[int, haniwers.v1.config.model.SensorConfig], step: int) Dict[int, int]#
Compute threshold values for all channels at current step.
Calculates the threshold value for each channel based on: threshold = center + (step * step_size)
Then clips each value to the valid range [1, 1023].
Args: sensor_configs: Dict mapping channel ID to SensorConfig. step: Current step index (0…nsteps-1).
Returns: Dict mapping channel ID to computed threshold value.
Example: >>> configs = { … 1: SensorConfig(id=1, center=512, step_size=5, nsteps=8), … 2: SensorConfig(id=2, center=512, step_size=5, nsteps=8), … } >>> thresholds = _generate_threshold_values(configs, step=2) >>> print(thresholds) {1: 522, 2: 522} # center + (2 * step_size) = 512 + 10 = 522
- haniwers.v1.threshold.parallel._apply_threshold_with_retry(device: Any, channel: int, threshold: int, max_retry: int = 3) bool#
Write threshold to device with retry logic.
Attempts to write a threshold value to a detector channel with automatic retries on failure. Logs each attempt and final status.
Args: device: Connected Device or RandomMocker instance. channel: Detector channel number (1-3). threshold: Threshold value to write (1-1023). max_retry: Maximum number of retry attempts. Defaults to 3.
Returns: True if threshold write succeeded, False if all retries exhausted.
Example: >>> device = Device(config.device) >>> device.connect() >>> success = _apply_threshold_with_retry(device, channel=1, threshold=250, max_retry=3) >>> print(f"Write success: {success}")
- haniwers.v1.threshold.parallel.run_parallel_threshold_scan(config: haniwers.v1.config.model.HaniwersConfig, device: Any, max_retry: int = 3) haniwers.v1.threshold.model.ParallelScanResult#
Run parallel threshold scanning for detector characterization.
Scans all detector channels across a range of threshold values in parallel. All channels step together at each measurement point, with all active during data collection (unlike serial mode).
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.
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: ParallelScanResult: 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 - skipped_steps (list[int]): Step indices skipped due to failures
Example: >>> config = HaniwersConfig.from_toml(Path(“config.toml”)) >>> device = Device(config.device) >>> device.connect() >>> results = run_parallel_threshold_scan(config, device) >>> device.disconnect() >>> print(results.success) # True if scan completed