Mock¶

Simulate an OSECHI detector without real hardware for testing and development. MockDevice provides a drop-in replacement for PortDevice, while MockGenerator creates simulated detection events with configurable timing and data sources.

Architecture¶

Class Diagram¶

classDiagram
    class DeviceProtocol {
        <<interface>>
        +send_command(command: str) bool
        +receive_response(field_type: str) DeviceResponse
        +query(command: str) DeviceResponse
        +close() None
        +__enter__() Self
        +__exit__() None
    }

    class MockDevice {
        -port: str
        -is_open: bool
        -generator: MockGenerator | None
        -events: list[dict]
        -event_index: int
        -events_path: Path
        +send_command(command: str) bool
        +receive_response(field_type: str) DeviceResponse
        +query(command: str) DeviceResponse
        +close() None
        -_get_events() dict | None
        -_get_response(command: str) dict
    }

    class MockGenerator {
        -events: list[dict]
        -index: int
        -jitter: float
        -speed: float
        -seed: int | None
        +from_file(path: str) MockGenerator
        +from_random(count: int, seed: int | None) MockGenerator
        +set_jitter(jitter: float) MockGenerator
        +set_speed(speed: float) MockGenerator
        +get_next_event() dict | None
        +reset() None
    }

    DeviceProtocol <|.. MockDevice: implements
    MockDevice --> MockGenerator: uses (optional)

Event Priority System¶

When MockDevice.receive_response(field_type="event") is called:

flowchart TD
    A["receive_response(field_type='event')"] --> B{MockGenerator<br/>provided?}
    B -->|Yes| C["Return event from<br/>MockGenerator"]
    B -->|No| D{Loaded events<br/>available?}
    D -->|Yes| E["Return event from<br/>data/events/events.jsonl"]
    D -->|No| F{Command<br/>response cached?}
    F -->|Yes| G["Return command<br/>response"]
    F -->|No| H["Return default<br/>STATUS response"]

    C --> I["type = 'mock'<br/>received_us = timestamp"]
    E --> I
    G --> I
    H --> I

MockDevice¶

`kazunoko.mock.MockDevice` ¶

Mock OSECHI detector for testing without real hardware

Loads device responses from data/responses/ directory. Automatically loads detection events from data/events/events.jsonl. Supports advanced event simulation via MockGenerator with speed/jitter.

Parameters:

Name	Type	Description	Default
`generator`	`MockGenerator \| None`	Optional MockGenerator for dynamic event simulation with speed/jitter control. If set, takes priority over loaded events.	`None`

Event Priority

MockGenerator (if provided) - dynamic events with speed/jitter
data/events/events.jsonl - static events for testing
Command responses - if neither above available

Example

from kazunoko import MockDevice, MockGenerator, Command

# Simple mock device (no code changes needed for testing)
device = MockDevice()  # Auto-loads events from data/events/
event = device.receive_response(field_type="event")

# Mock device with command responses
response = device.query("GET_STATUS")
device.close()

# Mock device with speed-controlled events (for demos/simulations)
gen = MockGenerator.from_random(count=100, seed=42)
gen.set_speed(2.0).set_jitter(0.1)
with MockDevice(generator=gen) as device:
    cmd = Command(device)
    status = cmd.status()
    event = cmd.read()

# Mock device with custom events from file
gen = MockGenerator.from_file("detector_data.jsonl")
device = MockDevice(generator=gen)
event = device.receive_response(field_type="event")

Source code in src/kazunoko/mock.py

