`haniwers.v1.daq.mocker`#

Fake detector for testing software without real hardware.

What is this module? This module simulates an OSECHI cosmic ray detector when you don’t have the real hardware. Perfect for development, testing, and teaching.

Two ways to get fake detector data:

Mocker: Replays recorded data from a CSV file
- Use real detector measurements saved from previous experiments
- Control playback speed (1x, 2x, 10x, etc.)
- Add timing variation (jitter) for realism
- Loop the data or play once
- Perfect for reproducible testing
RandomMocker: Generates random synthetic data
- Creates fake detector measurements with realistic ranges
- No CSV file needed
- Useful for stress testing (fast playback)
- Great when you have no recorded data available

Why use fake detectors? ✓ Develop code without hardware (saves money, time) ✓ Test code reliably (same data every time) ✓ Teach students without hardware (democratizes learning) ✓ Debug code issues (can replay specific scenarios) ✓ Benchmark code performance (synthetic data is fast)

Key interfaces: Both Mocker and RandomMocker implement the same interface as real detectors:

readline(): Get next measurement
write(): Fake send commands (no-op)
flush(): Fake buffer clear (no-op)
close(): Stop the fake detector
is_open: Boolean flag (True/False)

Example (Mocker - replay recorded data):

from pathlib import Path
from haniwers.v1.config.model import MockerConfig
from haniwers.v1.daq.mocker import Mocker

config = MockerConfig(
    csv_path=Path("recorded_data.csv"),
    shuffle=False,      # Don't randomize order
    speed=2.0,          # 2x faster replay
    jitter=0.05,        # Add tiny timing variation
    loop=True           # Repeat forever
)

mocker = Mocker(config, seed=42)
for i in range(5):
    line = mocker.readline()
    print(f"Event {i}: {line.decode()}")

Example (RandomMocker - generate random data):

from haniwers.v1.config.model import MockerConfig
from haniwers.v1.daq.mocker import RandomMocker

config = MockerConfig(speed=10.0)  # csv_path not needed
mocker = RandomMocker(config, seed=42)

for i in range(5):
    line = mocker.readline()
    print(f"Random event {i}: {line.decode()}")

When to use each:

Mocker: - Testing code with realistic data - Reproducing specific scenarios - CI/CD pipelines (deterministic testing) - Teaching with real examples

RandomMocker: - Stress testing (lots of data quickly) - Testing error handling - When you have no recorded data - Benchmarking performance

Module Contents#

Classes#

`BaseMocker`	Abstract base class for all mock serial devices.
`Mocker`	Mock serial device that replays events from a CSV file.
`RandomMocker`	Mock serial device that generates random sensor values.

Functions#

`load_events`	Load recorded detector measurements from a CSV file.
`generate_fields`	Generate one random fake detector measurement.

Data#

log

API#

haniwers.v1.daq.mocker.log#: ‘bind(…)’

haniwers.v1.daq.mocker.load_events(path: pathlib.Path, jitter: float, speed: float, shuffle: bool) → list[haniwers.v1.daq.model.MockEvent]#

Load recorded detector measurements from a CSV file.

What this does: Reads a CSV file containing previous detector measurements and prepares them for playback with optional speed control and randomization.

Args: path: Path to CSV file (e.g., “data/detector_run_001.csv”) jitter: Timing variation in seconds (0.0 = exact, 0.1 = ±0.1s random) speed: Playback speed (1.0 = normal, 2.0 = twice as fast, 10.0 = 10x fast) shuffle: Mix up the event order randomly (True/False)

Returns: List of MockEvent objects ready for playback with Mocker

Raises: FileNotFoundError: If the CSV file doesn’t exist ValueError: If the CSV format is wrong (missing columns, bad types, etc.)

CSV file format: - No header row (just data) - Columns (in order): timestamp, top, mid, btm, adc, tmp, atm, hmd - timestamp: ISO8601 timestamp (e.g., “2025-10-19T14:23:45+09:00”) - Numbers: integers for sensors, floats for environmental data

