`haniwers.v1.threshold.writer`#

Threshold writing functions for detector channels.

Provides functions to write threshold values with validation, retry logic, CSV integration, and logging. Uses v1 text-based command protocol with strict JSON validation.

Two-layer architecture: - Low-level: write_threshold(), write_threshold_with_retry() - High-level (recommended): apply_threshold(), apply_thresholds()

Public API: - apply_threshold(): Single sensor with retry and optional logging - apply_thresholds(): Multiple sensors in batch - write_threshold_to_csv(): Append operation result to CSV log

Module Contents#

Functions#

`write_threshold`	Write a threshold value to a detector channel (low-level API).
`write_threshold_with_retry`	Write threshold with automatic retry on communication failure.
`write_threshold_to_csv`	Write threshold operation result to CSV file for audit trail.
`apply_threshold`	Apply threshold to a single sensor (recommended high-level API).
`apply_thresholds`	Apply thresholds to multiple sensors (batch operation).

Data#

`DEVICE_STABILIZATION_DELAY`
`RETRY_INTERVAL`

API#

haniwers.v1.threshold.writer.DEVICE_STABILIZATION_DELAY#: 0.1

haniwers.v1.threshold.writer.RETRY_INTERVAL#: 0.5

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

Write a threshold value to a detector channel (low-level API).

Sends the threshold value using the v1 text-based protocol.
The device must be connected before calling this function.

Protocol (v1 text-based):
    1. Validate channel (1-3) and threshold (1-1023)
    2. Send text command: "SET_THRESHOLD {channel} {value}

" 3. Read device response and parse: - JSON format: {“type”:“response”,“status”:“ok”,“channel”:ch,“threshold”:vth} - Threshold field must match requested value (strict validation)

Args:
    device: Connected Device (real or mock) for serial communication
    ch: Channel number (1=top, 2=mid, 3=btm)
    vth: Threshold value (1-1023)

Returns:
    True if device confirmed write with matching threshold, False if rejected or mismatched

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

Example:
    >>> device = Device("/dev/tty.usbserial")
    >>> device.connect()
    >>> success = write_threshold(device, ch=1, vth=280)
    >>> device.disconnect()

Note:
    This is a low-level function with no retry logic. For most use cases,
    use `apply_threshold()` which adds retry and logging automatically.
    Requires kurikintons v1.6+ firmware.

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.model.ThresholdWriteResult#

Write threshold with automatic retry on communication failure.

Wraps write_threshold() with retry logic for reliability. Retries up to max_retry times with 0.5 second intervals between attempts. Returns structured result with attempt count.

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

Args: device: Connected Device for serial communication ch: Channel number (1-3) vth: Threshold value (1-1023) max_retry: Maximum write attempts (default: 3)

Returns: ThresholdWriteResult with success status and attempts made

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

Example: >>> device = Device(“/dev/tty.usbserial”) >>> device.connect() >>> result = write_threshold_with_retry(device, ch=1, vth=280, max_retry=3) >>> print(f"Success: {result.success}, Attempts: {result.attempts}")

haniwers.v1.threshold.writer.write_threshold_to_csv(csv_path: pathlib.Path, result: haniwers.v1.threshold.model.ThresholdWriteResult) → None#

Write threshold operation result to CSV file for audit trail.

Appends a CSV row with all ThresholdWriteResult fields to create a complete audit log. Creates parent directory automatically if needed. Uses ISO8601 timestamp with timezone for unambiguous time recording.

CSV format (header on first creation only): timestamp,id,vth,success,attempts 2024-05-20T11:00:00.123456+09:00,1,280,True,1

Args: csv_path: Path to CSV log file (appended if exists, created if missing) result: ThresholdWriteResult with all operation details

Raises: PermissionError: If csv directory is not writable

Example: >>> from pathlib import Path >>> result = ThresholdWriteResult( … timestamp=“2024-05-20T11:00:00.123456+09:00”, … id=1, … vth=280, … success=True, … attempts=1 … ) >>> write_threshold_to_csv(Path(“logs/ops.csv”), result)

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.model.ThresholdWriteResult#

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

Writes a sensor’s threshold value to the detector with automatic retry and optional CSV logging. Recommended for most use cases. For batch operations, use apply_thresholds().

Args: device: Connected Device for serial communication sensor: SensorConfig with id and threshold fields max_retry: Maximum write attempts (default: 3) history_path: Optional path to append operation log (CSV format)

Returns: ThresholdWriteResult with success status, attempts made, and values written

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

Example: >>> from haniwers.v1.daq.device import Device >>> from haniwers.v1.config.model import SensorConfig >>> device = Device(“/dev/tty.usbserial”) >>> device.connect() >>> sensor = SensorConfig(id=1, name=“ch1”, label=“top”, step_size=1, … threshold=280, center=512, nsteps=10) >>> result = apply_threshold(device, sensor) >>> print(f"Ch{result.id}: {result.vth} ({result.attempts} attempts)")

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.model.ThresholdWriteResult]#

Apply thresholds to multiple sensors (batch operation).

Applies threshold values to multiple sensors sequentially. If one sensor fails, remaining sensors are still attempted. Results are returned in the same order as input sensors.

Args: device: Connected Device 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 (CSV format)

Returns: List of ThresholdWriteResult objects (one per sensor), in input order

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

Example: >>> from haniwers.v1.daq.device import Device >>> from haniwers.v1.config.model import SensorConfig >>> device = Device(“/dev/tty.usbserial”) >>> device.connect() >>> 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), … ] >>> results = apply_thresholds(device, sensors) >>> for r in results: … print(f"Ch{r.id}: {‘OK’ if r.success else ‘FAILED’}")

haniwers.v1.threshold.writer

目次

haniwers.v1.threshold.writer#

Module Contents#

Functions#

Data#

API#

`haniwers.v1.threshold.writer`#