haniwers.v1.threshold.writer

Threshold writing functions for OSECHI detector channels.

This module provides functions to write threshold values to the detector, with support for validation, retry logic, CSV integration, and logging.

All functions use the v0-compatible bit manipulation protocol for hardware communication with the OSECHI detector.

Two-Layer Architecture:

**Layer 1: Low-level functions (prefix: underscore)**
- Work directly with Device object and integer values (ch, vth)
- Examples: _write_threshold(), _write_threshold_with_retry()
- For internal use; users rarely call these directly

**Layer 2: High-level functions (prefix: none, start with 'apply')**
- Work with Device + SensorConfig objects
- Examples: apply_threshold(), apply_thresholds()
- Recommended for most users

Public API Summary:

- apply_threshold(): Write single sensor threshold with retry and optional logging
- apply_thresholds(): Write multiple sensor thresholds in batch
- load_thresholds_from_csv(): Load threshold configuration (utility function)
- set_threshold(): Legacy API (kept for backward compatibility)
- set_thresholds_from_csv(): Legacy API (kept for backward compatibility)

Module Contents

Classes

ThresholdResult

Immutable result for threshold write operations.

Functions

write_threshold	Write a threshold value to a specific detector channel.
write_threshold_with_retry	Write threshold value with automatic retry on failure.
log_threshold_operation	Log threshold write operation to CSV file for audit trail.
load_thresholds_from_csv	Load threshold configuration from CSV file.
apply_threshold	Apply threshold configuration to a single sensor (high-level API).
apply_thresholds	Apply threshold configuration to multiple sensors (batch operation).
set_threshold	High-level convenience function to set threshold with retry and optional logging.
set_thresholds_from_csv	Set thresholds for all channels from a CSV file in batch operation.

API

class haniwers.v1.threshold.writer.ThresholdResult

Bases: typing.NamedTuple

Immutable result for threshold write operations.

What is this? A structured result object returned by all high-level threshold writing functions (apply_threshold, apply_thresholds). Provides detailed information about write success, the threshold value written, and how many attempts it took.

Why this matters? Callers can easily check success status and access detailed information: - result.success - Did the write succeed? - result.attempts - How many retries were needed? - result.id - Which channel? - result.vth - What threshold value was written?

This is especially useful for displaying status to users and debugging
communication issues with the detector.

NamedTuple Benefits: - Immutable: Results can’t be accidentally modified after creation - Attribute access: Use result.success instead of result[“success”] - Lightweight: No overhead compared to dict or dataclass - Type-safe: IDE provides autocomplete support (result.id, result.success, etc.) - Hashable: Can be used as dict keys if needed

Attributes: id: Channel/sensor ID (typically 1-3 for OSECHI detector) vth: Threshold value that was written (1-1023) success: Whether the write operation succeeded (True/False) attempts: Number of write attempts made (1 for immediate success, >1 if retried)

Example for beginners: >>> result = apply_threshold(device, sensor, max_retry=3) >>> if result.success: … print(f"Channel {result.id}: Set to {result.vth} in {result.attempts} attempts") … else: … print(f"Channel {result.id}: Failed after {result.attempts} attempts")

Performance note: Creating ThresholdResult is O(1) - just tuple creation, no copying or allocation.

id: int

None

vth: int

None

success: bool

None

attempts: int

None

haniwers.v1.threshold.writer.write_threshold(device: haniwers.v1.daq.device.Device, ch: int, vth: int) → bool

Write a threshold value to a specific detector channel.

This function sends the threshold value using the OSECHI bit manipulation protocol (v0-compatible). The device must be connected before calling.

The protocol: 1. Validate inputs (channel 1-3, threshold 1-1023) 2. Perform bit manipulation to create byte sequence: - val1 = 0b10000 + (vth >> 6) # Header + upper 3 bits - val2 = (vth << 2) & 0xFF # Lower 8 bits shifted 3. Write 3 bytes to device: [ch, val1, val2] 4. Read 3 response lines from device 5. Wait 0.1 seconds for device stabilization 6. Check response: channel echo = success, “dame” = rejection

