Трохи пофіксив функцію post і провів тести post і websoket.

.gitignore

@@ -1,2 +0,0 @@
-agent/docker/mosquitto/data/
-agent/docker/mosquitto/log/

agent/src/file_datasource.py

@@ -1,9 +1,5 @@
-import csv
+from csv import reader
-import time
 from datetime import datetime
-from pathlib import Path
-from typing import Optional, List
-
 from domain.accelerometer import Accelerometer
 from domain.gps import Gps
 from domain.aggregated_data import AggregatedData
@@ -11,141 +7,24 @@ import config
 
 
 class FileDatasource:
-
+    def __init__(
-    def __init__(self, accelerometer_filename: str, gps_filename: str) -> None:
+        self,
-        self.accelerometer_filename = accelerometer_filename
+        accelerometer_filename: str,
-        self.gps_filename = gps_filename
+        gps_filename: str,
-
+    ) -> None:
-        self._acc_f = None
-        self._gps_f = None
-        self._acc_reader: Optional[csv.reader] = None
-        self._gps_reader: Optional[csv.reader] = None
-
-        self._started = False
-
-    def startReading(self, *args, **kwargs):
-        """Must be called before read()"""
-        if self._started:
-            return
-
-        if not Path(self.accelerometer_filename).exists():
-            raise FileNotFoundError(f"Accelerometer file not found: {self.accelerometer_filename}")
-        if not Path(self.gps_filename).exists():
-            raise FileNotFoundError(f"GPS file not found: {self.gps_filename}")
-
-        self._open_files()
-        self._started = True
-
-    def stopReading(self, *args, **kwargs):
-        """Must be called when finishing reading"""
-        self._close_files()
-        self._started = False
-
-    def read(self) -> AggregatedData:
-        """Return one combined sample (acc + gps)."""
-        if not self._started:
-            raise RuntimeError("Datasource is not started. Call startReading() before read().")
-
-        acc_row = self._get_next_row(self._acc_reader, source="acc")
-        gps_row = self._get_next_row(self._gps_reader, source="gps")
-
-        acc = self._parse_acc(acc_row)
-        gps = self._parse_gps(gps_row)
-
-        # IMPORTANT: timing belongs to datasource (not MQTT / main.py)
-        if config.DELAY and config.DELAY > 0:
-            time.sleep(float(config.DELAY))
-
-        return AggregatedData(
-            accelerometer=acc,
-            gps=gps,
-            timestamp=datetime.utcnow(),
-            user_id=config.USER_ID,
-        )
-
-    # ---------------- internal ----------------
-
-    def _open_files(self) -> None:
-        self._close_files()
-
-        self._acc_f = open(self.accelerometer_filename, "r", newline="", encoding="utf-8")
-        self._gps_f = open(self.gps_filename, "r", newline="", encoding="utf-8")
-
-        self._acc_reader = csv.reader(self._acc_f, skipinitialspace=True)
-        self._gps_reader = csv.reader(self._gps_f, skipinitialspace=True)
-
-        # File pointer is already at 0 right after open(), so no need to rewind here.
-        # Skip header row once.
-        next(self._acc_reader, None)
-        next(self._gps_reader, None)
-
-    def _close_files(self) -> None:
-        for f in (self._acc_f, self._gps_f):
-            try:
-                if f is not None:
-                    f.close()
-            except Exception:
         pass
 
-        self._acc_f = None
+    def read(self) -> AggregatedData:
-        self._gps_f = None
+        """Метод повертає дані отримані з датчиків"""
-        self._acc_reader = None
+        return AggregatedData(
-        self._gps_reader = None
+            Accelerometer(1, 2, 3),
+            Gps(4, 5),
+            datetime.now(),
+            config.USER_ID,
+        )
 
-    def _rewind_acc(self) -> None:
+    def startReading(self, *args, **kwargs):
-        if self._acc_f is None:
+        """Метод повинен викликатись перед початком читання даних"""
-            raise RuntimeError("Accelerometer file is not open.")
-        self._acc_f.seek(0)
-        self._acc_reader = csv.reader(self._acc_f, skipinitialspace=True)
-        next(self._acc_reader, None)  # skip header row
 