How it works: 1. Read all rows from CSV file 2. For each row: Create a MockEvent with timing info 3. Calculate time between events (deltaT) 4. If shuffle=True: Randomize the order 5. Return list of events

Performance: - Fast: Loads 100K rows in <3 seconds - Memory: ~300-350 MB for 100K rows

Example:

from pathlib import Path
from haniwers.v1.daq.mocker import load_events

# Load recorded data
events = load_events(
    path=Path("data/detector_run_001.csv"),
    jitter=0.05,        # ±0.05s timing variation
    speed=2.0,          # Play back 2x faster
    shuffle=False       # Keep original order
)

print(f"Loaded {len(events)} events")
# Output: Loaded 10000 events

# Peek at first event
first = events[0]
print(f"First event: {first.raw.to_serial()}")
# Output: First event: 5 2 8 512 25.43 100550.12 55.67

When to use: - Testing code with real detector data - Reproducing specific measurement scenarios - Automated testing (same data every time)

haniwers.v1.daq.mocker.generate_fields() → list[str]#

Generate one random fake detector measurement.

What this does: Creates a single detector event with realistic random values. Similar to what the detector would send, but all values are random.

Returns: List of 8 text values: [timestamp, top, mid, btm, adc, temp, pressure, humidity] Ready to be parsed into a RawEvent

Value ranges (realistic for cosmic ray detector): - timestamp: Current time (when this function is called) - top sensor: 0-10 (cosmic ray signal) - mid sensor: 0-10 (cosmic ray signal) - btm sensor: 0-10 (cosmic ray signal) - adc: 0-1024 (pulse height) - temperature: 20-30 °C - pressure: 100500-100600 Pa (atmospheric) - humidity: 30-70 % (relative)

Example:

from haniwers.v1.daq.mocker import generate_fields
from haniwers.v1.daq.model import RawEvent

# Generate random sensor values
fields = generate_fields()
print(fields)
# Output: ['2025-10-19T14:23:45+09:00', '5', '2', '8', '512', '25.43', '100550.12', '55.67']

# Convert to RawEvent
event = RawEvent.from_list(fields)
print(f"Sensor readings: top={event.top}, mid={event.mid}, btm={event.btm}")
# Output: Sensor readings: top=5, mid=2, btm=8

Note: This uses Python’s global random state. For reproducible (deterministic) generation, use RandomMocker with a seed instead.

When to use: - Quick testing with random data - Generating many events for stress testing - When exact reproducibility isn’t important

class haniwers.v1.daq.mocker.BaseMocker(config: haniwers.v1.config.model.MockerConfig)#

Bases: abc.ABC

Abstract base class for all mock serial devices.

Provides the standard serial device interface (readline, write, flush, close) that real hardware implements. This ensures mock devices can be swapped in transparently.

Attributes: config (MockerConfig): Device configuration csv_path (Optional[Path]): Path to CSV file (may be None for RandomMocker) shuffle (bool): Whether to shuffle events speed (float): Playback speed multiplier jitter (float): Timing jitter amount loop (bool): Whether to loop playback is_open (bool): Device state (True = open, False = closed)

Initialization

Initialize base mocker with configuration.

Args: config: Mocker configuration object

Postconditions: - self.is_open = True - Configuration values stored in instance attributes - self._response_queue initialized for threshold write responses

set_next_response(response: str) → None#

Queue response for next readline() call.

What this does: When a threshold write operation needs to simulate a device response (like confirming the channel number), this method queues the response so the next readline() call will return it instead of detector data.

Args: response: Response string to queue (e.g., “1” for channel 1 success, “dame” for rejection)

Example: >>> mocker.set_next_response(“1”) >>> response = mocker.readline() # Returns “1”

Note: This is primarily used by threshold write operations to simulate device responses during testing with mock devices. Normal DAQ operations don’t use this method.

connect() → None#

Open the mock device (set is_open=True).

What this does: Opens the mock device for reading. For mock devices, this just sets is_open=True. Compatible with Device.connect() interface.