Args: device: Connected Device instance for serial communication ch: Channel number (1=top, 2=mid, 3=btm layer) vth: Threshold value in 10-bit range (1-1023)

Returns: True if device confirmed successful write, False if device rejected

Raises: InvalidChannelError: If ch not in range 1-3 InvalidThresholdError: If vth not in range 1-1023

Example: >>> from haniwers.v1.daq.device import Device >>> device = Device(“/dev/tty.usbserial”) >>> device.connect() >>> success = write_threshold(device, ch=1, vth=280) >>> if success: … print(“Threshold set successfully”) >>> device.disconnect()

Note for beginners: This is a low-level function. For most use cases, prefer the higher-level set_threshold() which includes retry and logging.

The bit manipulation preserves the v0 protocol exactly for hardware
compatibility. The magic number 0b10000 (16) is the protocol header.

haniwers.v1.threshold.writer.write_threshold_with_retry(device: haniwers.v1.daq.device.Device, ch: int, vth: int, max_retry: int = 3) → haniwers.v1.threshold.writer.ThresholdResult

Write threshold value with automatic retry on failure.

This function wraps write_threshold() with retry logic to handle transient communication errors. It retries up to max_retry times with fixed 0.5 second intervals between failed attempts.

Serial communication can occasionally fail due to noise, timing issues, or temporary device unavailability. Automatic retry makes the system more reliable without requiring manual intervention.

Args: device: Connected Device instance for serial communication ch: Channel number (1=top, 2=mid, 3=btm layer) vth: Threshold value in 10-bit range (1-1023) max_retry: Maximum number of attempts (default: 3)

Returns: ThresholdResult with id, vth, success status, and attempt count

Raises: InvalidChannelError: If ch not in range 1-3 InvalidThresholdError: If vth not in range 1-1023

Example: >>> from haniwers.v1.daq.device import Device >>> device = Device(“/dev/tty.usbserial”) >>> device.connect() >>> # Will retry up to 3 times with 0.5s intervals >>> result = write_threshold_with_retry(device, ch=1, vth=280, max_retry=3) >>> if result.success: … print(f"Threshold set successfully in {result.attempts} attempts") >>> device.disconnect()

Note for beginners: This function uses fixed 0.5 second intervals between retries (not exponential backoff). This is appropriate for serial communication errors, which are typically random noise rather than rate limiting.

Validation errors (invalid channel/threshold) will raise exceptions
immediately without retry, since retrying won't fix bad input.

Performance: - Best case (immediate success): ~140ms - Worst case (3 retries fail): ~1.4s (140ms + 500ms + 140ms + 500ms + 140ms)

haniwers.v1.threshold.writer.log_threshold_operation(log_path: pathlib.Path, ch: int, vth: int, success: bool) → None

Log threshold write operation to CSV file for audit trail.

This function appends a single CSV row with timestamp, channel, threshold, and success status. The log provides an audit trail for troubleshooting detector configuration changes.

CSV Format (no header): timestamp,ch,vth,success 2024-05-20T11:00:00.123456+09:00,1,280,True

The parent directory is automatically created if it doesn’t exist, making this function safe to use without manual directory setup.

Args: log_path: Path to CSV log file (created if missing, appended if exists) ch: Channel number that was written (1-3) vth: Threshold value that was written (1-1023) success: Whether the write operation succeeded

Raises: PermissionError: If log_path directory is not writable OSError: If disk full or other I/O error occurs

Example: >>> from pathlib import Path >>> log_path = Path(“logs/threshold_operations.csv”) >>> log_threshold_operation(log_path, ch=1, vth=280, success=True) >>> # File now contains: 2024-05-20T11:00:00.123456+09:00,1,280,True

