lab 5: implement set_bump_marker()

SCRUM-77 Post method from Hub to Store

.dockerignore

@@ -0,0 +1,23 @@
+# IDEs
+.idea/
+.vscode/
+.git/
+.gitignore
+.dockerignore
+.DS_Store
+Thumbs.db
+
+# Python
+**/__pycache__/
+**/*.py[cod]
+**/*.pyo
+**/*.pyd
+venv/
+.env
+
+# Logs & Database & Broker data
+*.log
+**/mosquitto/data/
+**/mosquitto/log/
+**/postgres_data/
+**/pgadmin-data/

.gitignore

@@ -0,0 +1,25 @@
+# IDEs
+.idea/
+.vscode/
+*.swp
+*.swo
+
+# Python
+venv/
+__pycache__/
+*.py[cod]
+*$py.class
+.env
+
+# Logs
+*.log
+app.log
+
+# Database & Broker data
+**/mosquitto/data/
+**/mosquitto/log/
+**/postgres_data/
+**/pgadmin-data/
+
+# OS specific
+.DS_Store

MapView/.gitignore

@@ -1,3 +0,0 @@
-.idea
-venv
-__pycache__

MapView/domain/accelerometer.py

@@ -0,0 +1,7 @@
+from dataclasses import dataclass
+
+@dataclass
+class Accelerometer:
+  x: int
+  y: int
+  z: int

MapView/fileReader.py

@@ -0,0 +1,50 @@
+from csv import reader
+import config
+from domain.accelerometer import Accelerometer
+
+
+
+class FileReader:
+    def __init__(
+        self, data_filename: str,
+    ) -> None:
+        self.file_path = data_filename
+        pass
+
+    def read(self):
+        return self.getNextValue()
+
+    def startReading(self, *args, **kwargs):
+        self.file = open(self.file_path, newline='')
+        self.file_reader = reader(self.file, skipinitialspace=True)
+        file_header = next(self.file_reader)
+        
+        self.x_idx = file_header.index('X')
+        self.y_idx = file_header.index('Y')
+        self.z_idx = file_header.index('Z')
+
+    def getNextValue(self):
+        while True:
+            row = next(self.file_reader, None)
+            if row is None:
+                self._rewind_file()
+                continue
+            try:
+                x = int(row[self.x_idx])
+                y = int(row[self.y_idx])
+                z = int(row[self.z_idx])
+                return Accelerometer(x=x, y=y, z=z)
+            except Exception as e:
+                continue
+        
+    def _rewind_file(self):
+        self.file.seek(0)
+        self.file_reader = reader(self.file)
+        next(self.file_reader)
+        
+    def stopReading(self, *args, **kwargs):
+        if self.file:
+            self.file.close()
+        self.file_reader = None
+        
+

MapView/main.py

@@ -8,31 +8,93 @@ from datasource import Datasource
 
 class MapViewApp(App):
     def __init__(self, **kwargs):
-        super().__init__()
+        super().__init__(**kwargs)
+
+        self.mapview = None
+        self.datasource = Datasource(user_id=1)
+        self.line_layer = None
+        self.car_marker = None
+
         # додати необхідні змінні
+        self.pothole_markers = []
 
     def on_start(self):
         """
         Встановлює необхідні маркери, викликає функцію для оновлення мапи
         """
+        Clock.schedule_interval(self.update, 0.3)
 
     def update(self, *args):
         """
         Викликається регулярно для оновлення мапи
         """
+        new_points = self.datasource.get_new_points()
 