class MockDevice:
    """
    Mock OSECHI detector for testing without real hardware

    Loads device responses from data/responses/ directory.
    Automatically loads detection events from data/events/events.jsonl.
    Supports advanced event simulation via MockGenerator with speed/jitter.

    Args:
        generator: Optional MockGenerator for dynamic event simulation with
                   speed/jitter control. If set, takes priority over loaded
                   events.

    Event Priority:
        1. MockGenerator (if provided) - dynamic events with speed/jitter
        2. data/events/events.jsonl - static events for testing
        3. Command responses - if neither above available

    Example:
        ```python
        from kazunoko import MockDevice, MockGenerator, Command

        # Simple mock device (no code changes needed for testing)
        device = MockDevice()  # Auto-loads events from data/events/
        event = device.receive_response(field_type="event")

        # Mock device with command responses
        response = device.query("GET_STATUS")
        device.close()

        # Mock device with speed-controlled events (for demos/simulations)
        gen = MockGenerator.from_random(count=100, seed=42)
        gen.set_speed(2.0).set_jitter(0.1)
        with MockDevice(generator=gen) as device:
            cmd = Command(device)
            status = cmd.status()
            event = cmd.read()

        # Mock device with custom events from file
        gen = MockGenerator.from_file("detector_data.jsonl")
        device = MockDevice(generator=gen)
        event = device.receive_response(field_type="event")
        ```
    """

    def __init__(self, generator: MockGenerator | None = None):
        """Initialize mock device with data/responses from package"""
        logger.debug("Initializing mock device", extra={"has_generator": generator is not None})
        self.port = "mock"
        self.is_open = True

        # Load available commands from data/responses/
        self.path = Path(__file__).parent / "data" / "responses"
        self.available_commands = _load_available_commands(self.path)

        # Data generator for simulating events
        self.generator = generator
        if generator:
            logger.debug(
                "Mock device initialized with generator",
                extra={
                    "event_count": len(generator.events),
                    "speed": generator.speed,
                    "jitter": generator.jitter,
                }
            )

        # Load events from data/events/ (independent of generator)
        self.events_path = Path(__file__).parent / "data" / "events"
        self.events = _load_events(self.events_path)
        self.event_index = 0
        logger.debug(
            "Mock device initialized",
            extra={
                "commands_loaded": len(self.available_commands),
                "events_loaded": len(self.events),
            }
        )

    def send_command(self, command: str) -> bool:
        """
        Send command to mock device (no-op, just validates)

        Args:
            command: Text command to send

        Returns:
            True if command is valid

        Raises:
            ValueError: If device is not open
        """
        if not self.is_open:
            raise ValueError("Device not open")

        # No state updates needed - just validation
        return True

    def receive_response(self, field_type: str) -> DeviceResponse:
        """
        Receive response from mock device

        Args:
            field_type: Expected response type ("response" or "event")

        Returns:
            Simulated DeviceResponse object with received_us timestamp
            (in microseconds)

        Raises:
            ValueError: If device is not open
        """
        if not self.is_open:
            raise ValueError("Device not open")

        received_at_us = int(time.time() * 1_000_000)

        # Priority 1: Return data from generator if available
        # (dynamic events with speed/jitter)
        if field_type == FIELD_TYPE_EVENT and self.generator:
            event = self.generator.get_next_event()
            if event:
                response = parse_jsonl(json.dumps(event))
                response.type = FIELD_TYPE_MOCK
                response.received_us = received_at_us
                return response

        # Priority 2: Return data from loaded events
        # (static events from data/events/)
        if field_type == FIELD_TYPE_EVENT and self.events:
            event = self._get_events()
            if event:
                response = parse_jsonl(json.dumps(event))
                response.type = FIELD_TYPE_MOCK
                response.received_us = received_at_us
                return response

        # Priority 3: Return last sent command response
        if hasattr(self, "_sent_cmd"):
            response = self._get_response(self._sent_cmd)
            response["type"] = FIELD_TYPE_MOCK
            response["received_us"] = received_at_us
            return DeviceResponse(**response)

        # Default STATUS response
        return DeviceResponse(
            type=FIELD_TYPE_MOCK, status="ok", received_us=received_at_us
        )

    def query(self, command: str) -> DeviceResponse:
        """
        Send command and receive response in one operation

        Args:
            command: Text command to send

        Returns:
            Simulated DeviceResponse object
        """
        self._sent_cmd = command.strip()
        self.send_command(command)
        return self.receive_response(field_type=FIELD_TYPE_RESPONSE)

    def close(self) -> None:
        """Close mock device connection"""
        self.is_open = False

    def __enter__(self):
        """Context manager entry"""
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        """Context manager exit"""
        self.close()

    def _get_events(self) -> dict[str, Any] | None:
        """
        Get next event from loaded events

        Returns:
            Next event dict or None if no more events
        """
        if self.event_index >= len(self.events):
            return None

        event = self.events[self.event_index].copy()
        self.event_index += 1
        # Remove timestamp fields that will be set dynamically
        event.pop("received_at", None)
        event.pop("sent_us", None)
        return event

    def _get_response(self, command: str) -> dict[str, Any]:
        """Get response for command"""
        cmd = command.strip()

        # Extract base command for parameterized commands
        base_cmd = cmd.split()[0].upper() if cmd else ""

        logger.debug("Getting response for command", extra={"command": cmd, "base_cmd": base_cmd})

        # Validate command exists
        if base_cmd not in self.available_commands:
            logger.warning(
                "Unknown command for mock device",
                extra={"command": base_cmd, "available_commands": sorted(self.available_commands.keys())}
            )
            return {
                "type": "mock",
                "status": "error",
                "message": f"Unknown command: {base_cmd}",
            }

        # Load from device responses directory
        response = _load_responses(base_cmd, self.path)
        if response is not None:
            # Copy response and override type to "mock"
            data = response.copy()
            data["type"] = "mock"
            # Remove timestamp fields that should be dynamic
            data.pop("received_at", None)
            data.pop("sent_us", None)
            logger.debug(
                "Response loaded for command",
                extra={"command": base_cmd, "field_count": len(data)}
            )
            return data

        # Fallback if JSON file missing (shouldn't happen if responses complete)
        logger.warning(
            "Response file not found for command",
            extra={"command": base_cmd, "path": str(self.path)}
        )
        return {
            "type": "mock",
            "status": "error",
            "message": f"Response file not found for: {base_cmd}",
        }