Note for beginners: This function uses append mode (‘a’) so it won’t overwrite existing logs. Each call adds one line to the end of the file.

The timestamp includes timezone (e.g., +09:00) to ensure logs are
unambiguous when comparing across different systems or time zones.

Performance: O(1) amortized - constant time per log entry (~5ms typical)

haniwers.v1.threshold.writer.load_thresholds_from_csv(csv_path: pathlib.Path) → dict[int, int]

Load threshold configuration from CSV file.

This function reads a CSV file containing channel and threshold columns, and returns a dictionary mapping channel numbers to threshold values.

Supported column names: - “threshold” (preferred): Standard threshold column name - “3sigma” (v0 format): Automatically mapped to threshold for backward compatibility

The function is flexible about extra columns - any columns not used are simply ignored, allowing use of CSV files from the threshold scanning workflow which may contain additional analysis columns.

Args: csv_path: Path to CSV file with “ch” and “threshold” (or “3sigma”) columns

Returns: Dictionary mapping channel (int) to threshold value (int), e.g., {1: 283, 2: 278, 3: 288}

Raises: FileNotFoundError: If csv_path does not exist ValueError: If “ch” column is missing or both “threshold” and “3sigma” are missing pd.errors.ParserError: If CSV format is invalid

Example: >>> from pathlib import Path >>> config = load_thresholds_from_csv(Path(“thresholds.csv”)) >>> print(config) {1: 283, 2: 278, 3: 288} >>> # Use config to set thresholds on device >>> for ch, vth in config.items(): … write_threshold(device, ch, vth)

Note for beginners: - CSV must have a “ch” column identifying which channel (1, 2, or 3) - CSV must have either “threshold” or “3sigma” column for values - Extra columns (like timestamps, sigma values) are ignored - If a channel appears multiple times, the last value wins

Performance: O(n) where n is the number of rows in the CSV file

haniwers.v1.threshold.writer.apply_threshold(device: haniwers.v1.daq.device.Device, sensor: haniwers.v1.config.model.SensorConfig, max_retry: int = 3, history_path: pathlib.Path | None = None) → haniwers.v1.threshold.writer.ThresholdResult

Apply threshold configuration to a single sensor (high-level API).

What is this? The recommended high-level API for applying threshold values. This function takes a SensorConfig object and applies its threshold value to the detector with automatic retry and optional audit logging.

Why this matters? This is the primary public API for threshold operations. It abstracts away low-level device communication details and provides a clean configuration-based interface. Users work with SensorConfig objects instead of raw integers.

Args: device: Connected Device instance for serial communication sensor: SensorConfig object with id and threshold fields max_retry: Maximum number of write attempts (default: 3) history_path: Optional Path to append operation log to. If None, no logging occurs.

Returns: ThresholdResult with operation success status, attempts made, and values written. Attributes: - result.success: True if write succeeded, False if all retries failed - result.attempts: Number of write attempts made (1 = first try, >1 = retried) - result.id: Channel ID (matches sensor.id) - result.vth: Threshold value written (matches sensor.threshold)

Raises: InvalidChannelError: If sensor.id not in range 1-3 InvalidThresholdError: If sensor.threshold not in range 1-1023 PermissionError: If history_path is provided but directory is not writable

Example for beginners: >>> from haniwers.v1.daq.device import Device >>> from haniwers.v1.config.model import SensorConfig >>> from haniwers.v1.threshold import apply_threshold >>> >>> device = Device(“/dev/tty.usbserial”) >>> device.connect() >>> >>> # Create sensor configuration >>> sensor = SensorConfig( … id=1, name=“ch1”, label=“top”, … step_size=1, threshold=280, … center=512, nsteps=10 … ) >>> >>> # Apply threshold with optional logging >>> result = apply_threshold(device, sensor, max_retry=3) >>> if result.success: … print(f"Channel {result.id}: Set to {result.vth} in {result.attempts} attempts") … else: … print(f"Channel {result.id}: Failed after {result.attempts} attempts") >>> >>> device.disconnect()