+        if not new_points:
+            return
+
+        for point in new_points:
+
+            lat, lon, road_state = point
+
+            # Оновлює лінію маршрута
+            self.line_layer.add_point((lat, lon))
+
+            # Оновлює маркер маниши
+            self.update_car_marker((lat, lon))
+
+            # Перевіряємо стан дороги
+            self.check_road_quality(point)
+
+    def check_road_quality(self, point):
+        """
+        Аналізує дані акселерометра для подальшого визначення
+        та відображення ям та лежачих поліцейських
+        """
+        if len(point) < 3:
+            return
+
+        lat, lon, road_state = point
+
+        if road_state == "pothole":
+            self.set_pothole_marker((lat, lon))
+        elif road_state == "bump":
+            self.set_bump_marker((lat, lon))
 
     def update_car_marker(self, point):
         """
         Оновлює відображення маркера машини на мапі
         :param point: GPS координати
         """
+        lat, lon = point[0], point[1]
+
+        if not hasattr(self, 'car_marker'):
+            self.car_marker = MapMarker(lat=lat, lon=lon, source='./images/car')
+            self.mapview.add_marker(self.car_marker)
+        else:
+            self.car_marker.lat = lat
+            self.car_marker.lon = lon
+
+        self.mapview.center_on(lat, lon)
 
     def set_pothole_marker(self, point):
-        """
+        if isinstance(point, dict):
-        Встановлює маркер для ями
+            lat = point.get("lat")
-        :param point: GPS координати
+            lon = point.get("lon")
-        """
+        else:
+            lat, lon = point
+
+        if lat is None or lon is None:
+            return
+
+        marker = MapMarker(
+            lat=lat,
+            lon=lon,
+            source="images/pothole.png"
+        )
+
+        self.mapview.add_marker(marker)
+        self.pothole_markers.append(marker)
 
     def set_bump_marker(self, point):
         """
@@ -45,7 +107,15 @@ class MapViewApp(App):
         Ініціалізує мапу MapView(zoom, lat, lon)
         :return: мапу
         """
-        self.mapview = MapView()
+        self.mapview = MapView(
+            zoom=15,
+            lat=50.4501,
+            lon=30.5234
+        )
+
+        self.line_layer = LineMapLayer()
+        self.mapview.add_layer(self.line_layer)
+
         return self.mapview
 
 

agent/.gitignore

@@ -1,2 +0,0 @@
-venv
-__pycache__

agent/Dockerfile

@@ -3,10 +3,10 @@ FROM python:latest
 # set the working directory in the container
 WORKDIR /usr/agent
 # copy the dependencies file to the working directory
-COPY requirements.txt .
+COPY agent/requirements.txt .
 # install dependencies
-RUN pip install -r requirements.txt
+RUN pip install --no-cache-dir -r requirements.txt
 # copy the content of the local src directory to the working directory
-COPY src/ .
+COPY agent/src/ .
 # command to run on container start
 CMD ["python", "main.py"]

agent/docker/docker-compose.yaml

@@ -16,7 +16,9 @@ services:
 
   fake_agent:
     container_name: agent
-    build: ../
+    build:
+      context: ../../
+      dockerfile: agent/Dockerfile
     depends_on:
       - mqtt
     environment:

agent/src/data/parking.csv

@@ -0,0 +1,22 @@
+longitude,latitude,empty_count
+50.450386085935094,30.524547100067142,10
+50.450386085935094,30.524547100067142,11
+50.450386085935094,30.524547100067142,13
+50.450386085935094,30.524547100067142,15
+50.450386085935094,30.524547100067142,7
+50.450386085935094,30.524547100067142,9
+50.450386085935094,30.524547100067142,4
+50.450386085935094,30.524547100067142,0
+50.450386085935094,30.524547100067142,0
+50.450386085935094,30.524547100067142,3
+50.450386085935094,30.524547100067142,4
+50.450069433207545,30.52406822530458,16
+50.450069433207545,30.52406822530458,20
+50.450069433207545,30.52406822530458,25
+50.450069433207545,30.52406822530458,30
+50.450069433207545,30.52406822530458,29
+50.450069433207545,30.52406822530458,12
+50.450069433207545,30.52406822530458,10
+50.450069433207545,30.52406822530458,14
+50.450069433207545,30.52406822530458,3
+50.450069433207545,30.52406822530458,2