`enter()` ¶

Context manager entry

Source code in src/kazunoko/mock.py

def __enter__(self):
    """Context manager entry"""
    return self

`exit(exc_type, exc_val, exc_tb)` ¶

Context manager exit

Source code in src/kazunoko/mock.py

def __exit__(self, exc_type, exc_val, exc_tb):
    """Context manager exit"""
    self.close()

`init(generator=None)` ¶

Initialize mock device with data/responses from package

Source code in src/kazunoko/mock.py

def __init__(self, generator: MockGenerator | None = None):
    """Initialize mock device with data/responses from package"""
    logger.debug("Initializing mock device", extra={"has_generator": generator is not None})
    self.port = "mock"
    self.is_open = True

    # Load available commands from data/responses/
    self.path = Path(__file__).parent / "data" / "responses"
    self.available_commands = _load_available_commands(self.path)

    # Data generator for simulating events
    self.generator = generator
    if generator:
        logger.debug(
            "Mock device initialized with generator",
            extra={
                "event_count": len(generator.events),
                "speed": generator.speed,
                "jitter": generator.jitter,
            }
        )

    # Load events from data/events/ (independent of generator)
    self.events_path = Path(__file__).parent / "data" / "events"
    self.events = _load_events(self.events_path)
    self.event_index = 0
    logger.debug(
        "Mock device initialized",
        extra={
            "commands_loaded": len(self.available_commands),
            "events_loaded": len(self.events),
        }
    )

`close()` ¶

Close mock device connection

Source code in src/kazunoko/mock.py

def close(self) -> None:
    """Close mock device connection"""
    self.is_open = False

`query(command)` ¶

Send command and receive response in one operation

Parameters:

Name	Type	Description	Default
`command`	`str`	Text command to send	required

Returns:

Type	Description
`DeviceResponse`	Simulated DeviceResponse object

Source code in src/kazunoko/mock.py

def query(self, command: str) -> DeviceResponse:
    """
    Send command and receive response in one operation

    Args:
        command: Text command to send

    Returns:
        Simulated DeviceResponse object
    """
    self._sent_cmd = command.strip()
    self.send_command(command)
    return self.receive_response(field_type=FIELD_TYPE_RESPONSE)

`receive_response(field_type)` ¶

Receive response from mock device

Parameters:

Name	Type	Description	Default
`field_type`	`str`	Expected response type ("response" or "event")	required

Returns:

Type	Description
`DeviceResponse`	Simulated DeviceResponse object with received_us timestamp
`DeviceResponse`	(in microseconds)

Raises:

Type	Description
`ValueError`	If device is not open

Source code in src/kazunoko/mock.py

