UDM Event & Builder¶

def to_dict(self) -> dict[str, Any]:
    """Convert event to dictionary, excluding None values."""
    return self.model_dump(exclude_none=True, mode="json")

def to_json(self) -> str:
    """Convert event to JSON string."""
    return self.model_dump_json(exclude_none=True)

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "UDMEvent":
    """Create event from dictionary."""
    return cls.model_validate(data)

@classmethod
def from_json(cls, json_str: str) -> "UDMEvent":
    """Create event from JSON string."""
    return cls.model_validate_json(json_str)

def with_event_type(self, event_type: EventType) -> Self:
    """Set the event type."""
    self._event_type = event_type
    return self

def with_event_id(self, event_id: str) -> Self:
    """Set a custom event ID (default is auto-generated UUID v7)."""
    self._event_id = event_id
    return self

def with_captured_at(self, captured_at: datetime) -> Self:
    """Set the capture timestamp."""
    self._captured_at = captured_at
    return self

def with_session(self, session_id: str) -> Self:
    """Set the session ID for grouping events."""
    self._session_id = session_id
    return self

def with_sequence(self, sequence_num: int) -> Self:
    """Set the sequence number."""
    self._sequence_num = sequence_num
    return self

def with_identity(
    self,
    source_name: str | None = None,
    platform: str | None = None,
    platform_version: str | None = None,
    firmware_version: str | None = None,
    hardware_revision: str | None = None,
    serial_number: str | None = None,
    fleet_id: str | None = None,
    site_id: str | None = None,
    zone_id: str | None = None,
    organization_id: str | None = None,
    tags: list[str] | None = None,
) -> Self:
    """Set identity domain information."""
    self._identity = IdentityDomain(
        source_id=self._source_id,
        source_type=(
            self._source_type.value
            if isinstance(self._source_type, SourceType)
            else self._source_type
        ),
        source_name=source_name,
        platform=platform,
        platform_version=platform_version,
        firmware_version=firmware_version,
        hardware_revision=hardware_revision,
        serial_number=serial_number,
        fleet_id=fleet_id,
        site_id=site_id,
        zone_id=zone_id,
        organization_id=organization_id,
        tags=tags,
    )
    return self

def with_location(
    self,
    latitude: float | None = None,
    longitude: float | None = None,
    altitude_m: float | None = None,
    heading_deg: float | None = None,
    heading_reference: HeadingReference | None = None,
    coordinate_system: CoordinateSystem = CoordinateSystem.WGS84,
    horizontal_accuracy_m: float | None = None,
    vertical_accuracy_m: float | None = None,
    floor: int | None = None,
    building: str | None = None,
    local: LocalCoordinates | None = None,
    grid: GridPosition | None = None,
    semantic: SemanticLocation | None = None,
) -> Self:
    """Set location domain information."""
    self._location = LocationDomain(
        coordinate_system=coordinate_system,
        latitude=latitude,
        longitude=longitude,
        altitude_m=altitude_m,
        heading_deg=heading_deg,
        heading_reference=heading_reference,
        horizontal_accuracy_m=horizontal_accuracy_m,
        vertical_accuracy_m=vertical_accuracy_m,
        floor=floor,
        building=building,
        local=local,
        grid=grid,
        semantic=semantic,
    )
    return self

def with_local_coordinates(
    self,
    x_m: float,
    y_m: float,
    z_m: float = 0.0,
    yaw_deg: float | None = None,
    roll_deg: float | None = None,
    pitch_deg: float | None = None,
    coordinate_frame: str = "map",
) -> Self:
    """Set local coordinate position (convenience method)."""
    local = LocalCoordinates(
        coordinate_frame=coordinate_frame,
        x_m=x_m,
        y_m=y_m,
        z_m=z_m,
        roll_deg=roll_deg,
        pitch_deg=pitch_deg,
        yaw_deg=yaw_deg,
    )
    if self._location is None:
        self._location = LocationDomain(
            coordinate_system=CoordinateSystem.LOCAL,
            local=local,
        )
    else:
        self._location.local = local
    return self