Note for beginners: This is the recommended way to set thresholds for most use cases. It handles: - Extracting channel ID and threshold value from SensorConfig - Automatic retry on communication failures - Optional audit trail logging to CSV file - Returning structured result with attempt count

For batch operations on multiple sensors, use apply_thresholds() instead.

Performance: - Best case (success on first try): ~140ms - Worst case (3 retries fail): ~1.4s

haniwers.v1.threshold.writer.apply_thresholds(device: haniwers.v1.daq.device.Device, sensors: list[haniwers.v1.config.model.SensorConfig], max_retry: int = 3, history_path: pathlib.Path | None = None) → list[haniwers.v1.threshold.writer.ThresholdResult]

Apply threshold configuration to multiple sensors (batch operation).

What is this? Batch version of apply_threshold(). Applies threshold values to multiple sensors in sequence and returns structured results for each sensor.

Why this matters? Typical experiments use multiple detector channels (e.g., top, middle, bottom). This function simplifies batch operations by applying all thresholds in one call and returning detailed results for each channel.

Args: device: Connected Device instance for serial communication sensors: List of SensorConfig objects to apply max_retry: Maximum write attempts per sensor (default: 3) history_path: Optional Path to append operation log to. If None, no logging occurs.

Returns: List of ThresholdResult objects, one per input sensor, preserving input order. Each result contains success status, attempts made, and values written.

Raises: InvalidChannelError: If any sensor.id not in range 1-3 InvalidThresholdError: If any sensor.threshold not in range 1-1023 PermissionError: If history_path is provided but directory is not writable

Example for beginners: >>> from haniwers.v1.daq.device import Device >>> from haniwers.v1.config.model import SensorConfig >>> from haniwers.v1.threshold import apply_thresholds >>> >>> device = Device(“/dev/tty.usbserial”) >>> device.connect() >>> >>> # Create sensors for all three channels >>> sensors = [ … SensorConfig(id=1, name=“ch1”, label=“top”, step_size=1, threshold=280, center=512, nsteps=10), … SensorConfig(id=2, name=“ch2”, label=“mid”, step_size=1, threshold=320, center=512, nsteps=10), … SensorConfig(id=3, name=“ch3”, label=“btm”, step_size=1, threshold=300, center=512, nsteps=10), … ] >>> >>> # Apply all thresholds with logging >>> results = apply_thresholds(device, sensors, max_retry=3) >>> >>> # Check results for each channel >>> for r in results: … status = “✓” if r.success else “✗” … print(f"{status} ch{r.id}: {r.vth} ({r.attempts} attempts)") >>> >>> device.disconnect()

Note for beginners: This function processes sensors sequentially. If one sensor fails, remaining sensors are still attempted. Check the returned list to see which channels succeeded and which failed.

Results are returned in the same order as input sensors, so results[i]
corresponds to sensors[i].

Performance: O(n) where n is number of sensors - Typical: 3 channels × ~140ms = ~420ms (all succeed on first try) - Worst case: 3 channels × 1.4s = ~4.2s (all retry 3 times)

haniwers.v1.threshold.writer.set_threshold(device: haniwers.v1.daq.device.Device, ch: int, vth: int, max_retry: int = 3, log_path: pathlib.Path | None = None) → bool

High-level convenience function to set threshold with retry and optional logging.

This function combines write_threshold_with_retry() and log_threshold_operation() into a single convenient call. It handles the complete flow of writing a threshold value with automatic retry and optionally recording the operation to an audit log.

This is the recommended way for most users to set thresholds, as it provides:

• Automatic retry on communication failures

• Optional audit trail for troubleshooting

• Clear success/failure feedback