def receive_response(self, field_type: str) -> DeviceResponse:
    """
    Receive response from mock device

    Args:
        field_type: Expected response type ("response" or "event")

    Returns:
        Simulated DeviceResponse object with received_us timestamp
        (in microseconds)

    Raises:
        ValueError: If device is not open
    """
    if not self.is_open:
        raise ValueError("Device not open")

    received_at_us = int(time.time() * 1_000_000)

    # Priority 1: Return data from generator if available
    # (dynamic events with speed/jitter)
    if field_type == FIELD_TYPE_EVENT and self.generator:
        event = self.generator.get_next_event()
        if event:
            response = parse_jsonl(json.dumps(event))
            response.type = FIELD_TYPE_MOCK
            response.received_us = received_at_us
            return response

    # Priority 2: Return data from loaded events
    # (static events from data/events/)
    if field_type == FIELD_TYPE_EVENT and self.events:
        event = self._get_events()
        if event:
            response = parse_jsonl(json.dumps(event))
            response.type = FIELD_TYPE_MOCK
            response.received_us = received_at_us
            return response

    # Priority 3: Return last sent command response
    if hasattr(self, "_sent_cmd"):
        response = self._get_response(self._sent_cmd)
        response["type"] = FIELD_TYPE_MOCK
        response["received_us"] = received_at_us
        return DeviceResponse(**response)

    # Default STATUS response
    return DeviceResponse(
        type=FIELD_TYPE_MOCK, status="ok", received_us=received_at_us
    )

`send_command(command)` ¶

Send command to mock device (no-op, just validates)

Parameters:

Name	Type	Description	Default
`command`	`str`	Text command to send	required

Returns:

Type	Description
`bool`	True if command is valid

Raises:

Type	Description
`ValueError`	If device is not open

Source code in src/kazunoko/mock.py

def send_command(self, command: str) -> bool:
    """
    Send command to mock device (no-op, just validates)

    Args:
        command: Text command to send

    Returns:
        True if command is valid

    Raises:
        ValueError: If device is not open
    """
    if not self.is_open:
        raise ValueError("Device not open")

    # No state updates needed - just validation
    return True

MockGenerator¶

`kazunoko.mock.MockGenerator` ¶

Generator for simulated detection events

Supports multiple data sources (file, random) with configurable speed and jitter to simulate realistic detector behavior.

Source code in src/kazunoko/mock.py

