feat(schema): add support of vehicle state (#31)

* feat: add new schema name for vehicle state

Signed-off-by: ktro2828 <[email protected]>

* feat: add schema table for vehicle state

Signed-off-by: ktro2828 <[email protected]>

* feat: update to load vehicle state in Tier4 class

Signed-off-by: ktro2828 <[email protected]>

* test: add unit testing for vehicle state

Signed-off-by: ktro2828 <[email protected]>

---------

Signed-off-by: ktro2828 <[email protected]>

t4_devkit/schema/name.py

@@ -25,6 +25,7 @@ class SchemaName(str, Enum):
         OBJECT_ANN (optional): The annotation of a foreground object in an image.
         SURFACE_ANN (optional): The annotation of a background object in an image.
         KEYPOINT (optional): The annotation of pose keypoints of an object in an image.
+        VEHICLE_STATE (optional): The annotation of ego vehicle states.
     """
 
     ATTRIBUTE = "attribute"
@@ -43,6 +44,7 @@ class SchemaName(str, Enum):
     OBJECT_ANN = "object_ann"  # optional
     SURFACE_ANN = "surface_ann"  # optional
     KEYPOINT = "keypoint"  # optional
+    VEHICLE_STATE = "vehicle_state"  # optional
 
     @property
     def filename(self) -> str:
@@ -63,4 +65,5 @@ def is_optional(self) -> bool:
             SchemaName.OBJECT_ANN,
             SchemaName.SURFACE_ANN,
             SchemaName.KEYPOINT,
+            SchemaName.VEHICLE_STATE,
         )

t4_devkit/schema/tables/__init__.py

@@ -15,4 +15,5 @@
 from .scene import *  # noqa
 from .sensor import *  # noqa
 from .surface_ann import *  # noqa
+from .vehicle_state import *  # noqa
 from .visibility import *  # noqa

t4_devkit/schema/tables/vehicle_state.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+from enum import Enum
+
+from attrs import define, field
+
+from ..name import SchemaName
+from .base import SchemaBase
+from .registry import SCHEMAS
+
+__all__ = ["ShiftState", "IndicatorState", "Indicators", "AdditionalInfo", "VehicleState"]
+
+
+class ShiftState(str, Enum):
+    """An enum to represent gear shift state."""
+
+    PARK = "PARK"
+    REVERSE = "REVERSE"
+    NEUTRAL = "NEUTRAL"
+    HIGH = "HIGH"
+    FORWARD = "FORWARD"
+    LOW = "LOW"
+    NONE = "NONE"
+
+
+class IndicatorState(str, Enum):
+    """An enum to represent indicator state."""
+
+    ON = "on"
+    OFF = "off"
+
+
+@define
+class Indicators:
+    """A dataclass to represent state of each indicator.
+
+    Attributes:
+        left (IndicatorState): State of the left indicator.
+        right (IndicatorState): State of the right indicator.
+        hazard (IndicatorState): State of the hazard lights.
+    """
+
+    left: IndicatorState = field(converter=IndicatorState)
+    right: IndicatorState = field(converter=IndicatorState)
+    hazard: IndicatorState = field(converter=IndicatorState)
+
+
+@define
+class AdditionalInfo:
+    """A dataclass to represent additional state information of the ego vehicle.
+
+    Attributes:
+        speed (float | None): Speed of the ego vehicle.
+    """
+
+    speed: float | None = field(default=None)
+
+
+@define(slots=False)
+@SCHEMAS.register(SchemaName.VEHICLE_STATE)
+class VehicleState(SchemaBase):
+    """A dataclass to represent schema table of `vehicle_state.json`.
+
+    Attributes:
+        token (str): Unique record identifier.
+        timestamp (int): Unix time stamp.
+        accel_pedal (float | None): Accel pedal position [%].
+        brake_pedal (float | None): Brake pedal position [%].
+        steer_pedal (float | None): Steering wheel position [%].
+        steering_tire_angle (float | None): Steering tire angle [rad].
+        steering_wheel_angle (float | None): Steering wheel angle [rad].
+        shift_state (ShiftState | None): Gear shift state.
+        indicators (Indicators | None): State of each indicator.
+        additional_info (AdditionalInfo | None): Additional state information.
+    """
+
+    token: str
+    timestamp: int
+    accel_pedal: float | None = field(default=None)
+    brake_pedal: float | None = field(default=None)
+    steer_pedal: float | None = field(default=None)
+    steering_tire_angle: float | None = field(default=None)
+    steering_wheel_angle: float | None = field(default=None)
+    shift_state: ShiftState | None = field(
+        default=None, converter=lambda x: None if x is None else ShiftState(x)
+    )
+    indicators: Indicators | None = field(
+        default=None, converter=lambda x: Indicators(**x) if isinstance(x, dict) else x
+    )
+    additional_info: AdditionalInfo | None = field(
+        default=None, converter=lambda x: AdditionalInfo(**x) if isinstance(x, dict) else x
+    )

t4_devkit/tier4.py

@@ -44,6 +44,7 @@
         SchemaTable,
         Sensor,
         SurfaceAnn,
+        VehicleState,
         Visibility,
     )
 
@@ -120,6 +121,7 @@ def __init__(self, version: str, data_root: str, verbose: bool = True) -> None:
         self.scene: list[Scene] = self.__load_table__(SchemaName.SCENE)
         self.sensor: list[Sensor] = self.__load_table__(SchemaName.SENSOR)
         self.surface_ann: list[SurfaceAnn] = self.__load_table__(SchemaName.SURFACE_ANN)
+        self.vehicle_state: list[VehicleState] = self.__load_table__(SchemaName.VEHICLE_STATE)
         self.visibility: list[Visibility] = self.__load_table__(SchemaName.VISIBILITY)
 
         # make reverse indexes for common lookups

tests/schema/conftest.py

@@ -328,3 +328,29 @@ def surface_ann_json(surface_ann_dict) -> Generator[str, Any, None]:
     with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
         save_json([surface_ann_dict], f.name)
         yield f.name
+
+
+# === VehicleState ===
+@pytest.fixture(scope="module")
+def vehicle_state_dict() -> dict:
+    """Return a dummy vehicle state as dictionary."""
+    return {
+        "token": "269572c280bd5cf9630ca542e6a60185",
+        "timestamp": 1724306784277396,
+        "accel_pedal": 0.0,
+        "brake_pedal": 1.0,
+        "steer_pedal": 0.6063521901837905,
+        "steering_tire_angle": 0.6063522100448608,
+        "steering_wheel_angle": 9.291000366210938,
+        "shift_state": "PARK",
+        "indicators": {"left": "off", "right": "on", "hazard": "off"},
+        "additional_info": {"speed": 0.0},
+    }
+
+
+@pytest.fixture(scope="module")
+def vehicle_state_json(vehicle_state_dict) -> Generator[str, Any, None]:
+    """Return a file path of dummy vehicle state record."""
+    with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
+        save_json([vehicle_state_dict], f.name)
+        yield f.name

tests/schema/tables/test_vehicle_state_table.py

@@ -0,0 +1,12 @@
+from t4_devkit.schema import VehicleState
+
+
+def test_vehicle_state_json(vehicle_state_json) -> None:
+    """Test loading vehicle state from a json file."""
+    _ = VehicleState.from_json(vehicle_state_json)
+
+
+def test_vehicle_state(vehicle_state_dict) -> None:
+    """Test loading vehicle state from a dictionary."""
+    s = VehicleState.from_dict(vehicle_state_dict)
+    print(s)

tests/schema/test_schema_builder.py

@@ -66,6 +66,11 @@ def test_build_surface_ann(surface_ann_json) -> None:
     _ = build_schema(SchemaName.SURFACE_ANN, surface_ann_json)
 
 
+def test_build_vehicle_state(vehicle_state_json) -> None:
+    """Test building vehicle state."""
+    _ = build_schema(SchemaName.VEHICLE_STATE, vehicle_state_json)
+
+
 def test_build_visibility(visibility_json) -> None:
     """Test building visibility."""
     _ = build_schema(SchemaName.VISIBILITY, visibility_json)

tests/schema/test_schema_name.py

@@ -22,6 +22,7 @@ def test_schema_name() -> None:
         "object_ann": True,
         "surface_ann": True,
         "keypoint": True,
+        "vehicle_state": True,
     }
 
     # check all enum members are covered by above names