-    def _rewind_gps(self) -> None:
+    def stopReading(self, *args, **kwargs):
-        if self._gps_f is None:
+        """Метод повинен викликатись для закінчення читання даних"""
-            raise RuntimeError("GPS file is not open.")
-        self._gps_f.seek(0)
-        self._gps_reader = csv.reader(self._gps_f, skipinitialspace=True)
-        next(self._gps_reader, None)  # skip header row
-
-    def _get_next_row(self, reader, source: str) -> List[str]:
-        """Get the next valid row from the reader."""
-        if reader is None:
-            raise RuntimeError("Reader is not initialized.")
-
-        while True:
-            row = next(reader, None)
-            if row is None:
-                # EOF -> rewind & continue
-                if source == "acc":
-                    self._rewind_acc()
-                    reader = self._acc_reader
-                else:
-                    self._rewind_gps()
-                    reader = self._gps_reader
-                continue
-
-            if not row or not any(cell.strip() for cell in row):
-                continue
-
-            return row
-
-    @staticmethod
-    def _parse_acc(row: List[str]) -> Accelerometer:
-        if len(row) < 3:
-            raise ValueError(f"Accelerometer row must have 3 values (x,y,z). Got: {row}")
-
-        try:
-            x = int(row[0])
-            y = int(row[1])
-            z = int(row[2])
-        except ValueError as e:
-            raise ValueError(f"Invalid accelerometer values (expected integers): {row}") from e
-
-        return Accelerometer(x=x, y=y, z=z)
-
-    @staticmethod
-    def _parse_gps(row: List[str]) -> Gps:
-        if len(row) < 2:
-            raise ValueError(f"GPS row must have 2 values (longitude,latitude). Got: {row}")
-        lon = float(row[0])
-        lat = float(row[1])
-        return Gps(longitude=lon, latitude=lat)

agent/src/main.py

@@ -1,4 +1,6 @@
 from paho.mqtt import client as mqtt_client
+import json
+import time
 from schema.aggregated_data_schema import AggregatedDataSchema
 from file_datasource import FileDatasource
 import config
@@ -22,14 +24,19 @@ def connect_mqtt(broker, port):
     return client
 
 
-def publish(client, topic, datasource):
+def publish(client, topic, datasource, delay):
     datasource.startReading()
     while True:
+        time.sleep(delay)
         data = datasource.read()
         msg = AggregatedDataSchema().dumps(data)
         result = client.publish(topic, msg)
+        # result: [0, 1]
         status = result[0]
-        if status != 0:
+        if status == 0:
+            pass
+            # print(f"Send `{msg}` to topic `{topic}`")
+        else:
             print(f"Failed to send message to topic {topic}")
 
 
@@ -37,9 +44,9 @@ def run():
     # Prepare mqtt client
     client = connect_mqtt(config.MQTT_BROKER_HOST, config.MQTT_BROKER_PORT)
     # Prepare datasource
-    datasource = FileDatasource("data/accelerometer.csv", "data/gps.csv")
+    datasource = FileDatasource("data/data.csv", "data/gps_data.csv")
     # Infinity publish data
-    publish(client, config.MQTT_TOPIC, datasource)
+    publish(client, config.MQTT_TOPIC, datasource, config.DELAY)
 
 
 if __name__ == "__main__":

store/main.py

@@ -2,6 +2,7 @@ import asyncio
 import json
 from typing import Set, Dict, List, Any
 from fastapi import FastAPI, HTTPException, WebSocket, WebSocketDisconnect, Body
+from fastapi.encoders import jsonable_encoder
 from sqlalchemy import (
     create_engine,
     MetaData,
@@ -125,10 +126,36 @@ async def send_data_to_subscribers(user_id: int, data):
 
 
 @app.post("/processed_agent_data/")
-async def create_processed_agent_data(data: List[ProcessedAgentData]):
+async def create_processed_agent_data(data: List[ProcessedAgentData], user_id: int = Body(..., embed=True)):
-    # Insert data to database
+    session = SessionLocal()
-    # Send data to subscribers
+    try:
-    pass
+        created_data = [
+            {
+                "road_state": item.road_state,
+                "user_id": user_id,
+                "x": item.agent_data.accelerometer.x,
+                "y": item.agent_data.accelerometer.y,
+                "z": item.agent_data.accelerometer.z,
+                "latitude": item.agent_data.gps.latitude,
+                "longitude": item.agent_data.gps.longitude,
+                "timestamp": item.agent_data.timestamp,
+            }
+            for item in data
+        ]
+        stmt = processed_agent_data.insert().values(created_data).returning(processed_agent_data)
+        result = session.execute(stmt)
+        created_records = [dict(row._mapping) for row in result.fetchall()]
+        session.commit()
+
+        for record in created_records:
+            await send_data_to_subscribers(user_id, jsonable_encoder(record))
+        return created_records
+    except Exception as err:
+        session.rollback()
+        print(f"Database error: {err}")
+        raise HTTPException(status_code=500, detail="Internal Server Error")
+    finally:
+        session.close()
 
 
 @app.get(