class MockGenerator:
    """
    Generator for simulated detection events

    Supports multiple data sources (file, random) with configurable
    speed and jitter to simulate realistic detector behavior.
    """

    def __init__(self):
        """Initialize generator"""
        self.events: list[dict[str, Any]] = []
        self.index = 0
        self.jitter = 0.0  # Time variation between events (seconds)
        self.speed = 1.0  # Event generation speed multiplier (0.01-100x)
        self.seed = None  # Random seed for reproducibility
        self.last_event_time = 0.0

    @classmethod
    def from_file(cls, path: str) -> "MockGenerator":
        """
        Load detection events from JSONL file

        Args:
            path: Path to JSONL file (one event per line)

        Returns:
            MockGenerator instance with loaded events

        Raises:
            FileNotFoundError: If file does not exist
            json.JSONDecodeError: If JSONL format is invalid
        """
        logger.debug("Loading events from file", extra={"path": path})
        generator = cls()
        file_path = Path(path)

        if not file_path.exists():
            logger.error("Events file not found", extra={"path": path})
            raise FileNotFoundError(f"Data file not found: {path}")

        try:
            with file_path.open("r", encoding="utf-8") as f:
                for line_number, line in enumerate(f, 1):
                    line = line.strip()
                    if not line:
                        continue
                    try:
                        event = json.loads(line)
                        generator.events.append(event)
                    except json.JSONDecodeError as e:
                        logger.error(
                            "Invalid JSON in events file",
                            extra={"path": path, "line_number": line_number, "error": e.msg}
                        )
                        raise json.JSONDecodeError(
                            f"Invalid JSON at line {line_number}: {e.msg}",
                            e.doc,
                            e.pos,
                        ) from e

            logger.debug(
                "Loaded events from file",
                extra={"path": path, "event_count": len(generator.events)}
            )
        except IOError as e:
            logger.error("Failed to read events file", extra={"path": path, "error": str(e)})
            raise

        return generator

    @classmethod
    def from_random(cls, count: int = 1000, seed: int | None = None) -> "MockGenerator":
        """
        Load detection events from events.jsonl with optional shuffling

        Loads events from events.jsonl file and optionally shuffles them based on seed.
        If count is larger than available events, cycles through events repeatedly.

        Args:
            count: Number of events to return (cycles through file if needed)
            seed: Random seed for shuffling events (None = no shuffle)

        Returns:
            MockGenerator instance with events loaded from events.jsonl
        """
        logger.debug(
            "Loading events from events.jsonl with random mode",
            extra={"count": count, "seed": seed if seed is not None else "none"}
        )

        # Load events from events.jsonl
        events_path = Path(__file__).parent / "data" / "events"
        source_events = _load_events(events_path)

        if not source_events:
            logger.warning(
                "No events loaded from events.jsonl, returning empty generator",
                extra={"path": str(events_path)}
            )
            generator = cls()
            generator.seed = seed
            return generator

        # Shuffle events if seed is provided
        if seed is not None:
            random.seed(seed)
            random.shuffle(source_events)
            logger.debug(
                "Shuffled events with seed",
                extra={"seed": seed, "event_count": len(source_events)}
            )

        # Create generator and populate with count events (cycling if needed)
        generator = cls()
        generator.seed = seed

        for i in range(count):
            # Use modulo to cycle through events
            event_idx = i % len(source_events)
            generator.events.append(source_events[event_idx].copy())

        logger.debug(
            "Loaded events from events.jsonl",
            extra={
                "count": len(generator.events),
                "source_count": len(source_events),
                "seed": seed if seed is not None else "none",
                "cycled": count > len(source_events),
            }
        )
        return generator

    def set_jitter(self, jitter: float = 0.0) -> "MockGenerator":
        """
        Set time variation between events

        Args:
            jitter: Random time variation in seconds (0.0-inf)

        Returns:
            Self for method chaining
        """
        if jitter < 0:
            logger.warning("Invalid jitter value", extra={"jitter": jitter})
            raise ValueError("jitter must be non-negative")
        self.jitter = jitter
        logger.debug("Set jitter for mock generator", extra={"jitter": jitter})
        return self

    def set_speed(self, speed: float = 1.0) -> "MockGenerator":
        """
        Set event generation speed multiplier

        Args:
            speed: Speed multiplier (0.01-100x)
                   1.0 = normal speed, 2.0 = 2x faster, 0.5 = half speed

        Returns:
            Self for method chaining

        Raises:
            ValueError: If speed is outside valid range
        """
        if not (0.01 <= speed <= 100):
            logger.warning("Invalid speed value", extra={"speed": speed})
            raise ValueError("speed must be between 0.01 and 100")
        self.speed = speed
        logger.debug("Set speed for mock generator", extra={"speed": speed})
        return self

    def get_next_event(self) -> dict[str, Any] | None:
        """
        Get next event with speed and jitter applied

        Returns:
            Next event dict with type set to "mock" and detected_us preserved from events.jsonl,
            or None if no more events

        Raises:
            RuntimeError: If used with from_random() without seed
        """
        if self.index >= len(self.events):
            return None

        event = self.events[self.index].copy()
        self.index += 1

        # Apply time-based delays (speed and jitter)
        if self.index > 1:  # Skip first event
            current_time = time.time()
            # Base interval with speed factor
            base_interval = 1.0 / self.speed
            # Add jitter
            jitter_amount = random.uniform(-self.jitter, self.jitter)
            interval = max(0.0, base_interval + jitter_amount)

            elapsed = current_time - self.last_event_time
            if elapsed < interval:
                time.sleep(interval - elapsed)

        self.last_event_time = time.time()

        # Set type to "mock" to differentiate mock data from real device data
        event["type"] = FIELD_TYPE_MOCK

        return event

    def reset(self) -> None:
        """Reset generator to first event"""
        self.index = 0
        self.last_event_time = 0.0

`init()` ¶

Initialize generator

Source code in src/kazunoko/mock.py

def __init__(self):
    """Initialize generator"""
    self.events: list[dict[str, Any]] = []
    self.index = 0
    self.jitter = 0.0  # Time variation between events (seconds)
    self.speed = 1.0  # Event generation speed multiplier (0.01-100x)
    self.seed = None  # Random seed for reproducibility
    self.last_event_time = 0.0

`from_file(path)` `classmethod` ¶

Load detection events from JSONL file

Parameters:

Name	Type	Description	Default
`path`	`str`	Path to JSONL file (one event per line)	required

Returns:

Type	Description
`MockGenerator`	MockGenerator instance with loaded events