Raises: Nothing - mock devices always open successfully

Example: >>> mocker = RandomMocker(config) >>> mocker.connect() # Now ready to read >>> mocker.is_open True

Note: Mock devices are automatically open after init, so this is often a no-op unless close() was called first. But it’s provided for interface compatibility with Device.

abstractmethod readline() → bytes#

Read one line from the mock serial interface.

Abstract method - must be implemented by subclasses.

Returns: UTF-8 encoded bytes representing one event in serial format, or empty bytes (b"") if device is closed or no data available.

Example: b"2 0 0 1136 27.37 100594.35 41.43"

write(data: bytes) → int#

Simulate writing data to the device.

Mock devices don’t actually write data, but this method provides compatibility with real serial device interface.

Args: data: Bytes to “write” (ignored)

Returns: Number of bytes “written” (always len(data))

Example: >>> mocker.write(b"command") 7

flush() → None#

Simulate flushing the device buffer.

Mock devices don’t have buffers, but this method provides compatibility with real serial device interface.

This is a no-op (does nothing).

Example: >>> mocker.flush() # No effect

disconnect() → None#

Close the mock device.

Sets is_open = False. Subsequent readline() calls will return empty string.

Postconditions: - self.is_open = False - readline() returns “”

Example: >>> mocker.disconnect() >>> mocker.readline() ‘’

is_available() → bool#

Check if the mock device is available (always True for mocks).

What this does: Mock devices are always available (they don’t depend on hardware). This method exists for compatibility with Device.is_available().

Returns: bool: Always True (mock device is always available)

Example: >>> mocker = RandomMocker(config) >>> mocker.is_available() True

Note: Unlike Device which checks if a serial port exists, mock devices don’t depend on hardware, so this always returns True.

with_timeout(sec: float)#

Context manager for temporary timeout change (no-op for mocks).

What this does: For compatibility with Device.with_timeout(), but mock devices don’t have real timeouts, so this is a no-op (does nothing).

Args: sec (float): Timeout value (ignored for mock devices)

Yields: self: The mocker instance (for use in with statement)

Example: >>> mocker = RandomMocker(config) >>> with mocker.with_timeout(2.0): … line = mocker.readline() # Timeout doesn’t affect mock device

Note: Mock devices always respond instantly, so timeout has no effect. This method is provided for interface compatibility only.

class haniwers.v1.daq.mocker.Mocker(config: haniwers.v1.config.model.MockerConfig, seed: int | None = None)#

Bases: haniwers.v1.daq.mocker.BaseMocker

Mock serial device that replays events from a CSV file.

Loads recorded event data and replays it with configurable timing, shuffling, and looping. Useful for debugging with real event sequences.

Attributes: events (list[MockEvent]): Loaded events ready for playback index (int): Current playback position (0 to len(events)-1)

Example: >>> config = MockerConfig(csv_path=Path(“data.csv”), speed=2.0, loop=True) >>> mocker = Mocker(config) >>> data = mocker.readline() # Reads first event >>> data = mocker.readline() # Reads second event

Initialization

Initialize Mocker with CSV file.

Args: config: Mocker configuration (csv_path is required) seed: Optional random seed for deterministic shuffling

Raises: ValueError: If config.csv_path is None FileNotFoundError: If CSV file doesn’t exist ValueError: If CSV format is invalid

Postconditions: - self.events contains loaded MockEvent objects - Events are shuffled if config.shuffle=True - self.index = 0 (ready to read first event) - Random seed is set if provided

Example: >>> config = MockerConfig( … csv_path=Path(“data.csv”), … shuffle=True, … speed=2.0, … jitter=0.1, … loop=True … ) >>> mocker = Mocker(config, seed=42)

readline() → str#

Read one event from the loaded CSV data.