def with_motion(
    self,
    speed_mps: float | None = None,
    vx_mps: float | None = None,
    vy_mps: float | None = None,
    vz_mps: float | None = None,
    angular_velocity_yaw_dps: float | None = None,
    angular_velocity_roll_dps: float | None = None,
    angular_velocity_pitch_dps: float | None = None,
    distance_traveled_m: float | None = None,
    distance_session_m: float | None = None,
    motion_state: MotionState | None = None,
) -> Self:
    """Set motion domain information."""
    linear_velocity = None
    if any(v is not None for v in [speed_mps, vx_mps, vy_mps, vz_mps]):
        linear_velocity = LinearVelocity(
            speed_mps=speed_mps,
            x_mps=vx_mps,
            y_mps=vy_mps,
            z_mps=vz_mps,
        )

    angular_velocity = None
    if any(
        v is not None
        for v in [
            angular_velocity_yaw_dps,
            angular_velocity_roll_dps,
            angular_velocity_pitch_dps,
        ]
    ):
        angular_velocity = AngularVelocity(
            yaw_dps=angular_velocity_yaw_dps,
            roll_dps=angular_velocity_roll_dps,
            pitch_dps=angular_velocity_pitch_dps,
        )

    odometry = None
    if any(v is not None for v in [distance_traveled_m, distance_session_m]):
        odometry = Odometry(
            distance_traveled_m=distance_traveled_m,
            distance_session_m=distance_session_m,
        )

    self._motion = MotionDomain(
        linear_velocity=linear_velocity,
        angular_velocity=angular_velocity,
        odometry=odometry,
        motion_state=motion_state,
    )
    return self

def with_power(
    self,
    battery_soc_pct: float | None = None,
    battery_soh_pct: float | None = None,
    battery_voltage_v: float | None = None,
    battery_current_a: float | None = None,
    battery_temperature_c: float | None = None,
    time_to_empty_min: int | None = None,
    is_charging: bool | None = None,
    is_plugged_in: bool | None = None,
    charging_power_w: float | None = None,
    charger_id: str | None = None,
    power_state: PowerState | None = None,
    power_consumption_w: float | None = None,
) -> Self:
    """Set power domain information."""
    battery = None
    if any(
        v is not None
        for v in [
            battery_soc_pct,
            battery_soh_pct,
            battery_voltage_v,
            battery_current_a,
            battery_temperature_c,
            time_to_empty_min,
        ]
    ):
        battery = Battery(
            state_of_charge_pct=battery_soc_pct,
            state_of_health_pct=battery_soh_pct,
            voltage_v=battery_voltage_v,
            current_a=battery_current_a,
            temperature_c=battery_temperature_c,
            time_to_empty_min=time_to_empty_min,
        )

    charging = None
    if any(
        v is not None
        for v in [is_charging, is_plugged_in, charging_power_w, charger_id]
    ):
        charging = Charging(
            is_charging=is_charging,
            is_plugged_in=is_plugged_in,
            charging_power_w=charging_power_w,
            charger_id=charger_id,
        )

    self._power = PowerDomain(
        battery=battery,
        charging=charging,
        power_state=power_state,
        power_consumption_w=power_consumption_w,
    )
    return self

def with_operational(
    self,
    mode: OperationalMode | None = None,
    state: OperationalState | None = None,
    sub_state: str | None = None,
    availability: str | None = None,
    enabled: bool | None = None,
    e_stop_active: bool | None = None,
    uptime_sec: int | None = None,
    task_id: str | None = None,
    task_type: str | None = None,
    task_state: str | None = None,
    task_progress_pct: float | None = None,
) -> Self:
    """Set operational domain information."""
    task = None
    if any(
        v is not None for v in [task_id, task_type, task_state, task_progress_pct]
    ):
        task = Task(
            task_id=task_id,
            task_type=task_type,
            task_state=task_state,
            task_progress_pct=task_progress_pct,
        )

    self._operational = OperationalDomain(
        mode=mode,
        state=state,
        sub_state=sub_state,
        availability=availability,
        enabled=enabled,
        e_stop_active=e_stop_active,
        uptime_sec=uptime_sec,
        task=task,
    )
    return self