Raises:

Type	Description
`FileNotFoundError`	If file does not exist
`JSONDecodeError`	If JSONL format is invalid

Source code in src/kazunoko/mock.py

@classmethod
def from_file(cls, path: str) -> "MockGenerator":
    """
    Load detection events from JSONL file

    Args:
        path: Path to JSONL file (one event per line)

    Returns:
        MockGenerator instance with loaded events

    Raises:
        FileNotFoundError: If file does not exist
        json.JSONDecodeError: If JSONL format is invalid
    """
    logger.debug("Loading events from file", extra={"path": path})
    generator = cls()
    file_path = Path(path)

    if not file_path.exists():
        logger.error("Events file not found", extra={"path": path})
        raise FileNotFoundError(f"Data file not found: {path}")

    try:
        with file_path.open("r", encoding="utf-8") as f:
            for line_number, line in enumerate(f, 1):
                line = line.strip()
                if not line:
                    continue
                try:
                    event = json.loads(line)
                    generator.events.append(event)
                except json.JSONDecodeError as e:
                    logger.error(
                        "Invalid JSON in events file",
                        extra={"path": path, "line_number": line_number, "error": e.msg}
                    )
                    raise json.JSONDecodeError(
                        f"Invalid JSON at line {line_number}: {e.msg}",
                        e.doc,
                        e.pos,
                    ) from e

        logger.debug(
            "Loaded events from file",
            extra={"path": path, "event_count": len(generator.events)}
        )
    except IOError as e:
        logger.error("Failed to read events file", extra={"path": path, "error": str(e)})
        raise

    return generator

`from_random(count=1000, seed=None)` `classmethod` ¶

Load detection events from events.jsonl with optional shuffling

Loads events from events.jsonl file and optionally shuffles them based on seed. If count is larger than available events, cycles through events repeatedly.

Parameters:

Name	Type	Description	Default
`count`	`int`	Number of events to return (cycles through file if needed)	`1000`
`seed`	`int \| None`	Random seed for shuffling events (None = no shuffle)	`None`

Returns:

Type	Description
`MockGenerator`	MockGenerator instance with events loaded from events.jsonl

Source code in src/kazunoko/mock.py

@classmethod
def from_random(cls, count: int = 1000, seed: int | None = None) -> "MockGenerator":
    """
    Load detection events from events.jsonl with optional shuffling

    Loads events from events.jsonl file and optionally shuffles them based on seed.
    If count is larger than available events, cycles through events repeatedly.

    Args:
        count: Number of events to return (cycles through file if needed)
        seed: Random seed for shuffling events (None = no shuffle)

    Returns:
        MockGenerator instance with events loaded from events.jsonl
    """
    logger.debug(
        "Loading events from events.jsonl with random mode",
        extra={"count": count, "seed": seed if seed is not None else "none"}
    )

    # Load events from events.jsonl
    events_path = Path(__file__).parent / "data" / "events"
    source_events = _load_events(events_path)

    if not source_events:
        logger.warning(
            "No events loaded from events.jsonl, returning empty generator",
            extra={"path": str(events_path)}
        )
        generator = cls()
        generator.seed = seed
        return generator

    # Shuffle events if seed is provided
    if seed is not None:
        random.seed(seed)
        random.shuffle(source_events)
        logger.debug(
            "Shuffled events with seed",
            extra={"seed": seed, "event_count": len(source_events)}
        )

    # Create generator and populate with count events (cycling if needed)
    generator = cls()
    generator.seed = seed

    for i in range(count):
        # Use modulo to cycle through events
        event_idx = i % len(source_events)
        generator.events.append(source_events[event_idx].copy())

    logger.debug(
        "Loaded events from events.jsonl",
        extra={
            "count": len(generator.events),
            "source_count": len(source_events),
            "seed": seed if seed is not None else "none",
            "cycled": count > len(source_events),
        }
    )
    return generator

`get_next_event()` ¶

Get next event with speed and jitter applied

Returns:

Type	Description
`dict[str, Any] \| None`	Next event dict with type set to "mock" and detected_us preserved from events.jsonl,
`dict[str, Any] \| None`	or None if no more events

Raises:

Type	Description
`RuntimeError`	If used with from_random() without seed

Source code in src/kazunoko/mock.py

