Config implementation

Implementation of a JSON file that save all the Signal descriptor

bench-data-connector/src/bench_data_connector/__init__.py

@@ -13,6 +13,8 @@ from time import monotonic
 import threading
 import asyncio
 import logging
+from bench_data_connector.config import Config, read_config, DEFAULT_CONFIG
+from pydantic import BaseModel, Field
 from uuid import uuid4
 from queue import Queue
 
@@ -38,7 +40,7 @@ async def perform_handshake(
         protocol_version=PROTOCOL_VERSION,
         connector_name="Test connector",
         connector_version="V1",
-        signals=[signal_1, signal_2],
+        signals=[config.rpmhz_sig, config.lambda_sig],
     )
 
     try:
@@ -177,7 +179,6 @@ async def receive_messages(
 def threadmain() -> None:
     asyncio.run(jsonhandler())
 
-
 DATA_STRUCT = struct.Struct("<BIf")
 HOST = "127.0.0.1"
 PORT = 8765
@@ -192,6 +193,8 @@ signal_2 = JsonSignalDescriptor(
 
 data_queue: Queue[JsonValueDescriptor] = Queue()
 
+config: Config | None = None
+
 
 @dataclass
 class SensorData:
@@ -206,6 +209,13 @@ class SensorData:
 
 
 def main() -> None:
+    try:
+        config = read_config("~/.config/dynalab/connectors/bench_v1.json")
+    except FileNotFoundError:
+        config = DEFAULT_CONFIG
+        config.write("~/.config/dynalab/connectors/bench_v1.json")
+        print("No config found, default config written")
+
 
     JSONthread.start()
     try:
@@ -257,7 +267,7 @@ def main() -> None:
             packet = SensorData.from_bytes(payload)
             data_queue.put(
                 JsonValueDescriptor(
-                    signal_id=(signal_1.id if packet.data_type == 0 else signal_2.id),
+                    signal_id=(config.rpmhz_sig.id if packet.data_type == 0 else config.lambda_sig.id),
                     value=packet.value,
                     timestamp=sync.convert_timestamp(packet.timestamp * 1000),
                 )

bench-data-connector/src/bench_data_connector/config.py

@@ -0,0 +1,45 @@
+from pydantic import BaseModel, Field
+from dynalab_core.protocols.packets.handshake import JsonSignalDescriptor
+from pathlib import Path
+from typing import Self
+from uuid import uuid4
+
+
+
+class Config(BaseModel):
+    rpmhz_sig: JsonSignalDescriptor | None = None
+    lambda_sig: JsonSignalDescriptor | None = None
+    comments: str = ""
+
+    def write(self, file: str | Path) -> None:
+        output_path = Path(file).expanduser()
+
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+
+        output_path.write_text(
+            self.model_dump_json(indent=4),
+            encoding="utf-8",
+        )
+
+def read_config(file: str | Path) -> Config:
+        input_path = Path(file).expanduser()
+        if input_path.exists():
+            return Config.model_validate_json(input_path.read_text(encoding="utf-8"))
+        raise FileNotFoundError
+
+
+DEFAULT_CONFIG = Config(
+    rpmhz_sig=JsonSignalDescriptor(
+        id=uuid4(),
+        name="Rpmhz",
+        type="number",
+        timeout_ms=5000,
+    ),
+    lambda_sig=JsonSignalDescriptor(
+        id=uuid4(),
+        name="Lambda",
+        type="number",
+        timeout_ms=5000,
+    ),
+    comments="Mon test",
+)