agent/src/domain/aggregated_data.py

@@ -1,13 +1,16 @@
 from dataclasses import dataclass
 
 from datetime import datetime
+
 from domain.accelerometer import Accelerometer
 from domain.gps import Gps
+from domain.parking import Parking
 
 
 @dataclass
 class AggregatedData:
     accelerometer: Accelerometer
     gps: Gps
+    parking: Parking
     timestamp: datetime
     user_id: int

agent/src/domain/parking.py

@@ -0,0 +1,9 @@
+from dataclasses import dataclass
+
+from domain.gps import Gps
+
+
+@dataclass
+class Parking:
+    empty_count: int
+    gps: Gps

agent/src/file_datasource.py

@@ -1,5 +1,10 @@
-from csv import reader
+import csv
+import time
 from datetime import datetime
+from pathlib import Path
+from typing import Optional, List
+
+from domain.parking import Parking
 from domain.accelerometer import Accelerometer
 from domain.gps import Gps
 from domain.aggregated_data import AggregatedData
@@ -7,24 +12,185 @@ import config
 
 
 class FileDatasource:
-    def __init__(
-        self,
-        accelerometer_filename: str,
-        gps_filename: str,
-    ) -> None:
-        pass
 
-    def read(self) -> AggregatedData:
+    def __init__(
-        """Метод повертає дані отримані з датчиків"""
+            self,
-        return AggregatedData(
+            accelerometer_filename: str,
-            Accelerometer(1, 2, 3),
+            gps_filename: str,
-            Gps(4, 5),
+            park_filename: str,
-            datetime.now(),
+    ) -> None:
-            config.USER_ID,
+
-        )
+        self.accelerometer_filename = accelerometer_filename
+        self.park_filename = park_filename
+        self.gps_filename = gps_filename
+
+        self._park_f = None
+        self._acc_f = None
+        self._gps_f = None
+
+        self._park_reader: Optional[csv.reader] = 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.park_filename).exists():
+            raise FileNotFoundError(f"Accelerometer file not found: {self.park_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")
+        park_row = self._get_next_row(self._park_reader, source="park")
+        gps_row = self._get_next_row(self._gps_reader, source="gps")
+
+        acc = self._parse_acc(acc_row)
+        park = self._parse_park(park_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,
+            parking=park,
+            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._park_f = open(self.park_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._park_reader = csv.reader(self._park_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._park_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
+        self._park_f = None
+        self._gps_f = None
+        self._acc_reader = None
+        self._park_reader = None
+        self._gps_reader = None
+
+    def _rewind_acc(self) -> None:
+        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:
+        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 _rewind_park(self) -> None:
+        if self._park_f is None:
+            raise RuntimeError("GPS file is not open.")
+        self._park_f.seek(0)
+        self._park_reader = csv.reader(self._park_f, skipinitialspace=True)
+        next(self._park_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
+
+                elif source == 'park':
+                    self._rewind_park()
+                    reader = self._park_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)
+
+    @staticmethod
+    def _parse_park(row: List[str]) -> Parking:
+        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])
+        empty_count = int(row[2])
+
+        return Parking(
+            gps=Gps(longitude=lon, latitude=lat),
+            empty_count=empty_count
+        )

agent/src/main.py

@@ -1,6 +1,4 @@
 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
@@ -24,19 +22,14 @@ def connect_mqtt(broker, port):
     return client
 
 
-def publish(client, topic, datasource, delay):
+def publish(client, topic, datasource):
     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}")
 
 
@@ -44,9 +37,9 @@ def run():
     # Prepare mqtt client
     client = connect_mqtt(config.MQTT_BROKER_HOST, config.MQTT_BROKER_PORT)
     # Prepare datasource
-    datasource = FileDatasource("data/data.csv", "data/gps_data.csv")
+    datasource = FileDatasource("data/accelerometer.csv", "data/gps.csv", "data/parking.csv")
     # Infinity publish data