Behavior: 1. Check response queue first for threshold write responses 2. If queue has items: return self._response_queue.popleft() 3. If device is closed or no events, returns “” 4. Gets current event from self.events[self.index] 5. Calculates sleep interval: max(0, gauss(deltaT/speed, jitter)) 6. Sleeps for calculated interval (simulates waiting for this event) 7. Advances index (loops if config.loop=True, stops if False) 8. Returns event as string (compatible with Device.readline())

Returns: Space-separated sensor values (no timestamp), or “” if device is closed or no events.

Example: >>> mocker = Mocker(config) >>> mocker.set_next_response(“1”) # Queue response >>> data = mocker.readline() # Returns “1” >>> data = mocker.readline() # Returns detector event >>> data ‘2 0 0 1136 27.37 100594.35 41.43’ >>> type(data) <class ‘str’>

Timing: - Sleep interval = max(0, random.gauss(event.deltaT / event.speed, event.jitter)) - Gaussian jitter adds realistic variation - Negative intervals clamped to 0 (no negative sleep) - Sleep happens BEFORE returning event (correct timing simulation)

Looping: - If config.loop=True: index wraps to 0 at end - If config.loop=False: index stops at last event (stays at last)

class haniwers.v1.daq.mocker.RandomMocker(config: haniwers.v1.config.model.MockerConfig, seed: int | None = None)#

Bases: haniwers.v1.daq.mocker.BaseMocker

Mock serial device that generates random sensor values.

Generates synthetic event data with realistic value ranges. Useful for stress testing or when no recorded data is available.

Attributes: deltaT (float): Fixed time interval (1.0 seconds) rng (random.Random): Dedicated random number generator

Example: >>> config = MockerConfig(csv_path=None, speed=10.0) >>> mocker = RandomMocker(config, seed=42) >>> data = mocker.readline() # Generates random event

Initialization

Initialize RandomMocker with optional seed.

Args: config: Mocker configuration (csv_path is ignored) seed: Optional random seed for reproducibility

Postconditions: - self.deltaT = 1.0 (fixed interval) - self.rng = random.Random(seed) (dedicated RNG instance) - Random seed set if provided

Note: csv_path in config is ignored. RandomMocker doesn’t use CSV files.

Example: >>> config = MockerConfig(csv_path=None, speed=10.0, jitter=0.05) >>> mocker = RandomMocker(config, seed=42) # Deterministic >>> mocker2 = RandomMocker(config) # Non-deterministic

readline() → str#

Generate one random event.

Behavior: 1. Check response queue first for threshold write responses 2. If queue has items: return self._response_queue.popleft() 3. If device is closed, returns “” 4. Generates random values using self.rng: - time: current timestamp (pendulum.now()) - top, mid, btm: randint(0, 10) - adc: randint(0, 1024) - tmp: uniform(20, 30) - atm: uniform(100500, 100600) # FIXED RANGE - hmd: uniform(30, 70) 5. Creates MockEvent with deltaT=1.0 6. Calculates sleep interval: max(0, gauss(1.0/speed, jitter)) 7. Sleeps for calculated interval (ONCE - double sleep bug fixed) 8. Returns event as string (compatible with Device.readline())

Returns: Space-separated sensor values (no timestamp), or “” if device is closed.

Value Ranges: - top, mid, btm: 0-10 (integer, uniform distribution) - adc: 0-1024 (integer, uniform distribution) - tmp: 20-30°C (float, uniform distribution) - atm: 100500-100600 Pa (float, uniform distribution) - hmd: 30-70% (float, uniform distribution)

Example: >>> mocker = RandomMocker(config, seed=42) >>> mocker.set_next_response(“1”) # Queue response >>> mocker.readline() ‘1’ >>> mocker.readline() ‘5 2 8 512 25.43 100550.12 55.67’ >>> mocker.readline() ‘3 0 1 789 28.91 100582.45 42.33’

Reproducibility: - Same seed produces identical sequence - No seed (seed=None) produces non-deterministic sequence

haniwers.v1.daq.mocker

目次

haniwers.v1.daq.mocker#

Module Contents#

Classes#

Functions#

Data#

API#

`haniwers.v1.daq.mocker`#