Emitters¶

async def start(self) -> None:
    """Start periodic emission."""
    if self._running:
        return

    self._running = True
    self._task = asyncio.create_task(self._emission_loop())

async def stop(self) -> None:
    """Stop periodic emission."""
    self._running = False
    if self._task:
        self._task.cancel()
        try:
            await self._task
        except asyncio.CancelledError:
            pass
        self._task = None

def set_interval(self, interval_ms: int) -> None:
    """Update the emission interval."""
    self.interval_ms = interval_ms

async def get_event(self, timeout: float | None = None) -> UDMEvent | None:
    """
    Get the next emitted event from the queue.

    Args:
        timeout: Maximum time to wait in seconds

    Returns:
        UDMEvent or None if timeout
    """
    try:
        if timeout is None:
            return await self._event_queue.get()
        else:
            return await asyncio.wait_for(self._event_queue.get(), timeout=timeout)
    except asyncio.TimeoutError:
        return None

def emit_state_transition(
    self,
    from_state: str,
    to_state: str,
    reason: str | None = None,
) -> UDMEvent:
    """
    Emit a state transition event.

    Args:
        from_state: Previous state
        to_state: New state
        reason: Optional reason for transition

    Returns:
        The emitted UDMEvent
    """
    builder = self._create_builder(EventType.STATE_TRANSITION)
    builder.with_operational(state=None, sub_state=to_state)

    extensions = {
        "state_transition": {
            "from_state": from_state,
            "to_state": to_state,
            "reason": reason,
        }
    }
    builder.with_extensions(extensions)

    event = builder.build()
    return self._emit_and_callback(event)

def emit_safety_violation(
    self,
    violation_id: str,
    rule_id: str,
    violation_type: str,
    severity: ViolationSeverity,
    details: dict[str, Any] | None = None,
) -> UDMEvent:
    """
    Emit a safety violation event.

    Args:
        violation_id: Unique violation identifier
        rule_id: Rule that was violated
        violation_type: Type of violation
        severity: Violation severity
        details: Additional violation details

    Returns:
        The emitted UDMEvent
    """
    builder = self._create_builder(EventType.SAFETY_VIOLATION)

    violation = SafetyViolation(
        violation_id=violation_id,
        rule_id=rule_id,
        violation_type=violation_type,
        severity=severity,
        timestamp=datetime.utcnow(),
        details=details,
    )
    builder.with_safety(violations=[violation])

    event = builder.build()
    return self._emit_and_callback(event)

def emit_task_event(
    self,
    task_id: str,
    event_type: TaskEventType,
    task_type: str | None = None,
    details: dict[str, Any] | None = None,
) -> UDMEvent:
    """
    Emit a task lifecycle event.

    Args:
        task_id: Task identifier
        event_type: Task event type (started, completed, failed, cancelled)
        task_type: Optional task type
        details: Additional task details

    Returns:
        The emitted UDMEvent
    """
    # Map to UDM event type
    type_mapping = {
        TaskEventType.STARTED: EventType.TASK_STARTED,
        TaskEventType.COMPLETED: EventType.TASK_COMPLETED,
        TaskEventType.FAILED: EventType.TASK_FAILED,
        TaskEventType.CANCELLED: EventType.TASK_CANCELLED,
    }

    builder = self._create_builder(type_mapping[event_type])
    builder.with_operational(
        task_id=task_id,
        task_type=task_type,
        task_state=event_type.value,
    )

    if details:
        builder.with_extensions({"task_details": details})

    event = builder.build()
    return self._emit_and_callback(event)

def emit_navigation_event(
    self,
    event_type: NavigationEventType,
    goal_id: str | None = None,
    goal_x_m: float | None = None,
    goal_y_m: float | None = None,
    details: dict[str, Any] | None = None,
) -> UDMEvent:
    """
    Emit a navigation event.

    Args:
        event_type: Navigation event type
        goal_id: Goal identifier
        goal_x_m: Goal X position
        goal_y_m: Goal Y position
        details: Additional details

    Returns:
        The emitted UDMEvent
    """
    # Map to UDM event type
    type_mapping = {
        NavigationEventType.GOAL_REACHED: EventType.NAVIGATION_GOAL_REACHED,
        NavigationEventType.PATH_BLOCKED: EventType.NAVIGATION_PATH_BLOCKED,
        NavigationEventType.REROUTING: EventType.NAVIGATION_REROUTING,
    }

    builder = self._create_builder(type_mapping[event_type])
    builder.with_navigation(
        goal_id=goal_id,
        goal_x_m=goal_x_m,
        goal_y_m=goal_y_m,
    )

    if details:
        builder.with_extensions({"navigation_details": details})

    event = builder.build()
    return self._emit_and_callback(event)

def emit_power_event(
    self,
    event_type: EventType,
    battery_soc_pct: float | None = None,
    is_charging: bool | None = None,
    charger_id: str | None = None,
) -> UDMEvent:
    """
    Emit a power-related event.

    Args:
        event_type: Power event type
        battery_soc_pct: Battery state of charge
        is_charging: Whether currently charging
        charger_id: Charger identifier

    Returns:
        The emitted UDMEvent
    """
    builder = self._create_builder(event_type)
    builder.with_power(
        battery_soc_pct=battery_soc_pct,
        is_charging=is_charging,
        charger_id=charger_id,
    )

    event = builder.build()
    return self._emit_and_callback(event)

def emit_custom_event(
    self,
    event_type_name: str,
    data: dict[str, Any],
) -> UDMEvent:
    """
    Emit a custom event (custom.* namespace).

    Args:
        event_type_name: Custom event type name (without 'custom.' prefix)
        data: Event data

    Returns:
        The emitted UDMEvent
    """
    # For custom events, we use TELEMETRY_ON_CHANGE as base
    # and add the custom type in extensions
    builder = self._create_builder(EventType.TELEMETRY_ON_CHANGE)
    builder.with_extensions(
        {
            "custom_event": {
                "type": f"custom.{event_type_name}",
                "data": data,
            }
        }
    )

    event = builder.build()
    return self._emit_and_callback(event)