-    publish(client, config.MQTT_TOPIC, datasource, config.DELAY)
+    publish(client, config.MQTT_TOPIC, datasource)
 
 
 if __name__ == "__main__":

agent/src/schema/aggregated_data_schema.py

@@ -1,10 +1,12 @@
 from marshmallow import Schema, fields
 from schema.accelerometer_schema import AccelerometerSchema
 from schema.gps_schema import GpsSchema
+from schema.parking_schema import ParkingSchema
 
 
 class AggregatedDataSchema(Schema):
     accelerometer = fields.Nested(AccelerometerSchema)
     gps = fields.Nested(GpsSchema)
+    parking = fields.Nested(ParkingSchema)
     timestamp = fields.DateTime("iso")
     user_id = fields.Int()

agent/src/schema/parking_schema.py

@@ -0,0 +1,8 @@
+from marshmallow import Schema, fields
+
+from schema.gps_schema import GpsSchema
+
+
+class ParkingSchema(Schema):
+    gps = fields.Nested(GpsSchema)
+    empty_count = fields.Int()

edge/.gitignore

@@ -1,2 +0,0 @@
-venv
-app.log

edge/Dockerfile

@@ -3,9 +3,9 @@ FROM python:3.9-slim
 # Set the working directory inside the container
 WORKDIR /app
 # Copy the requirements.txt file and install dependencies
-COPY requirements.txt .
+COPY edge/requirements.txt .
 RUN pip install --no-cache-dir -r requirements.txt
 # Copy the entire application into the container
-COPY . .
+COPY edge/. .
 # Run the main.py script inside the container when it starts
 CMD ["python", "main.py"]

edge/docker/docker-compose.yaml

@@ -17,7 +17,9 @@ services:
 
   edge:
     container_name: edge
-    build: ../
+    build:
+      context: ../../
+      dockerfile: edge/Dockerfile
     depends_on:
       - mqtt
     environment:

hub/.gitignore

@@ -1,2 +0,0 @@
-venv
-__pycache__

hub/Dockerfile

@@ -3,9 +3,10 @@ FROM python:3.9-slim
 # Set the working directory inside the container
 WORKDIR /app
 # Copy the requirements.txt file and install dependencies
-COPY requirements.txt .
+COPY hub/requirements.txt .
+
 RUN pip install --no-cache-dir -r requirements.txt
 # Copy the entire application into the container
-COPY . .
+COPY hub/. .
 # Run the main.py script inside the container when it starts
 CMD ["uvicorn", "main:app", "--host", "0.0.0.0"]

hub/app/adapters/store_api_adapter.py

@@ -14,11 +14,30 @@ class StoreApiAdapter(StoreGateway):
         self.api_base_url = api_base_url
 
     def save_data(self, processed_agent_data_batch: List[ProcessedAgentData]):
-        """
+        if not processed_agent_data_batch:
-        Save the processed road data to the Store API.
+            return False
-        Parameters:
+
-            processed_agent_data_batch (dict): Processed road data to be saved.
+        # Extract user_id from the first element
-        Returns:
+        user_id = processed_agent_data_batch[0].agent_data.user_id
-            bool: True if the data is successfully saved, False otherwise.
+
-        """
+        payload = {
-        # Implement it
+            "data": [item.model_dump(mode='json') for item in processed_agent_data_batch],
+            "user_id": user_id
+        }
+
+        try:
+            # Perform a POST request to the Store API with a 10-second timeout
+            response = requests.post(
+                f"{self.api_base_url}/processed_agent_data/",
+                json=payload,
+                timeout=10
+            )
+            if response.status_code == 200:
+                logging.info(f"Batch of {len(processed_agent_data_batch)} items sent to Store.")
+                return True
+            else:
+                logging.error(f"Store API error: {response.status_code} - {response.text}")
+                return False
+        except Exception as e:
+            logging.error(f"Failed to send data to Store: {e}")
+            return False

hub/docker/docker-compose.yaml