def with_navigation(
    self,
    localization_status: LocalizationStatus | None = None,
    localization_confidence: float | None = None,
    localization_method: str | None = None,
    map_id: str | None = None,
    has_path: bool | None = None,
    path_length_m: float | None = None,
    path_blocked: bool | None = None,
    goal_id: str | None = None,
    goal_x_m: float | None = None,
    goal_y_m: float | None = None,
    distance_to_goal_m: float | None = None,
    nearest_obstacle_m: float | None = None,
    obstacle_count: int | None = None,
) -> Self:
    """Set navigation domain information."""
    localization = None
    if any(
        v is not None
        for v in [
            localization_status,
            localization_confidence,
            localization_method,
            map_id,
        ]
    ):
        localization = Localization(
            status=localization_status,
            confidence=localization_confidence,
            method=localization_method,
            map_id=map_id,
        )

    path = None
    if any(v is not None for v in [has_path, path_length_m, path_blocked]):
        path = PathStatus(
            has_path=has_path,
            path_length_m=path_length_m,
            path_blocked=path_blocked,
        )

    goal = None
    if any(
        v is not None for v in [goal_id, goal_x_m, goal_y_m, distance_to_goal_m]
    ):
        goal = NavigationGoal(
            goal_id=goal_id,
            goal_x_m=goal_x_m,
            goal_y_m=goal_y_m,
            distance_to_goal_m=distance_to_goal_m,
        )

    obstacles = None
    if any(v is not None for v in [nearest_obstacle_m, obstacle_count]):
        obstacles = Obstacles(
            nearest_obstacle_m=nearest_obstacle_m,
            obstacle_count=obstacle_count,
        )

    self._navigation = NavigationDomain(
        localization=localization,
        path=path,
        goal=goal,
        obstacles=obstacles,
    )
    return self

def with_safety(
    self,
    safety_state: SafetyState | None = None,
    e_stop_active: bool | None = None,
    e_stop_source: str | None = None,
    protective_stop_active: bool | None = None,
    reduced_speed_active: bool | None = None,
    zone_speed_limit_mps: float | None = None,
    nearest_human_m: float | None = None,
    nearest_robot_m: float | None = None,
    nearest_obstacle_m: float | None = None,
    collision_imminent: bool | None = None,
    safety_score: float | None = None,
    violations: list[SafetyViolation] | None = None,
) -> Self:
    """Set safety domain information."""
    e_stop = None
    if any(v is not None for v in [e_stop_active, e_stop_source]):
        e_stop = EStop(
            active=e_stop_active,
            source=e_stop_source,
        )

    safety_zones = None
    if any(
        v is not None
        for v in [
            protective_stop_active,
            reduced_speed_active,
            zone_speed_limit_mps,
        ]
    ):
        safety_zones = SafetyZones(
            protective_stop_active=protective_stop_active,
            reduced_speed_active=reduced_speed_active,
            zone_speed_limit_mps=zone_speed_limit_mps,
        )

    proximity = None
    if any(
        v is not None
        for v in [
            nearest_human_m,
            nearest_robot_m,
            nearest_obstacle_m,
            collision_imminent,
        ]
    ):
        proximity = Proximity(
            nearest_human_m=nearest_human_m,
            nearest_robot_m=nearest_robot_m,
            nearest_obstacle_m=nearest_obstacle_m,
            collision_imminent=collision_imminent,
        )

    self._safety = SafetyDomain(
        safety_state=safety_state,
        e_stop=e_stop,
        safety_zones=safety_zones,
        proximity=proximity,
        safety_score=safety_score,
        violations=violations,
    )
    return self

def with_detections(self, detections: list[Detection]) -> Self:
    """Add object detections to perception domain."""
    if self._perception is None:
        self._perception = PerceptionDomain(detections=detections)
    else:
        self._perception.detections = detections
    return self

def with_simulation(
    self,
    is_simulated: bool = True,
    simulator: str | None = None,
    scenario_id: str | None = None,
    scenario_name: str | None = None,
) -> Self:
    """Mark event as from simulation."""
    self._simulation = {
        "is_simulated": is_simulated,
        "simulator": simulator,
        "scenario_id": scenario_id,
        "scenario_name": scenario_name,
    }
    return self

def with_extensions(self, extensions: dict[str, Any]) -> Self:
    """Add vendor-specific extensions."""
    self._extensions = extensions
    return self

def with_raw_payload(self, raw_payload: dict[str, Any]) -> Self:
    """Add original platform data for debugging/audit."""
    self._raw_payload = raw_payload
    return self

def build(self) -> UDMEvent:
    """
    Build and return the UDM event.

    Returns:
        A fully constructed UDMEvent instance.
    """
    return UDMEvent(
        udm_version=UDM_VERSION,
        event_id=self._event_id or uuid7str(),
        event_type=self._event_type,
        source_id=self._source_id,
        source_type=self._source_type,
        captured_at=self._captured_at or datetime.utcnow(),
        session_id=self._session_id,
        sequence_num=self._sequence_num,
        identity=self._identity,
        location=self._location,
        motion=self._motion,
        power=self._power,
        operational=self._operational,
        navigation=self._navigation,
        perception=self._perception,
        safety=self._safety,
        simulation=self._simulation,
        extensions=self._extensions,
        raw_payload=self._raw_payload,
    )