def get_next_event(self) -> dict[str, Any] | None:
    """
    Get next event with speed and jitter applied

    Returns:
        Next event dict with type set to "mock" and detected_us preserved from events.jsonl,
        or None if no more events

    Raises:
        RuntimeError: If used with from_random() without seed
    """
    if self.index >= len(self.events):
        return None

    event = self.events[self.index].copy()
    self.index += 1

    # Apply time-based delays (speed and jitter)
    if self.index > 1:  # Skip first event
        current_time = time.time()
        # Base interval with speed factor
        base_interval = 1.0 / self.speed
        # Add jitter
        jitter_amount = random.uniform(-self.jitter, self.jitter)
        interval = max(0.0, base_interval + jitter_amount)

        elapsed = current_time - self.last_event_time
        if elapsed < interval:
            time.sleep(interval - elapsed)

    self.last_event_time = time.time()

    # Set type to "mock" to differentiate mock data from real device data
    event["type"] = FIELD_TYPE_MOCK

    return event

`reset()` ¶

Reset generator to first event

Source code in src/kazunoko/mock.py

def reset(self) -> None:
    """Reset generator to first event"""
    self.index = 0
    self.last_event_time = 0.0

`set_jitter(jitter=0.0)` ¶

Set time variation between events

Parameters:

Name	Type	Description	Default
`jitter`	`float`	Random time variation in seconds (0.0-inf)	`0.0`

Returns:

Type	Description
`MockGenerator`	Self for method chaining

Source code in src/kazunoko/mock.py

def set_jitter(self, jitter: float = 0.0) -> "MockGenerator":
    """
    Set time variation between events

    Args:
        jitter: Random time variation in seconds (0.0-inf)

    Returns:
        Self for method chaining
    """
    if jitter < 0:
        logger.warning("Invalid jitter value", extra={"jitter": jitter})
        raise ValueError("jitter must be non-negative")
    self.jitter = jitter
    logger.debug("Set jitter for mock generator", extra={"jitter": jitter})
    return self

`set_speed(speed=1.0)` ¶

Set event generation speed multiplier

Parameters:

Name	Type	Description	Default
`speed`	`float`	Speed multiplier (0.01-100x) 1.0 = normal speed, 2.0 = 2x faster, 0.5 = half speed	`1.0`

Returns:

Type	Description
`MockGenerator`	Self for method chaining

Raises:

Type	Description
`ValueError`	If speed is outside valid range

Source code in src/kazunoko/mock.py

def set_speed(self, speed: float = 1.0) -> "MockGenerator":
    """
    Set event generation speed multiplier

    Args:
        speed: Speed multiplier (0.01-100x)
               1.0 = normal speed, 2.0 = 2x faster, 0.5 = half speed

    Returns:
        Self for method chaining

    Raises:
        ValueError: If speed is outside valid range
    """
    if not (0.01 <= speed <= 100):
        logger.warning("Invalid speed value", extra={"speed": speed})
        raise ValueError("speed must be between 0.01 and 100")
    self.speed = speed
    logger.debug("Set speed for mock generator", extra={"speed": speed})
    return self

Mock¶

Architecture¶

Class Diagram¶

Event Priority System¶

MockDevice¶

kazunoko.mock.MockDevice ¶

__enter__() ¶

__exit__(exc_type, exc_val, exc_tb) ¶

__init__(generator=None) ¶

close() ¶

query(command) ¶

receive_response(field_type) ¶

send_command(command) ¶

MockGenerator¶

kazunoko.mock.MockGenerator ¶

__init__() ¶

from_file(path) classmethod ¶

from_random(count=1000, seed=None) classmethod ¶

get_next_event() ¶

reset() ¶

set_jitter(jitter=0.0) ¶

set_speed(speed=1.0) ¶

`kazunoko.mock.MockDevice` ¶

`enter()` ¶

`exit(exc_type, exc_val, exc_tb)` ¶

`init(generator=None)` ¶

`close()` ¶

`query(command)` ¶

`receive_response(field_type)` ¶

`send_command(command)` ¶

`kazunoko.mock.MockGenerator` ¶

`init()` ¶

`from_file(path)` `classmethod` ¶

`from_random(count=1000, seed=None)` `classmethod` ¶

`get_next_event()` ¶

`reset()` ¶

`set_jitter(jitter=0.0)` ¶

`set_speed(speed=1.0)` ¶