@@ -1,4 +1,3 @@
-version: "3.9"
 name: "road_vision__hub"
 services:
   mqtt:
@@ -16,7 +15,7 @@ services:
 
 
   postgres_db:
-    image: postgres:latest
+    image: postgres:17
     container_name: postgres_db
     restart: always
     environment:
@@ -49,7 +48,9 @@ services:
 
   store:
     container_name: store
-    build: ../../store
+    build:
+      context: ../../
+      dockerfile: store/Dockerfile
     depends_on:
       - postgres_db
     restart: always
@@ -77,7 +78,9 @@ services:
 
   hub:
     container_name: hub
-    build: ../
+    build:
+      context: ../../
+      dockerfile: hub/Dockerfile
     depends_on:
       - mqtt
       - redis
@@ -90,7 +93,7 @@ services:
       MQTT_BROKER_HOST: "mqtt"
       MQTT_BROKER_PORT: 1883
       MQTT_TOPIC: "processed_data_topic"
-      BATCH_SIZE: 1
+      BATCH_SIZE: 20
     ports:
       - "9000:8000"
     networks:

hub/main.py

@@ -70,18 +70,20 @@ def on_message(client, userdata, msg):
         processed_agent_data = ProcessedAgentData.model_validate_json(
             payload, strict=True
         )
-
         redis_client.lpush(
             "processed_agent_data", processed_agent_data.model_dump_json()
         )
-        processed_agent_data_batch: List[ProcessedAgentData] = []
+
         if redis_client.llen("processed_agent_data") >= BATCH_SIZE:
+            processed_agent_data_batch: List[ProcessedAgentData] = []
             for _ in range(BATCH_SIZE):
-                processed_agent_data = ProcessedAgentData.model_validate_json(
+                raw_data = redis_client.lpop("processed_agent_data")
-                    redis_client.lpop("processed_agent_data")
+                if raw_data:
-                )
+                    data_item = ProcessedAgentData.model_validate_json(raw_data)
-                processed_agent_data_batch.append(processed_agent_data)
+                    processed_agent_data_batch.append(data_item)
-        store_adapter.save_data(processed_agent_data_batch=processed_agent_data_batch)
+
+            store_adapter.save_data(processed_agent_data_batch=processed_agent_data_batch)
+
         return {"status": "ok"}
     except Exception as e:
         logging.info(f"Error processing MQTT message: {e}")

store/.gitignore

@@ -1,3 +0,0 @@
-venv
-__pycache__
-.idea

store/Dockerfile

@@ -3,9 +3,10 @@ FROM python:latest
 # Set the working directory inside the container
 WORKDIR /app
 # Copy the requirements.txt file and install dependencies
-COPY requirements.txt .
+COPY store/requirements.txt .
+
 RUN pip install --no-cache-dir -r requirements.txt
 # Copy the entire application into the container
-COPY . .
+COPY store/. .
 # Run the main.py script inside the container when it starts
 CMD ["uvicorn", "main:app", "--host", "0.0.0.0"]

store/database.py

@@ -0,0 +1,15 @@
+from sqlalchemy import MetaData
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker, declarative_base
+
+from config import POSTGRES_USER, POSTGRES_PASSWORD, POSTGRES_HOST, POSTGRES_PORT, POSTGRES_DB
+
+
+DATABASE_URL = f"postgresql+psycopg2://{POSTGRES_USER}:{POSTGRES_PASSWORD}@{POSTGRES_HOST}:{POSTGRES_PORT}/{POSTGRES_DB}"
+engine = create_engine(DATABASE_URL)
+
+Base = declarative_base()
+
+metadata = MetaData()
+
+SessionLocal = sessionmaker(bind=engine)

store/docker/docker-compose.yaml

@@ -1,8 +1,7 @@
-version: "3.9"
 name: "road_vision__database"
 services:
   postgres_db:
-    image: postgres:latest
+    image: postgres:17
     container_name: postgres_db
     restart: always
     environment:
@@ -35,7 +34,9 @@ services:
 
   store:
     container_name: store
-    build: ..
+    build:
+      context: ../../
+      dockerfile: store/Dockerfile
     depends_on:
       - postgres_db
     restart: always

store/main.py

@@ -1,10 +1,8 @@
-import asyncio
 import json
-from typing import Set, Dict, List, Any
+from typing import Set, Dict, List
 from fastapi import FastAPI, HTTPException, WebSocket, WebSocketDisconnect, Body
+from fastapi.encoders import jsonable_encoder
 from sqlalchemy import (
-    create_engine,
-    MetaData,
     Table,
     Column,
     Integer,
@@ -12,25 +10,14 @@ from sqlalchemy import (
     Float,
     DateTime,
 )
-from sqlalchemy.orm import sessionmaker
 from sqlalchemy.sql import select
-from datetime import datetime
+
-from pydantic import BaseModel, field_validator
+from database import metadata, SessionLocal
-from config import (
+from schemas import ProcessedAgentData, ProcessedAgentDataInDB
-    POSTGRES_HOST,
-    POSTGRES_PORT,
-    POSTGRES_DB,
-    POSTGRES_USER,
-    POSTGRES_PASSWORD,
-)
 
 # FastAPI app setup
 app = FastAPI()
-# SQLAlchemy setup
+
-DATABASE_URL = f"postgresql+psycopg2://{POSTGRES_USER}:{POSTGRES_PASSWORD}@{POSTGRES_HOST}:{POSTGRES_PORT}/{POSTGRES_DB}"
-engine = create_engine(DATABASE_URL)
-metadata = MetaData()
-# Define the ProcessedAgentData table
 processed_agent_data = Table(
     "processed_agent_data",
     metadata,
@@ -44,57 +31,6 @@ processed_agent_data = Table(
     Column("longitude", Float),
     Column("timestamp", DateTime),
 )
-SessionLocal = sessionmaker(bind=engine)
-
-
-# SQLAlchemy model
-class ProcessedAgentDataInDB(BaseModel):
-    id: int
-    road_state: str
-    user_id: int
-    x: float
-    y: float
-    z: float
-    latitude: float
-    longitude: float
-    timestamp: datetime
-
-
-# FastAPI models
-class AccelerometerData(BaseModel):
-    x: float
-    y: float
-    z: float
-
-
-class GpsData(BaseModel):
-    latitude: float
-    longitude: float
-
-
-class AgentData(BaseModel):
-    user_id: int
-    accelerometer: AccelerometerData
-    gps: GpsData
-    timestamp: datetime
-
-    @classmethod
-    @field_validator("timestamp", mode="before")
-    def check_timestamp(cls, value):
-        if isinstance(value, datetime):
-            return value
-        try:
-            return datetime.fromisoformat(value)
-        except (TypeError, ValueError):
-            raise ValueError(
-                "Invalid timestamp format. Expected ISO 8601 format (YYYY-MM-DDTHH:MM:SSZ)."
-            )
-
-
-class ProcessedAgentData(BaseModel):
-    road_state: str
-    agent_data: AgentData
-
 
 # WebSocket subscriptions
 subscriptions: Dict[int, Set[WebSocket]] = {}
@@ -125,10 +61,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(
@@ -136,14 +98,34 @@ async def create_processed_agent_data(data: List[ProcessedAgentData]):
     response_model=ProcessedAgentDataInDB,
 )
 def read_processed_agent_data(processed_agent_data_id: int):
-    # Get data by id
+    session = SessionLocal()
-    pass
+    try:
+        stmt = select(processed_agent_data).where(
+            processed_agent_data.c.id == processed_agent_data_id
+        )
+        res = session.execute(stmt).fetchone()
+        if not res:
+            raise HTTPException(status_code=404, detail="Not found")
+
+        return dict(res._mapping)
+
+    finally:
+        session.close()
 
 
 @app.get("/processed_agent_data/", response_model=list[ProcessedAgentDataInDB])
 def list_processed_agent_data():
-    # Get list of data
+    session = SessionLocal()
-    pass
+    try:
+        stmt = select(processed_agent_data)
+        res = session.execute(stmt).fetchall()
+        if not res:
+            raise HTTPException(status_code=404, detail="Not found")
+
+        return [dict(r._mapping) for r in res]
+
+    finally:
+        session.close()
 
 
 @app.put(
@@ -152,7 +134,41 @@ def list_processed_agent_data():
 )
 def update_processed_agent_data(processed_agent_data_id: int, data: ProcessedAgentData):
     # Update data
-    pass
+    session = SessionLocal()
+
+    try:
+        query = select(processed_agent_data).where(
+            processed_agent_data.c.id == processed_agent_data_id
+        )
+        result = session.execute(query).fetchone()
+
+        if not result:
+            raise HTTPException(status_code=404, detail="Data not found")
+
+        update_query = (
+            processed_agent_data.update()
+            .where(processed_agent_data.c.id == processed_agent_data_id)
+            .values(
+                road_state=data.road_state,
+                user_id=data.agent_data.user_id,
+                x=data.agent_data.accelerometer.x,
+                y=data.agent_data.accelerometer.y,
+                z=data.agent_data.accelerometer.z,
+                latitude=data.agent_data.gps.latitude,
+                longitude=data.agent_data.gps.longitude,
+                timestamp=data.agent_data.timestamp,
+            )
+        )
+
+        session.execute(update_query)
+        session.commit()
+
+        updated_result = session.execute(query).fetchone()
+
+        return ProcessedAgentDataInDB(**updated_result._mapping)
+
+    finally:
+        session.close()
 
 
 @app.delete(
@@ -161,8 +177,28 @@ def update_processed_agent_data(processed_agent_data_id: int, data: ProcessedAge
 )
 def delete_processed_agent_data(processed_agent_data_id: int):
     # Delete by id
-    pass
+    session = SessionLocal()
 
+    try:
+        query = select(processed_agent_data).where(
+            processed_agent_data.c.id == processed_agent_data_id
+        )
+        result = session.execute(query).fetchone()
+
+        if not result:
+            raise HTTPException(status_code=404, detail="Data not found")
+
+        delete_query = processed_agent_data.delete().where(
+            processed_agent_data.c.id == processed_agent_data_id
+        )
+
+        session.execute(delete_query)
+        session.commit()
+
+        return ProcessedAgentDataInDB(**result._mapping)
+
+    finally:
+        session.close()
 
 if __name__ == "__main__":
     import uvicorn

store/schemas.py

@@ -0,0 +1,51 @@
+from datetime import datetime
+
+from pydantic import BaseModel, field_validator
+
+
+class ProcessedAgentDataInDB(BaseModel):
+    id: int
+    road_state: str
+    user_id: int
+    x: float
+    y: float
+    z: float
+    latitude: float
+    longitude: float
+    timestamp: datetime
+
+
+# FastAPI models
+class AccelerometerData(BaseModel):
+    x: float
+    y: float
+    z: float
+
+
+class GpsData(BaseModel):
+    latitude: float
+    longitude: float
+
+
+class AgentData(BaseModel):
+    user_id: int
+    accelerometer: AccelerometerData
+    gps: GpsData
+    timestamp: datetime
+
+    @classmethod
+    @field_validator("timestamp", mode="before")
+    def check_timestamp(cls, value):
+        if isinstance(value, datetime):
+            return value
+        try:
+            return datetime.fromisoformat(value)
+        except (TypeError, ValueError):
+            raise ValueError(
+                "Invalid timestamp format. Expected ISO 8601 format (YYYY-MM-DDTHH:MM:SSZ)."
+            )
+
+
+class ProcessedAgentData(BaseModel):
+    road_state: str
+    agent_data: AgentData