Support Saving Visual Data (#313)

* Added visual headers. (#307)

* Added header detection. (#307)

leads/data_persistence/core.py

@@ -4,7 +4,7 @@
     override as _override, Self as _Self, Iterator as _Iterator, Callable as _Callable, Iterable as _Iterable, \
     Generator as _Generator, Any as _Any
 
-from leads.types import Compressor as _Compressor
+from leads.types import Compressor as _Compressor, VisualHeader as _VisualHeader, VisualHeaderFull as _VisualHeaderFull
 from ._computational import mean as _mean, array as _array, norm as _norm, read_csv as _read_csv, \
     DataFrame as _DataFrame, TextFileReader as _TextFileReader
 
@@ -259,3 +259,9 @@ def close(self) -> None:
 DEFAULT_HEADER_FULL: tuple[str, str, str, str, str, str, str, str, str, str, str, str, str, str] = DEFAULT_HEADER + (
     "throttle", "brake"
 )
+VISUAL_HEADER_ONLY: tuple[str, str, str, str, str, str, str, str] = (
+    "front_view_base64", "front_view_latency", "left_view_base64", "left_view_latency", "right_view_base64",
+    "front_view_latency", "rear_view_base64", "rear_view_latency"
+)
+VISUAL_HEADER: _VisualHeader = DEFAULT_HEADER + VISUAL_HEADER_ONLY
+VISUAL_HEADER_FULL: _VisualHeaderFull = DEFAULT_HEADER_FULL + VISUAL_HEADER_ONLY

leads/types.py

@@ -6,3 +6,7 @@
 type OnRegisterChain[T] = _Callable[[OnRegister[T]], OnRegister[T]]
 type SupportedConfigValue = bool | int | float | str | None
 type SupportedConfig = SupportedConfigValue | tuple[SupportedConfig, ...]
+type VisualHeader = tuple[
+    str, str, str, str, str, str, str, str, str, str, str, str, str, str, str, str, str, str, str, str]
+type VisualHeaderFull = tuple[
+    str, str, str, str, str, str, str, str, str, str, str, str, str, str, str, str, str, str, str, str, str, str]

leads_vec_rc/cli.py

@@ -11,7 +11,7 @@
 
 from leads import require_config, L, DataContainer
 from leads.comm import Service, Client, start_client, create_client, Callback, Connection, ConnectionBase
-from leads.data_persistence import DataPersistence, Vector, CSV, DEFAULT_HEADER_FULL
+from leads.data_persistence import DataPersistence, Vector, CSV, DEFAULT_HEADER_FULL, VISUAL_HEADER_FULL
 from leads_gui import Config
 
 config: Config = require_config()
@@ -24,8 +24,17 @@
 acceleration_record: DataPersistence[float] = DataPersistence(2000)
 voltage_record: DataPersistence[float] = DataPersistence(2000)
 gps_record: DataPersistence[Vector[float]] = DataPersistence(2000)
-csv = CSV(f"{config.data_dir}/{datetime.now().strftime("%Y-%m-%d_%H-%M-%S")}.csv", DEFAULT_HEADER_FULL,
-          time_stamp_record, voltage_record, speed_record)
+csv: CSV | None = None
+
+
+def try_create_csv(data: dict[str, Any]) -> None:
+    global csv
+    if csv:
+        return
+    csv = CSV(f"{config.data_dir}/{datetime.now().strftime("%Y-%m-%d_%H-%M-%S")}.csv",
+              VISUAL_HEADER_FULL if set(VISUAL_HEADER_FULL).issubset(data.keys()) else DEFAULT_HEADER_FULL,
+              time_stamp_record, voltage_record, speed_record)
+    register(csv.close)
 
 
 def retry(service: Service) -> Client:
@@ -59,6 +68,7 @@ def on_receive(self, service: Service, msg: bytes) -> None:
             acceleration_record.append(Vector(d["forward_acceleration"], d["lateral_acceleration"]))
             gps_record.append(Vector(d["latitude"], d["longitude"]))
             if config.save_data:
+                try_create_csv(d)
                 csv.write_frame(*(d[key] for key in csv.header()))
             else:
                 time_stamp_record.append(int(d["t"]))
@@ -128,6 +138,3 @@ async def m1() -> str:
 async def m3() -> str:
     callback.client.send(b"m3")
     return "done"
-
-
-register(csv.close)