Args: device: Connected Device instance for serial communication ch: Channel number (1=top, 2=mid, 3=btm layer) vth: Threshold value in 10-bit range (1-1023) max_retry: Maximum number of write attempts (default: 3) log_path: Optional Path to append operation log to. If None, no logging occurs.

Returns: True if write succeeded (on any attempt), False if all attempts failed

Raises: InvalidChannelError: If ch not in range 1-3 InvalidThresholdError: If vth not in range 1-1023 PermissionError: If log_path is provided but directory is not writable

Example: >>> from pathlib import Path >>> from haniwers.v1.daq.device import Device >>> from haniwers.v1.threshold import set_threshold >>> >>> device = Device(“/dev/tty.usbserial”) >>> device.connect() >>> >>> # Simple case: write without logging >>> success = set_threshold(device, ch=1, vth=280) >>> >>> # With logging for audit trail >>> success = set_threshold( … device, ch=1, vth=280, … max_retry=3, … log_path=Path(“logs/threshold_ops.csv”) … ) >>> if success: … print(“Threshold set successfully”) >>> device.disconnect()

Note for beginners: This is the recommended high-level API for most use cases. It automatically: 1. Retries up to max_retry times on communication failures 2. Logs the operation if log_path is provided 3. Returns True/False to indicate success

For fine-grained control, use write_threshold_with_retry() and
log_threshold_operation() directly.

Performance: - Best case (success on first try, no logging): ~140ms - Worst case (3 retries, with logging): ~1.5s

haniwers.v1.threshold.writer.set_thresholds_from_csv(device: haniwers.v1.daq.device.Device, csv_path: pathlib.Path, max_retry: int = 3, log_path: pathlib.Path | None = None) → dict[int, bool]

Set thresholds for all channels from a CSV file in batch operation.

This function loads threshold configuration from a CSV file and applies it to all channels in the detector. It provides a convenient way to apply threshold scanning or fitting results to the detector in one operation.

The CSV file is processed using load_thresholds_from_csv(), supporting both standard “threshold” and v0 legacy “3sigma” column names.

All channels are processed sequentially. If a channel write fails, remaining channels are still attempted. Use the returned dict to check which channels succeeded and which failed.

Args: device: Connected Device instance for serial communication csv_path: Path to CSV file with “ch” and “threshold” (or “3sigma”) columns max_retry: Maximum write attempts per channel (default: 3) log_path: Optional Path to append operation log to. If None, no logging occurs.

Returns: Dictionary mapping channel (int) to success status (bool). Example: {1: True, 2: False, 3: True} means channels 1 and 3 succeeded, 2 failed.

Raises: FileNotFoundError: If csv_path does not exist ValueError: If CSV is missing required columns (“ch” and “threshold”/“3sigma”) PermissionError: If log_path is provided but directory is not writable

Example: >>> from pathlib import Path >>> from haniwers.v1.daq.device import Device >>> from haniwers.v1.threshold import set_thresholds_from_csv >>> >>> device = Device(“/dev/tty.usbserial”) >>> device.connect() >>> >>> # Apply threshold configuration from scanning results >>> csv_file = Path(“scan_results/thresholds.csv”) >>> results = set_thresholds_from_csv( … device, csv_file, … max_retry=3, … log_path=Path(“logs/batch_ops.csv”) … ) >>> >>> # Check which channels succeeded >>> for ch, success in results.items(): … status = “OK” if success else “FAILED” … print(f"Channel {ch}: {status}") >>> >>> device.disconnect()

Note for beginners: - CSV must contain “ch” and either “threshold” or “3sigma” columns - All channels in CSV are attempted, even if some fail - Failures don’t stop processing - all channels are tried - Check the returned dict to see which channels succeeded - Use with log_path to create an audit trail of batch operations

Performance: O(n) where n is the number of channels in CSV - Typical: 3 channels × ~140ms = ~420ms (all succeed on first try) - Worst case: 3 channels × 1.4s = ~4.2s (all retry 3 times)