fix: actually use session variable in websocket GET

[L4] SCRUM-81: implementation data processing function

.gitea/workflows/reset-docker.yaml

@@ -0,0 +1,16 @@
+name: Reset docker state
+on: workflow_dispatch
+
+jobs:
+  reset:
+    runs-on: host-arch-x86_64
+    name: Reset docker state
+    steps:
+      - name: Stop all containers
+        run: docker stop $(docker ps -a | cut -d " " -f 1 | tail -n +2)
+      
+      - name: Remove all containers
+        run: docker rm $(docker ps -a | cut -d " " -f 1 | tail -n +2)
+
+      - name: Remove extra volumes
+        run: docker volume rm road_vision_postgres_data road_vision_pgadmin-data

.gitea/workflows/tests.yaml

@@ -0,0 +1,71 @@
+name: Component testing
+on: [push, workflow_dispatch]
+
+jobs:
+  hub-test:
+    name: Hub testing
+    runs-on: host-arch-x86_64
+    steps:
+      - name: Clone repository
+        run: git clone --revision ${{ gitea.sha }} --depth 1 ${{ gitea.server_url }}/${{ gitea.repository }}
+
+      - name: Build Hub testing container
+        working-directory: IoT-Systems
+        run: docker build -t local/hub/${{gitea.sha}} -f hub/Dockerfile-test .
+
+      - name: Run Hub tests
+        working-directory: IoT-Systems
+        run: docker run --rm -it local/hub/${{gitea.sha}}
+
+      - name: Clean up containers
+        if: ${{always()}}
+        run: docker image rm local/hub/${{gitea.sha}}
+
+  store-test:
+    name: Store testing
+    runs-on: host-arch-x86_64
+    steps:
+      - name: Clone repository
+        run: git clone --revision ${{ gitea.sha }} --depth 1 ${{ gitea.server_url }}/${{ gitea.repository }}
+
+      - name: Build Store testing container
+        working-directory: IoT-Systems
+        run: docker build -t local/store/${{gitea.sha}} -f store/Dockerfile-test .
+
+      - name: Run Store tests
+        working-directory: IoT-Systems
+        run: docker run --rm -it local/store/${{gitea.sha}}
+
+      - name: Clean up containers
+        if: ${{always()}}
+        run: docker image rm local/store/${{gitea.sha}}
+
+  integration-smoke-test:
+    name: Integration smoke testing
+    runs-on: host-arch-x86_64
+    needs:
+      - hub-test
+      - store-test
+    steps:
+      - name: Clone repository
+        run: git clone --revision ${{ gitea.sha }} --depth 1 ${{ gitea.server_url }}/${{ gitea.repository }}
+
+      - name: Build all production containers
+        working-directory: IoT-Systems
+        run: docker-compose build
+
+      - name: Start all production containers
+        working-directory: IoT-Systems
+        run: docker-compose up -d
+
+      - name: Wait for crashes to happen
+        run: sleep 30
+
+      - name: Check for dead containers
+        working-directory: IoT-Systems
+        run: docker ps -a | python3 utils/check-up.py
+
+      - name: Clean up
+        if: ${{always()}}
+        working-directory: IoT-Systems
+        run: docker-compose down -v

MapView/config.py

@@ -2,3 +2,5 @@ import os
 
 STORE_HOST = os.environ.get("STORE_HOST") or "localhost"
 STORE_PORT = os.environ.get("STORE_PORT") or 8000
+
+TRACK_ID = int(os.environ.get("TID") or '1')

MapView/datasource.py

@@ -72,9 +72,10 @@ class Datasource:
         )
         new_points = [
             (
-                processed_agent_data.latitude,
                 processed_agent_data.longitude,
+                processed_agent_data.latitude,
                 processed_agent_data.road_state,
+                processed_agent_data.user_id
             )
             for processed_agent_data in processed_agent_data_list
         ]

MapView/domain/accelerometer.py

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

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

@@ -4,7 +4,16 @@ from kivy_garden.mapview import MapMarker, MapView
 from kivy.clock import Clock
 from lineMapLayer import LineMapLayer
 from datasource import Datasource
+import config
 
+line_layer_colors = [
+    [1, 0, 0, 1],
+    [1, 0.5, 0, 1],
+    [0, 1, 0, 1],
+    [0, 1, 1, 1],
+    [0, 0, 1, 1],
+    [1, 0, 1, 1],
+]
 
 class MapViewApp(App):
     def __init__(self, **kwargs):
@@ -12,14 +21,19 @@ class MapViewApp(App):
 
         self.mapview = None
         self.datasource = Datasource(user_id=1)
-        self.line_layer = None
-        self.car_marker = None
+        self.line_layers = dict()
+        self.car_markers = dict()
+
+        # додати необхідні змінні
+        self.bump_markers = []
+        self.pothole_markers = []
 
     def on_start(self):
         """
         Встановлює необхідні маркери, викликає функцію для оновлення мапи
         """
-        Clock.schedule_interval(self.update, 0.3)
+        self.update()
+        Clock.schedule_interval(self.update, 0.1)
 
     def update(self, *args):
         """
@@ -32,13 +46,17 @@ class MapViewApp(App):
 
         for point in new_points:
 
-            lat, lon, road_state = point
+            lat, lon, road_state, user_id = point
 
             # Оновлює лінію маршрута
-            self.line_layer.add_point((lat, lon))
+            if user_id not in self.line_layers:
+                self.line_layers[user_id] = LineMapLayer(color = line_layer_colors[user_id % len(line_layer_colors)])
+                self.mapview.add_layer(self.line_layers[user_id])
+
+            self.line_layers[user_id].add_point((lat, lon))
 
             # Оновлює маркер маниши
-            self.update_car_marker((lat, lon))
+            self.update_car_marker(lat, lon, user_id)
 
             # Перевіряємо стан дороги
             self.check_road_quality(point)
@@ -51,40 +69,66 @@ class MapViewApp(App):
         if len(point) < 3:
             return
 
-        lat, lon, road_state = point
+        lat, lon, road_state, user_id = 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):
+    def update_car_marker(self, lat, lon, user_id):
         """
         Оновлює відображення маркера машини на мапі
         :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)
+        if user_id not in self.car_markers:
+            self.car_markers[user_id] = MapMarker(lat=lat, lon=lon, source='./images/car.png')
+            self.mapview.add_marker(self.car_markers[user_id])
         else:
-            self.car_marker.lat = lat
-            self.car_marker.lon = lon
+            self.car_markers[user_id].lat = lat
+            self.car_markers[user_id].lon = lon
 
-        self.mapview.center_on(lat, lon)
+        if user_id == config.TRACK_ID:
+            self.mapview.center_on(lat, lon)
 
     def set_pothole_marker(self, point):
-        """
-        Встановлює маркер для ями
-        :param point: GPS координати
-        """
+        if isinstance(point, dict):
+            lat = point.get("lat")
+            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):
-        """
-        Встановлює маркер для лежачого поліцейського
-        :param point: GPS координати
-        """
+        if isinstance(point, dict):
+            lat = point.get("lat")
+            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/bump.png"  
+        )
+
+        self.mapview.add_marker(marker)
+        self.bump_markers.append(marker)
+
 
     def build(self):
         """
@@ -97,9 +141,6 @@ class MapViewApp(App):
             lon=30.5234
         )
 
-        self.line_layer = LineMapLayer()
-        self.mapview.add_layer(self.line_layer)
-
         return self.mapview
 
 

agent/docker/docker-compose.yaml

@@ -1,34 +0,0 @@
-name: "road_vision"
-services:
-  mqtt:
-    image: eclipse-mosquitto
-    container_name: mqtt
-    volumes:
-      - ./mosquitto:/mosquitto
-      - ./mosquitto/data:/mosquitto/data
-      - ./mosquitto/log:/mosquitto/log
-    ports:
-      - 1883:1883
-      - 9001:9001
-    networks:
-      mqtt_network:
-
-
-  fake_agent:
-    container_name: agent
-    build:
-      context: ../../
-      dockerfile: agent/Dockerfile
-    depends_on:
-      - mqtt
-    environment:
-      MQTT_BROKER_HOST: "mqtt"
-      MQTT_BROKER_PORT: 1883
-      MQTT_TOPIC: "agent_data_topic"
-      DELAY: 0.1
-    networks:
-      mqtt_network:
-
-
-networks:
-  mqtt_network:

agent/src/config.py

@@ -1,18 +1,17 @@
 import os
 
-
 def try_parse(type, value: str):
     try:
         return type(value)
     except Exception:
         return None
 
-
-USER_ID = 1
 # MQTT config
 MQTT_BROKER_HOST = os.environ.get("MQTT_BROKER_HOST") or "mqtt"
 MQTT_BROKER_PORT = try_parse(int, os.environ.get("MQTT_BROKER_PORT")) or 1883
 MQTT_TOPIC = os.environ.get("MQTT_TOPIC") or "agent"
 
-# Delay for sending data to mqtt in seconds
+# Data-related config
+USER_ID = try_parse(int, os.environ.get("USER_ID")) or 1
 DELAY = try_parse(float, os.environ.get("DELAY")) or 1
+GPS_SOURCE = os.environ.get("GPS_SOURCE") or "data/gps.csv"

agent/src/data/gpx-to-csv.py

@@ -0,0 +1,13 @@
+print('lat,lon')
+
+try:
+    while True:
+        i = input()
+
+        if '<trkpt' not in i:
+            continue
+
+        si = i.split('"')[1::2]
+        print(f"{si[0]},{si[1]}")
+except EOFError:
+    pass

agent/src/data/route2.csv

@@ -0,0 +1,801 @@
+lat,lon
+50.452166,30.5020695
+50.452164,30.5020648
+50.4523877,30.5018288
+50.4524748,30.5017403
+50.4525209,30.5016947
+50.4525755,30.501633
+50.4525977,30.501692
+50.4526934,30.5019307
+50.4527053,30.5019602
+50.4527122,30.5019736
+50.4528471,30.5022928
+50.4528625,30.5023304
+50.4529222,30.5024672
+50.4535387,30.5039424
+50.4538547,30.5046907
+50.4539059,30.5047256
+50.4539776,30.5047524
+50.4540494,30.5047524
+50.4541194,30.5047229
+50.4541894,30.5047202
+50.4542509,30.504739
+50.4543004,30.504798
+50.4544985,30.5052647
+50.4545156,30.5053049
+50.4546539,30.5056348
+50.4543641,30.5059133
+50.4543641,30.5059133
+50.4543636,30.5059138
+50.4542799,30.5059996
+50.4542458,30.5060318
+50.4541177,30.5061471
+50.4540152,30.5062652
+50.4539025,30.5063993
+50.4534568,30.5069518
+50.453356,30.5070752
+50.4532979,30.5071476
+50.453134,30.5073515
+50.4530998,30.5073944
+50.4530418,30.5074614
+50.4527886,30.5077796
+50.4527886,30.5077796
+50.4526575,30.5079442
+50.4525021,30.5081373
+50.4521947,30.5085209
+50.4521452,30.5085826
+50.4521179,30.5086148
+50.4520239,30.5087328
+50.4517097,30.5091351
+50.4515713,30.5093095
+50.4513544,30.5095857
+50.4511734,30.5098084
+50.4511188,30.5098781
+50.4510658,30.5099425
+50.4507738,30.5103126
+50.450726,30.5103743
+50.4506423,30.510495
+50.4506252,30.5105191
+50.4503622,30.5108464
+50.4501692,30.5110905
+50.4501026,30.5111709
+50.4498771,30.5114526
+50.4497456,30.5116162
+50.4495953,30.5118066
+50.4495578,30.5118522
+50.4495407,30.5118763
+50.4494911,30.5119354
+50.449428,30.5120185
+50.4493631,30.5121338
+50.4493135,30.5122519
+50.4492964,30.5123618
+50.4492606,30.5125818
+50.4492401,30.5127212
+50.4491564,30.5132577
+50.4491298,30.5134198
+50.4491298,30.5134198
+50.4491274,30.5134347
+50.4491017,30.5135876
+50.4489702,30.5143896
+50.4489395,30.5145746
+50.4490403,30.5146149
+50.449568,30.5148187
+50.4497525,30.5148938
+50.4499557,30.5149743
+50.4500138,30.5149984
+50.4503263,30.5151245
+50.4503963,30.5151513
+50.4507425,30.5152913
+50.4507425,30.5152913
+50.4512844,30.5155107
+50.4513749,30.5155483
+50.4514142,30.5155644
+50.4515594,30.5156261
+50.4518463,30.5157468
+50.4518753,30.5157602
+50.4523057,30.5159426
+50.4525687,30.5160525
+50.4526848,30.5161035
+50.4526507,30.5162591
+50.4526054,30.5164549
+50.4526054,30.5164549
+50.4525533,30.5166802
+50.4523672,30.5174848
+50.4521708,30.5183351
+50.4521554,30.5183995
+50.4521776,30.5184129
+50.4522032,30.518429
+50.4524919,30.5185765
+50.4526285,30.5186462
+50.452649,30.518657
+50.4526712,30.5186784
+50.4526917,30.5186918
+50.452707,30.5187187
+50.4528625,30.5191424
+50.4529991,30.5195448
+50.4530264,30.5196226
+50.4531033,30.5198452
+50.4531921,30.5201054
+50.4532143,30.5201751
+50.4532535,30.5201483
+50.4534756,30.519923
+50.4537573,30.5196655
+50.453964,30.519475
+50.4543892,30.5190969
+50.4545139,30.5189735
+50.45456,30.5190834
+50.4547393,30.5195406
+50.4547393,30.5195406
+50.4548008,30.5196977
+50.454852,30.5198237
+50.4549101,30.5200785
+50.4550842,30.5205238
+50.4550125,30.5206257
+50.4545873,30.5209771
+50.4544695,30.5210683
+50.4544251,30.5211005
+50.4541894,30.5212748
+50.4537898,30.5215698
+50.4537215,30.5216208
+50.4536378,30.5216825
+50.4534824,30.5217898
+50.453426,30.5218247
+50.4533782,30.5218488
+50.4532006,30.5219239
+50.4530469,30.5219802
+50.4530144,30.5219936
+50.4528163,30.5220661
+50.4527737,30.5220821
+50.452707,30.5221036
+50.452415,30.5222002
+50.4523569,30.5222189
+50.4520769,30.5223182
+50.4517507,30.5224308
+50.4514791,30.522522
+50.4514689,30.5225274
+50.4514586,30.5225354
+50.4514484,30.5225462
+50.4514381,30.5225569
+50.4514313,30.5225703
+50.4514108,30.5226079
+50.451392,30.5226427
+50.4513271,30.5228063
+50.4512844,30.5229056
+50.4512503,30.522978
+50.4512349,30.5230075
+50.4512127,30.5230531
+50.4511632,30.5231443
+50.4505586,30.5242413
+50.4505286,30.5242958
+50.4505286,30.5242958
+50.4504168,30.5244988
+50.4496551,30.5234367
+50.4495304,30.5232811
+50.4493835,30.5231068
+50.4492384,30.5229566
+50.4491222,30.5228466
+50.4489412,30.5226856
+50.448755,30.5225408
+50.4485637,30.5224094
+50.4483707,30.5222887
+50.4481726,30.5221868
+50.4479728,30.5220956
+50.4477593,30.5220178
+50.4476961,30.5219963
+50.4472059,30.5218354
+50.4470436,30.5217817
+50.446673,30.5216584
+50.445696,30.5213392
+50.4449137,30.5210817
+50.4447976,30.5210415
+50.444239,30.5208457
+50.4438598,30.5207196
+50.4435865,30.5206284
+50.4435455,30.520615
+50.4430843,30.5204648
+50.4426351,30.5203092
+50.4425274,30.5202717
+50.442413,30.5202127
+50.4423225,30.5201751
+50.4422234,30.5201349
+50.4417553,30.5199713
+50.4412668,30.519813
+50.4409747,30.5197191
+50.4409132,30.5197325
+50.4408876,30.519754
+50.4408551,30.5197889
+50.4408244,30.5198506
+50.4408005,30.5199525
+50.4407817,30.5201241
+50.4407748,30.5211997
+50.4407663,30.5221868
+50.440739,30.5223665
+50.4407031,30.5224469
+50.4406467,30.5225086
+50.440575,30.5225435
+50.4404349,30.5226025
+50.4403946,30.5226263
+50.4403946,30.5226263
+50.4403085,30.5226776
+50.440235,30.5227232
+50.4401547,30.5228144
+50.439632,30.5234501
+50.4395483,30.5235574
+50.4393655,30.5238175
+50.4386856,30.5247724
+50.4384755,30.5250487
+50.4382944,30.525282
+50.4382739,30.5253088
+50.4381116,30.5255395
+50.4379237,30.5258104
+50.4375855,30.5262879
+50.4368645,30.5272749
+50.4362598,30.5280849
+50.4356584,30.528844
+50.4355747,30.5289432
+50.4344727,30.5302709
+50.4343617,30.530405
+50.4341857,30.5306143
+50.433938,30.5309147
+50.4337808,30.5310997
+50.4336817,30.5312231
+50.4322978,30.5329397
+50.4307413,30.5348656
+50.4306149,30.5350238
+50.4305192,30.5351418
+50.4304064,30.535284
+50.4301125,30.5356488
+50.4286209,30.5374995
+50.4279904,30.5383015
+50.4279426,30.5383605
+50.4278708,30.538449
+50.4277973,30.5385375
+50.427664,30.5386984
+50.427095,30.5393931
+50.4270779,30.5394307
+50.4270643,30.5394682
+50.4270574,30.5395111
+50.4270592,30.5395514
+50.4270677,30.5395862
+50.4270831,30.5396211
+50.4271036,30.5396453
+50.4271275,30.5396587
+50.4271514,30.539664
+50.4271719,30.5396587
+50.4271907,30.5396479
+50.4272129,30.5396345
+50.4277273,30.5389827
+50.4278708,30.5388138
+50.4280058,30.5390766
+50.4282142,30.5394977
+50.4285338,30.5400932
+50.4286841,30.5403426
+50.4287713,30.5404687
+50.4290703,30.5408844
+50.4291284,30.5409676
+50.4292497,30.5411232
+50.4296324,30.541665
+50.4297418,30.5418232
+50.4297213,30.5418581
+50.4296854,30.5419198
+50.4295487,30.5421799
+50.4295145,30.5422363
+50.4294735,30.5423141
+50.4292856,30.5426654
+50.4292155,30.5428264
+50.4291574,30.5429631
+50.4290293,30.5433252
+50.4289729,30.5435157
+50.4289302,30.5436927
+50.4288994,30.5438724
+50.4288772,30.5440924
+50.4288789,30.5442882
+50.4288977,30.5445161
+50.4289387,30.5447736
+50.4290737,30.545361
+50.4291403,30.5456185
+50.4292856,30.546096
+50.4296649,30.5473137
+50.4297008,30.5474532
+50.4297315,30.5476061
+50.4297554,30.5477831
+50.4297691,30.5479467
+50.4297794,30.5483302
+50.429776,30.5485448
+50.4297537,30.5487487
+50.4297093,30.5489552
+50.4296581,30.5491054
+50.4296136,30.5491778
+50.4295777,30.5492288
+50.4295709,30.5492368
+50.4295265,30.5492878
+50.4293573,30.5494165
+50.4290874,30.5496177
+50.4283731,30.5501676
+50.4281369,30.5503436
+50.4281369,30.5503436
+50.4280314,30.5504224
+50.427893,30.5505323
+50.4277119,30.5506825
+50.4275769,30.5508086
+50.4274727,30.5509186
+50.4273906,30.5510151
+50.4273308,30.5510956
+50.4272676,30.5511922
+50.4272164,30.5512941
+50.4271514,30.551447
+50.4270506,30.5517823
+50.4269515,30.5522919
+50.4269139,30.5525064
+50.426878,30.5527693
+50.4266439,30.5546013
+50.426579,30.555194
+50.4265499,30.5554274
+50.4264731,30.5561301
+50.4263996,30.5568033
+50.4263808,30.5569911
+50.4263449,30.557254
+50.4262919,30.557549
+50.426133,30.5584019
+50.4259707,30.5591878
+50.4259519,30.5592898
+50.4258938,30.5595526
+50.4258374,30.5597752
+50.4257827,30.5599576
+50.4257502,30.5600515
+50.4256426,30.5601749
+50.4255691,30.5602607
+50.4254598,30.5603895
+50.4253606,30.5604136
+50.4253436,30.5604163
+50.4252308,30.5604538
+50.4251573,30.5604941
+50.4250684,30.5605236
+50.4249933,30.560545
+50.42493,30.560545
+50.424848,30.5605423
+50.424754,30.5605128
+50.4246874,30.560478
+50.4245831,30.5603895
+50.4245353,30.5603385
+50.424455,30.5602178
+50.4243012,30.5599576
+50.424144,30.5593622
+50.4241115,30.5592388
+50.4240568,30.5590296
+50.4237407,30.557895
+50.4234844,30.5569348
+50.4233407,30.5563814
+50.4233407,30.5563814
+50.4232434,30.5560067
+50.4231494,30.5556554
+50.4231221,30.5555534
+50.4229102,30.5547434
+50.422794,30.5543384
+50.4226436,30.5538502
+50.4226094,30.5537537
+50.4223719,30.5531019
+50.4222215,30.5527076
+50.4220882,30.5523643
+50.4220233,30.5522007
+50.4218524,30.5518171
+50.4216268,30.5513424
+50.4214986,30.5510902
+50.4211329,30.5503848
+50.4208287,30.5498055
+50.4208168,30.5496499
+50.4208116,30.5494621
+50.420827,30.549328
+50.4209295,30.5489552
+50.421003,30.5487701
+50.4210526,30.5485824
+50.4210782,30.5483893
+50.4210936,30.548231
+50.4211004,30.5480406
+50.4211039,30.5478555
+50.4211004,30.547657
+50.4210953,30.547539
+50.4210714,30.5473807
+50.4210475,30.5472842
+50.4210167,30.5471742
+50.4209159,30.5468953
+50.4208202,30.5467129
+50.4211611,30.5462681
+50.4211611,30.5462681
+50.4216798,30.5455917
+50.4219105,30.5452994
+50.4227256,30.5442721
+50.4228059,30.5441701
+50.422888,30.5440682
+50.4231939,30.5436605
+50.4234075,30.5433896
+50.4235049,30.5432689
+50.4235835,30.5431858
+50.4237663,30.5430436
+50.4239611,30.5429068
+50.4241218,30.5427861
+50.42493,30.5422336
+50.4250736,30.542129
+50.4252051,30.5420056
+50.4254119,30.5417749
+50.4258528,30.5412439
+50.4261484,30.5408925
+50.4266422,30.5403292
+50.4272129,30.5396345
+50.4277273,30.5389827
+50.4278708,30.5388138
+50.4280058,30.5390766
+50.4282142,30.5394977
+50.4285338,30.5400932
+50.4286841,30.5403426
+50.4287713,30.5404687
+50.4290703,30.5408844
+50.4291284,30.5409676
+50.4292497,30.5411232
+50.4296324,30.541665
+50.4297418,30.5418232
+50.4298545,30.5416194
+50.4297503,30.5414799
+50.4292787,30.5408201
+50.4292104,30.5407262
+50.4288994,30.5402917
+50.4288003,30.5401388
+50.4286824,30.5399403
+50.4286602,30.5398974
+50.4281493,30.5388701
+50.4280194,30.5386341
+50.4278708,30.538449
+50.4278332,30.5384141
+50.42777,30.5383578
+50.4277597,30.538339
+50.4276692,30.5381566
+50.4271378,30.5370998
+50.4264867,30.5357373
+50.4264303,30.5355576
+50.4264901,30.5354851
+50.426702,30.5352196
+50.4269361,30.5349514
+50.427242,30.5346
+50.427377,30.534364
+50.4274317,30.5342352
+50.4274915,30.5340958
+50.4276042,30.5337766
+50.427676,30.5335137
+50.4277426,30.5333501
+50.4278212,30.5331945
+50.4278042,30.5330148
+50.4278127,30.5328324
+50.4278229,30.5327466
+50.4278417,30.5325857
+50.4278503,30.5325159
+50.4278691,30.5323523
+50.4278947,30.5319554
+50.427905,30.5318078
+50.4279169,30.5311856
+50.4278964,30.5308852
+50.4278435,30.5305606
+50.4278281,30.5304882
+50.4278076,30.5303943
+50.4277597,30.5301932
+50.4276794,30.5299571
+50.4276343,30.529882
+50.4276343,30.529882
+50.4275359,30.5297184
+50.4273548,30.5294877
+50.4273138,30.5294287
+50.4270711,30.5291069
+50.4269498,30.5289325
+50.4268302,30.5287689
+50.4265875,30.5284256
+50.426309,30.5280313
+50.4262612,30.5279642
+50.426186,30.5278569
+50.4259638,30.5275458
+50.425834,30.5273393
+50.4257554,30.5272079
+50.4256682,30.5270416
+50.4255589,30.5267975
+50.4253726,30.5263495
+50.4253299,30.5262423
+50.4251026,30.5256709
+50.4250411,30.5255046
+50.4247899,30.5248448
+50.4245216,30.524134
+50.4244857,30.5240375
+50.4243439,30.5236378
+50.423956,30.5222431
+50.4237646,30.5215618
+50.423739,30.5214679
+50.423686,30.5212614
+50.4234399,30.5204085
+50.4231494,30.5193731
+50.4230982,30.5191934
+50.4235681,30.5189037
+50.4239372,30.5186757
+50.4243097,30.518437
+50.4243654,30.5184017
+50.4243654,30.5184017
+50.424631,30.5182332
+50.4250719,30.5179542
+50.4251812,30.5178899
+50.4252752,30.5178282
+50.4256135,30.5176163
+50.4258955,30.5174446
+50.4269395,30.5167982
+50.4270762,30.5167124
+50.4275854,30.5163959
+50.4279853,30.5161518
+50.4280673,30.5161116
+50.4282074,30.5160901
+50.4283629,30.5160901
+50.428462,30.5160874
+50.4288721,30.5160847
+50.4289763,30.516082
+50.4291096,30.5160794
+50.4292531,30.5160767
+50.4297691,30.5160686
+50.4299195,30.5160606
+50.4300903,30.5160633
+50.4303962,30.5160633
+50.4306354,30.5160633
+50.4311394,30.516066
+50.431746,30.5160686
+50.4319578,30.5160686
+50.4319954,30.5160686
+50.4322893,30.516066
+50.4325336,30.516066
+50.4333913,30.5160606
+50.4338201,30.5160552
+50.4339465,30.5160499
+50.4339414,30.5162671
+50.4339414,30.5167043
+50.433938,30.5182251
+50.4339431,30.518319
+50.4339687,30.5183914
+50.4340183,30.518429
+50.4340832,30.5184397
+50.4343719,30.518437
+50.4344796,30.5184343
+50.4350382,30.5184236
+50.4354107,30.5184236
+50.435439,30.5184243
+50.435439,30.5184243
+50.4355149,30.5184263
+50.4358087,30.5184236
+50.4359471,30.5184263
+50.4359471,30.5181313
+50.4359471,30.5176619
+50.4359471,30.517568
+50.4359471,30.517391
+50.4359471,30.5171818
+50.4359488,30.5168197
+50.4359488,30.5163369
+50.4359505,30.516023
+50.436135,30.516074
+50.4365826,30.5160606
+50.4386532,30.5160606
+50.4387335,30.5160606
+50.4388445,30.5160847
+50.4389145,30.5161089
+50.4389726,30.5161491
+50.4390239,30.5162027
+50.4391093,30.5163315
+50.4391947,30.5165005
+50.4392818,30.5167097
+50.439304,30.5167714
+50.4393501,30.5168921
+50.4394168,30.5168197
+50.4395295,30.5162457
+50.4395825,30.515956
+50.439591,30.5159131
+50.4396013,30.5158567
+50.439726,30.5151111
+50.4399378,30.5138612
+50.4399515,30.5137941
+50.4399651,30.5137056
+50.4399515,30.513601
+50.4399856,30.5134293
+50.4401155,30.5127186
+50.4402316,30.5120265
+50.4403327,30.5113796
+50.4403327,30.5113796
+50.4403444,30.511305
+50.4403734,30.5111226
+50.440411,30.5109027
+50.4404998,30.5103609
+50.4405733,30.5099291
+50.4408346,30.5083305
+50.4408449,30.5082661
+50.440903,30.5082205
+50.4417656,30.5029258
+50.44174,30.5027434
+50.4417724,30.5025396
+50.441822,30.5022258
+50.4418373,30.5021319
+50.441933,30.5015016
+50.4418937,30.5014077
+50.4418442,30.501295
+50.4418203,30.5012387
+50.4414479,30.5003777
+50.4414052,30.5002812
+50.4412634,30.4999593
+50.441166,30.4997474
+50.4413044,30.4996026
+50.4415418,30.4993424
+50.4416785,30.4992029
+50.4417434,30.4991359
+50.4421072,30.4987496
+50.4423669,30.4984733
+50.442618,30.4982105
+50.4426453,30.4981783
+50.4429921,30.4978135
+50.443045,30.4977572
+50.4436377,30.4971322
+50.4437214,30.4970437
+50.4438308,30.4969257
+50.4439907,30.496752
+50.4439907,30.496752
+50.4441741,30.4965529
+50.444309,30.496408
+50.4445892,30.496105
+50.4446968,30.4959843
+50.4449923,30.4956678
+50.4452212,30.495421
+50.4454774,30.4951447
+50.4459334,30.4946458
+50.4459642,30.4946297
+50.4460803,30.4947102
+50.4461606,30.4947907
+50.4462289,30.4949087
+50.4462887,30.4950187
+50.4460786,30.4963624
+50.4459368,30.4972556
+50.4458839,30.4975989
+50.4458771,30.4976445
+50.4458361,30.4979074
+50.4457865,30.4982239
+50.4457336,30.4985511
+50.4456755,30.4989213
+50.4453732,30.5007908
+50.4444935,30.506292
+50.4444474,30.5065843
+50.4440152,30.5091968
+50.4439708,30.5094624
+50.4438683,30.5100873
+50.4437228,30.5109637
+50.4437228,30.5109637
+50.443537,30.5120829
+50.4435301,30.5121365
+50.4434891,30.5123886
+50.4434584,30.5125684
+50.4432756,30.5136278
+50.443209,30.5140194
+50.4431099,30.5146015
+50.4430894,30.5147195
+50.4430553,30.5149341
+50.4430399,30.5150118
+50.4427358,30.5169028
+50.4427188,30.517002
+50.4426966,30.5171549
+50.4428247,30.5172059
+50.4429425,30.5172488
+50.4430416,30.517289
+50.4433405,30.5174151
+50.4433713,30.5174285
+50.4434157,30.5174473
+50.4435472,30.5174983
+50.4439657,30.5176672
+50.4440477,30.5176994
+50.4440887,30.5177155
+50.4442937,30.5177987
+50.4442971,30.5177987
+50.4445909,30.5179194
+50.4449342,30.5180588
+50.4451306,30.5181366
+50.445158,30.5181473
+50.4452621,30.5181903
+50.4452878,30.5180266
+50.4453407,30.5176967
+50.4454517,30.5170074
+50.4455013,30.5167043
+50.4455474,30.516412
+50.4455918,30.5161437
+50.4456191,30.515964
+50.4456277,30.5159158
+50.4458617,30.5144781
+50.4459505,30.5139309
+50.4460052,30.5135876
+50.4460376,30.5133972
+50.4460666,30.5132148
+50.4461469,30.5127132
+50.4461503,30.5126971
+50.4461606,30.51263
+50.4462699,30.5119407
+50.4462801,30.5118844
+50.4463177,30.5116457
+50.4463468,30.5114633
+50.4463604,30.5113855
+50.4463724,30.5113104
+50.4464971,30.5105326
+50.4465346,30.5102912
+50.4465415,30.5102482
+50.4465831,30.5100038
+50.4465831,30.5100038
+50.4466405,30.5096662
+50.4466918,30.5093497
+50.4467874,30.5087703
+50.4468079,30.508647
+50.4468267,30.5085397
+50.4468335,30.5084994
+50.4470197,30.5073434
+50.4470317,30.507271
+50.4470368,30.5072388
+50.447071,30.5070457
+50.4471171,30.5067694
+50.4471205,30.5067426
+50.4471376,30.5066407
+50.4472161,30.506174
+50.4472315,30.5060828
+50.4473272,30.5055115
+50.4474484,30.5047899
+50.4474621,30.5046988
+50.4475834,30.5039907
+50.4475987,30.5038726
+50.4476158,30.5037519
+50.4476722,30.5034113
+50.4479437,30.5017617
+50.4480462,30.5011395
+50.4481248,30.5006674
+50.4481367,30.5005896
+50.4481436,30.5005494
+50.4481641,30.5004287
+50.4483485,30.4993021
+50.4483912,30.499042
+50.4484032,30.4989749
+50.4484954,30.498417
+50.4485022,30.4983741
+50.4485176,30.498299
+50.4485672,30.4982105
+50.4486013,30.498189
+50.4486372,30.4981676
+50.4487243,30.4981515
+50.44889,30.4979798
+50.4490061,30.4978591
+50.4491,30.4977626
+50.4491359,30.497725
+50.4493101,30.4975426
+50.4493477,30.4975051
+50.4494929,30.4973549
+50.449539,30.4973066
+50.4496107,30.4972342
+50.4498003,30.4970357
+50.449872,30.4969633
+50.4502699,30.4965502
+50.4502802,30.4965395
+50.4503707,30.4964456
+50.45041,30.4965422
+50.4505278,30.496805
+50.4507567,30.4973254
+50.4507789,30.4973763
+50.4508984,30.497658
+50.4513732,30.4987684
+50.4514193,30.498881
+50.4514791,30.4990259
+50.4515799,30.49927
+50.4516157,30.4993585
+50.4516277,30.4993853
+50.4516738,30.4994953
+50.4519932,30.5002436
+50.4520188,30.5003026
+50.4520717,30.5004233
+50.4521725,30.500662
+50.4522237,30.5007881
+50.4522886,30.5009329
+50.4523262,30.5010295
+50.4524833,30.501405
+50.4525755,30.501633
+50.4525209,30.5016947
+50.4524748,30.5017403
+50.4523877,30.5018288
+50.452164,30.5020648
+50.4521662,30.5020702

agent/src/domain/accelerometer.py

@@ -3,6 +3,6 @@ from dataclasses import dataclass
 
 @dataclass
 class Accelerometer:
-    x: int
-    y: int
-    z: int
+    x: float
+    y: float
+    z: float

agent/src/file_datasource.py

@@ -15,6 +15,7 @@ class FileDatasource:
 
     def __init__(
             self,
+            acc_divisor: float,
             accelerometer_filename: str,
             gps_filename: str,
             park_filename: str,
@@ -34,6 +35,8 @@ class FileDatasource:
 
         self._started = False
 
+        self.acc_divisor = acc_divisor
+
     def startReading(self, *args, **kwargs):
         """Must be called before read()"""
         if self._started:
@@ -160,15 +163,14 @@ class FileDatasource:
 
             return row
 
-    @staticmethod
-    def _parse_acc(row: List[str]) -> Accelerometer:
+    def _parse_acc(self, 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])
+            x = int(row[0]) / self.acc_divisor
+            y = int(row[1]) / self.acc_divisor
+            z = int(row[2]) / self.acc_divisor
         except ValueError as e:
             raise ValueError(f"Invalid accelerometer values (expected integers): {row}") from e
 

agent/src/main.py

@@ -1,6 +1,7 @@
 from paho.mqtt import client as mqtt_client
 from schema.aggregated_data_schema import AggregatedDataSchema
 from file_datasource import FileDatasource
+import logging
 import config
 
 
@@ -15,6 +16,8 @@ def connect_mqtt(broker, port):
             print("Failed to connect {broker}:{port}, return code %d\n", rc)
             exit(rc)  # Stop execution
 
+    logging.info(f"Acting as USER_ID = {config.USER_ID}")
+
     client = mqtt_client.Client()
     client.on_connect = on_connect
     client.connect(broker, port)
@@ -28,16 +31,18 @@ def publish(client, topic, datasource):
         data = datasource.read()
         msg = AggregatedDataSchema().dumps(data)
         result = client.publish(topic, msg)
+        logging.debug(f"Published to {topic}: {msg[:50]}...")
         status = result[0]
         if status != 0:
-            print(f"Failed to send message to topic {topic}")
+            logging.error(f"Failed to send message to topic {topic}")
 
 
 def run():
+    logging.basicConfig(level = logging.INFO)
     # Prepare mqtt client
     client = connect_mqtt(config.MQTT_BROKER_HOST, config.MQTT_BROKER_PORT)
     # Prepare datasource
-    datasource = FileDatasource("data/accelerometer.csv", "data/gps.csv", "data/parking.csv")
+    datasource = FileDatasource(16384.0, "data/accelerometer.csv", config.GPS_SOURCE, "data/parking.csv")
     # Infinity publish data
     publish(client, config.MQTT_TOPIC, datasource)
 

agent/src/schema/accelerometer_schema.py

@@ -2,6 +2,6 @@ from marshmallow import Schema, fields
 
 
 class AccelerometerSchema(Schema):
-    x = fields.Int()
-    y = fields.Int()
-    z = fields.Int()
+    x = fields.Float()
+    y = fields.Float()
+    z = fields.Float()

docker-compose.yaml

@@ -1,12 +1,12 @@
-name: "road_vision__hub"
+name: "road_vision"
 services:
   mqtt:
     image: eclipse-mosquitto
     container_name: mqtt
     volumes:
-      - ./mosquitto:/mosquitto
-      - ./mosquitto/data:/mosquitto/data
-      - ./mosquitto/log:/mosquitto/log
+      - ./agent/docker/mosquitto:/mosquitto
+      - ./agent/docker/mosquitto/data:/mosquitto/data
+      - ./agent/docker/mosquitto/log:/mosquitto/log
     ports:
       - 1883:1883
       - 9001:9001
@@ -14,6 +14,60 @@ services:
       mqtt_network:
 
 
+  agent1:
+    build:
+      context: .
+      dockerfile: agent/Dockerfile
+    depends_on:
+      - mqtt
+    environment:
+      PYTHONUNBUFFERED: 1
+      MQTT_BROKER_HOST: "mqtt"
+      MQTT_BROKER_PORT: 1883
+      MQTT_TOPIC: "agent_data_topic"
+      DELAY: 1.2
+      USER_ID: 2
+    networks:
+      mqtt_network:
+
+  agent2:
+    build:
+      context: .
+      dockerfile: agent/Dockerfile
+    depends_on:
+      - mqtt
+    environment:
+      PYTHONUNBUFFERED: 1
+      MQTT_BROKER_HOST: "mqtt"
+      MQTT_BROKER_PORT: 1883
+      MQTT_TOPIC: "agent_data_topic"
+      GPS_SOURCE: "data/route2.csv"
+      DELAY: 1.0
+      USER_ID: 3
+    networks:
+      mqtt_network:
+
+  edge:
+    container_name: edge
+    build:
+      context: .
+      dockerfile: edge/Dockerfile
+    depends_on:
+      - mqtt
+    environment:
+      MQTT_BROKER_HOST: "mqtt"
+      MQTT_BROKER_PORT: 1883
+      MQTT_TOPIC: "agent_data_topic"
+      HUB_HOST: "hub"
+      HUB_PORT: 8000
+      HUB_CONNECTION_TYPE: "http"
+      HUB_MQTT_BROKER_HOST: "mqtt"
+      HUB_MQTT_BROKER_PORT: 1883
+      HUB_MQTT_TOPIC: "processed_data_topic"
+    networks:
+      mqtt_network:
+      edge_hub:
+
   postgres_db:
     image: postgres:17
     container_name: postgres_db
@@ -24,13 +78,12 @@ services:
       POSTGRES_DB: test_db
     volumes:
       - postgres_data:/var/lib/postgresql/data
-      - ./db/structure.sql:/docker-entrypoint-initdb.d/structure.sql
+      - ./store/docker/db/structure.sql:/docker-entrypoint-initdb.d/structure.sql
     ports:
       - "5432:5432"
     networks:
       db_network:
 
-
   pgadmin:
     container_name: pgadmin4
     image: dpage/pgadmin4
@@ -49,12 +102,13 @@ services:
   store:
     container_name: store
     build:
-      context: ../../
+      context: .
       dockerfile: store/Dockerfile
     depends_on:
       - postgres_db
     restart: always
     environment:
+      PYTHONUNBUFFERED: 1
       POSTGRES_USER: user
       POSTGRES_PASSWORD: pass
       POSTGRES_DB: test_db
@@ -79,13 +133,14 @@ services:
   hub:
     container_name: hub
     build:
-      context: ../../
+      context: .
       dockerfile: hub/Dockerfile
     depends_on:
       - mqtt
       - redis
       - store
     environment:
+      PYTHONUNBUFFERED: 1
       STORE_API_HOST: "store"
       STORE_API_PORT: 8000
       REDIS_HOST: "redis"
@@ -93,7 +148,7 @@ services:
       MQTT_BROKER_HOST: "mqtt"
       MQTT_BROKER_PORT: 1883
       MQTT_TOPIC: "processed_data_topic"
-      BATCH_SIZE: 1
+      BATCH_SIZE: 4
     ports:
       - "9000:8000"
     networks:
@@ -101,10 +156,11 @@ services:
       hub_store:
       hub_redis:
 
-
 networks:
   mqtt_network:
   db_network:
+  edge_hub:
+  hub:
   hub_store:
   hub_redis:
 

edge/app/adapters/agent_mqtt_adapter.py

@@ -13,9 +13,7 @@ class AgentMQTTAdapter(AgentGateway):
         broker_port,
         topic,
         hub_gateway: HubGateway,
-        batch_size=10,
     ):
-        self.batch_size = batch_size
         # MQTT
         self.broker_host = broker_host
         self.broker_port = broker_port
@@ -35,42 +33,21 @@ class AgentMQTTAdapter(AgentGateway):
         """Processing agent data and sent it to hub gateway"""
         try:
             payload: str = msg.payload.decode("utf-8")
-            # Create AgentData instance with the received data
+
             agent_data = AgentData.model_validate_json(payload, strict=True)
-            # Process the received data (you can call a use case here if needed)
             processed_data = process_agent_data(agent_data)
-            # Store the agent_data in the database (you can send it to the data processing module)
-            if not self.hub_gateway.save_data(processed_data):
-                logging.error("Hub is not available")
+
+            if self.hub_gateway.save_data(processed_data):
+                logging.info("Processed data successfully forwarded to the Hub.")
+            else:
+                logging.error("Failed to send data: Hub gateway is unavailable.")
         except Exception as e:
-            logging.info(f"Error processing MQTT message: {e}")
+            logging.error(f"Error processing MQTT message: {e}")
 
     def connect(self):
         self.client.on_connect = self.on_connect
         self.client.on_message = self.on_message
         self.client.connect(self.broker_host, self.broker_port, 60)
 
-    def start(self):
-        self.client.loop_start()
-
-    def stop(self):
-        self.client.loop_stop()
-
-
-# Usage example:
-if __name__ == "__main__":
-    broker_host = "localhost"
-    broker_port = 1883
-    topic = "agent_data_topic"
-    # Assuming you have implemented the StoreGateway and passed it to the adapter
-    store_gateway = HubGateway()
-    adapter = AgentMQTTAdapter(broker_host, broker_port, topic, store_gateway)
-    adapter.connect()
-    adapter.start()
-    try:
-        # Keep the adapter running in the background
-        while True:
-            pass
-    except KeyboardInterrupt:
-        adapter.stop()
-        logging.info("Adapter stopped.")
+    def loop_forever(self):
+        self.client.loop_forever()

edge/app/entities/agent_data.py

@@ -14,6 +14,7 @@ class GpsData(BaseModel):
 
 
 class AgentData(BaseModel):
+    user_id: int
     accelerometer: AccelerometerData
     gps: GpsData
     timestamp: datetime

edge/app/interfaces/agent_gateway.py

@@ -26,15 +26,8 @@ class AgentGateway(ABC):
         pass
 
     @abstractmethod
-    def start(self):
+    def loop_forever(self):
         """
-        Method to start listening for messages from the agent.
-        """
-        pass
-
-    @abstractmethod
-    def stop(self):
-        """
-        Method to stop the agent gateway and clean up resources.
+        Method to await for new messages.
         """
         pass

edge/app/usecases/data_processing.py

@@ -1,6 +1,7 @@
 from app.entities.agent_data import AgentData
 from app.entities.processed_agent_data import ProcessedAgentData
 
+_last_detection_state = {}
 
 def process_agent_data(
     agent_data: AgentData,
@@ -12,4 +13,24 @@ def process_agent_data(
     Returns:
         processed_data_batch (ProcessedAgentData): Processed data containing the classified state of the road surface and agent data.
     """
-    # Implement it
+    user_id = agent_data.user_id
+    road_state = "normal"
+
+    last_detection_state = _last_detection_state.get(user_id, False)
+
+    if (agent_data.accelerometer.z < 0.6):
+        road_state = "pothole"
+    elif (agent_data.accelerometer.z > 1.2):
+        road_state = "bump"
+
+    detection_happened = road_state != "normal"
+
+    if not (not last_detection_state and detection_happened):
+        road_state = "normal"
+
+    _last_detection_state[user_id] = detection_happened
+
+    return ProcessedAgentData(
+        road_state=road_state,
+        agent_data=agent_data
+    )

edge/config.py

@@ -16,9 +16,12 @@ MQTT_TOPIC = os.environ.get("MQTT_TOPIC") or "agent_data_topic"
 # Configuration for hub MQTT
 HUB_MQTT_BROKER_HOST = os.environ.get("HUB_MQTT_BROKER_HOST") or "localhost"
 HUB_MQTT_BROKER_PORT = try_parse_int(os.environ.get("HUB_MQTT_BROKER_PORT")) or 1883
-HUB_MQTT_TOPIC = os.environ.get("HUB_MQTT_TOPIC") or "processed_agent_data_topic"
+HUB_MQTT_TOPIC = os.environ.get("HUB_MQTT_TOPIC") or "processed_data_topic"
 
 # Configuration for the Hub
 HUB_HOST = os.environ.get("HUB_HOST") or "localhost"
-HUB_PORT = try_parse_int(os.environ.get("HUB_PORT")) or 12000
+HUB_PORT = try_parse_int(os.environ.get("HUB_PORT")) or 8000
 HUB_URL = f"http://{HUB_HOST}:{HUB_PORT}"
+
+# For choosing type of connection
+HUB_CONNECTION_TYPE = os.environ.get("HUB_CONNECTION_TYPE") or "mqtt"

edge/docker/docker-compose.yaml

@@ -1,50 +0,0 @@
-version: "3.9"
-# name: "road_vision"
-services:
-  mqtt:
-    image: eclipse-mosquitto
-    container_name: mqtt
-    volumes:
-      - ./mosquitto:/mosquitto
-      - ./mosquitto/data:/mosquitto/data
-      - ./mosquitto/log:/mosquitto/log
-    ports:
-      - 1883:1883
-      - 19001:9001
-    networks:
-      mqtt_network:
-
-
-  edge:
-    container_name: edge
-    build:
-      context: ../../
-      dockerfile: edge/Dockerfile
-    depends_on:
-      - mqtt
-    environment:
-      MQTT_BROKER_HOST: "mqtt"
-      MQTT_BROKER_PORT: 1883
-      MQTT_TOPIC: " "
-      HUB_HOST: "store"
-      HUB_PORT: 8000
-      HUB_MQTT_BROKER_HOST: "mqtt"
-      HUB_MQTT_BROKER_PORT: 1883
-      HUB_MQTT_TOPIC: "processed_data_topic"
-    networks:
-      mqtt_network:
-      edge_hub:
-
-
-networks:
-  mqtt_network:
-  db_network:
-  edge_hub:
-  hub:
-  hub_store:
-  hub_redis:
-
-
-volumes:
-  postgres_data:
-  pgadmin-data:

edge/main.py

@@ -10,42 +10,51 @@ from config import (
     HUB_MQTT_BROKER_HOST,
     HUB_MQTT_BROKER_PORT,
     HUB_MQTT_TOPIC,
+    HUB_CONNECTION_TYPE,
 )
 
 if __name__ == "__main__":
     # Configure logging settings
     logging.basicConfig(
-        level=logging.INFO,  # Set the log level to INFO (you can use logging.DEBUG for more detailed logs)
+        level=logging.INFO,
         format="[%(asctime)s] [%(levelname)s] [%(module)s] %(message)s",
         handlers=[
-            logging.StreamHandler(),  # Output log messages to the console
-            logging.FileHandler("app.log"),  # Save log messages to a file
+            logging.StreamHandler(),
+            logging.FileHandler("app.log"),
         ],
     )
-    # Create an instance of the StoreApiAdapter using the configuration
-    # hub_adapter = HubHttpAdapter(
-    #     api_base_url=HUB_URL,
-    # )
-    hub_adapter = HubMqttAdapter(
-        broker=HUB_MQTT_BROKER_HOST,
-        port=HUB_MQTT_BROKER_PORT,
-        topic=HUB_MQTT_TOPIC,
-    )
-    # Create an instance of the AgentMQTTAdapter using the configuration
+
+    # Logic to select the adapter based on configuration (SCRUM-93 & SCRUM-94)
+    # This allows easy switching between HTTP and MQTT protocols
+    if HUB_CONNECTION_TYPE.lower() == "http":
+        logging.info("Initializing HubHttpAdapter (SCRUM-93 integration)")
+        hub_adapter = HubHttpAdapter(
+            api_base_url=HUB_URL,
+        )
+    else:
+        logging.info("Initializing HubMqttAdapter (SCRUM-94 integration)")
+        hub_adapter = HubMqttAdapter(
+            broker=HUB_MQTT_BROKER_HOST,
+            port=HUB_MQTT_BROKER_PORT,
+            topic=HUB_MQTT_TOPIC,
+        )
+
+    # Create an instance of the AgentMQTTAdapter using the selected hub adapter
+    # This adapter acts as a bridge between the Agent and the Hub
     agent_adapter = AgentMQTTAdapter(
         broker_host=MQTT_BROKER_HOST,
         broker_port=MQTT_BROKER_PORT,
         topic=MQTT_TOPIC,
         hub_gateway=hub_adapter,
     )
+
     try:
-        # Connect to the MQTT broker and start listening for messages
+        logging.info(f"Connecting to MQTT broker at {MQTT_BROKER_HOST}:{MQTT_BROKER_PORT}")
         agent_adapter.connect()
-        agent_adapter.start()
-        # Keep the system running indefinitely (you can add other logic as needed)
-        while True:
-            pass
+
+        logging.info("Broker connection success. Waiting for data...")
+        agent_adapter.loop_forever()
     except KeyboardInterrupt:
-        # Stop the MQTT adapter and exit gracefully if interrupted by the user
-        agent_adapter.stop()
-        logging.info("System stopped.")
+        logging.info("Interrupt signal received. Shutting down...")
+        agent_adapter.disconnect()
+        logging.info("Disconnected from MQTT broker.")

hub/Dockerfile-test

@@ -0,0 +1,12 @@
+# Use the official Python image as the base image
+FROM python:3.9-slim
+# Set the working directory inside the container
+WORKDIR /app
+# Copy the requirements.txt file and install dependencies
+COPY hub/requirements.txt .
+
+RUN pip install --no-cache-dir -r requirements.txt
+# Copy the entire application into the container
+COPY hub/. .
+# Run the main.py script inside the container when it starts
+CMD ["./test-entry.sh"]

hub/app/adapters/store_api_adapter.py

@@ -13,12 +13,39 @@ class StoreApiAdapter(StoreGateway):
     def __init__(self, api_base_url):
         self.api_base_url = api_base_url
 
+    def processed_agent_data_batch_to_payload(self, processed_agent_data_batch: List[ProcessedAgentData]):
+        if not processed_agent_data_batch:
+            return False
+
+        # Extract user_id from the first element
+        user_id = processed_agent_data_batch[0].agent_data.user_id
+
+        payload = {
+            "data": [item.model_dump(mode='json') for item in processed_agent_data_batch],
+            "user_id": user_id
+        }
+
+        return payload
+
     def save_data(self, processed_agent_data_batch: List[ProcessedAgentData]):
-        """
-        Save the processed road data to the Store API.
-        Parameters:
-            processed_agent_data_batch (dict): Processed road data to be saved.
-        Returns:
-            bool: True if the data is successfully saved, False otherwise.
-        """
-        # Implement it
+        payload = self.processed_agent_data_batch_to_payload(processed_agent_data_batch)
+
+        if payload == False:
+            return False
+
+        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/app/adapters/store_api_adapter_test.py

@@ -0,0 +1,41 @@
+from app.adapters.store_api_adapter import StoreApiAdapter
+from app.entities.agent_data import AccelerometerData, AgentData, GpsData
+from app.entities.processed_agent_data import ProcessedAgentData
+
+def _test_processed_agent_data_batch_to_payload():
+    processed_data_batch = [
+            ProcessedAgentData(road_state = "normal",
+                               agent_data = AgentData(user_id = 1,
+                                                      accelerometer = AccelerometerData(x = 0.1, y = 0.2, z = 0.3),
+                                                      gps = GpsData(latitude = 10.123, longitude = 20.456),
+                                                      timestamp = "2023-07-21T12:34:56Z")
+            ),
+            ProcessedAgentData(road_state = "normal",
+                               agent_data = AgentData(user_id = 2,
+                                                      accelerometer = AccelerometerData(x = 0.1, y = 0.2, z = 0.3),
+                                                      gps = GpsData(latitude = 10.123, longitude = 20.456),
+                                                      timestamp = "2023-07-21T12:34:56Z")
+            ),
+            ProcessedAgentData(road_state = "normal",
+                               agent_data = AgentData(user_id = 3,
+                                                      accelerometer = AccelerometerData(x = 0.1, y = 0.2, z = 0.3),
+                                                      gps = GpsData(latitude = 10.123, longitude = 20.456),
+                                                      timestamp = "2023-07-21T12:34:56Z")
+            ),
+    ]
+
+    res = StoreApiAdapter(None).processed_agent_data_batch_to_payload(processed_data_batch)
+
+    assert res["data"][0]["agent_data"]["user_id"] == 1
+    assert res["data"][1]["agent_data"]["user_id"] == 2
+    assert res["data"][2]["agent_data"]["user_id"] == 3
+
+    assert StoreApiAdapter(None).processed_agent_data_batch_to_payload([]) == False
+
+
+if __name__ == "__main__":
+    test_functions = [i for i in dir() if i.startswith('_test_')]
+
+    for i in test_functions:
+        print(i)
+        eval(i)()

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(
-                    redis_client.lpop("processed_agent_data")
-                )
-                processed_agent_data_batch.append(processed_agent_data)
-        store_adapter.save_data(processed_agent_data_batch=processed_agent_data_batch)
+                raw_data = redis_client.lpop("processed_agent_data")
+                if raw_data:
+                    data_item = ProcessedAgentData.model_validate_json(raw_data)
+                    processed_agent_data_batch.append(data_item)
+
+            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}")

hub/test-entry.sh

@@ -0,0 +1,3 @@
+#!/bin/sh
+
+PYTHONPATH=$PWD python3 app/adapters/store_api_adapter_test.py

store/Dockerfile-test

@@ -0,0 +1,13 @@
+# Use the official Python image as the base image
+FROM python:latest
+# Set the working directory inside the container
+WORKDIR /app
+# Copy the requirements.txt file and install dependencies
+COPY store/requirements.txt .
+
+RUN pip install --no-cache-dir -r requirements.txt
+# Copy the entire application into the container
+COPY store/. .
+# Run the main.py script inside the container when it starts
+#CMD ["uvicorn", "main:app", "--host", "0.0.0.0"]
+CMD ["./test-entry.sh"]

store/docker/db/structure.sql

@@ -7,5 +7,6 @@ CREATE TABLE processed_agent_data (
     z FLOAT,
     latitude FLOAT,
     longitude FLOAT,
-    timestamp TIMESTAMP
-);
+    timestamp TIMESTAMP,
+    visible BOOLEAN
+);

store/docker/docker-compose.yaml

@@ -1,61 +0,0 @@
-name: "road_vision__database"
-services:
-  postgres_db:
-    image: postgres:17
-    container_name: postgres_db
-    restart: always
-    environment:
-      POSTGRES_USER: user
-      POSTGRES_PASSWORD: pass
-      POSTGRES_DB: test_db
-    volumes:
-      - postgres_data:/var/lib/postgresql/data
-      - ./db/structure.sql:/docker-entrypoint-initdb.d/structure.sql
-    ports:
-      - "5432:5432"
-    networks:
-      db_network:
-
-
-  pgadmin:
-    container_name: pgadmin4
-    image: dpage/pgadmin4
-    restart: always
-    environment:
-      PGADMIN_DEFAULT_EMAIL: admin@admin.com
-      PGADMIN_DEFAULT_PASSWORD: root
-    volumes:
-      - pgadmin-data:/var/lib/pgadmin
-    ports:
-      - "5050:80"
-    networks:
-      db_network:
-
-
-  store:
-    container_name: store
-    build:
-      context: ../../
-      dockerfile: store/Dockerfile
-    depends_on:
-      - postgres_db
-    restart: always
-    environment:
-      POSTGRES_USER: user
-      POSTGRES_PASSWORD: pass
-      POSTGRES_DB: test_db
-      POSTGRES_HOST: postgres_db
-      POSTGRES_PORT: 5432
-    ports:
-      - "8000:8000"
-    networks:
-      db_network:
-
-
-networks:
-  db_network:
-
-
-volumes:
-  postgres_data:
-  pgadmin-data:

store/main.py

@@ -8,12 +8,13 @@ from sqlalchemy import (
     Integer,
     String,
     Float,
+    Boolean,
     DateTime,
 )
 from sqlalchemy.sql import select
 
 from database import metadata, SessionLocal
-from schemas import ProcessedAgentData, ProcessedAgentDataInDB
+from schemas import ProcessedAgentData, ProcessedAgentDataInDB, WebSocketData
 
 # FastAPI app setup
 app = FastAPI()
@@ -30,60 +31,95 @@ processed_agent_data = Table(
     Column("latitude", Float),
     Column("longitude", Float),
     Column("timestamp", DateTime),
+    Column("visible", Boolean),
 )
 
 # WebSocket subscriptions
-subscriptions: Dict[int, Set[WebSocket]] = {}
+subscriptions: Set[WebSocket] = set()
 
 
 # FastAPI WebSocket endpoint
 @app.websocket("/ws/{user_id}")
 async def websocket_endpoint(websocket: WebSocket, user_id: int):
     await websocket.accept()
-    if user_id not in subscriptions:
-        subscriptions[user_id] = set()
-    subscriptions[user_id].add(websocket)
+
+    subscriptions.add(websocket)
+
     try:
+        # send already available data
+        r = processed_agent_data.select().order_by(processed_agent_data.c.timestamp)
+        session = SessionLocal()
+        stored_data = session.execute(r).fetchall()
+        session.close() 
+
+        jsonable_data = [{c.name: getattr(i, c.name) for c in processed_agent_data.columns} for i in stored_data]
+        for i in jsonable_data:
+            i['timestamp'] = i['timestamp'].strftime("%Y-%m-%dT%H:%M:%SZ")
+
+        for i in jsonable_data:
+            await websocket.send_json(json.dumps([i]))
+
+        # receive forever
         while True:
-            await websocket.receive_text()
+            data = await websocket.receive_text()
+            try:
+                if (data):
+                    ws_data = WebSocketData.model_validate(json.loads(data))
+                    session = SessionLocal()
+                    update_query = (
+                        processed_agent_data.update()
+                        .where(processed_agent_data.c.id == ws_data.id)
+                        .values(visible=False)
+                    ).returning(processed_agent_data)
+                    res = session.execute(update_query).fetchone()
+                    if (not res):
+                        session.rollback()
+                        raise Exception("Error while websocket PUT")
+                    session.commit()
+            finally:
+                session.close()                  
+         
     except WebSocketDisconnect:
-        subscriptions[user_id].remove(websocket)
+        subscriptions.remove(websocket)
 
 
 # Function to send data to subscribed users
-async def send_data_to_subscribers(user_id: int, data):
-    if user_id in subscriptions:
-        for websocket in subscriptions[user_id]:
-            await websocket.send_json(json.dumps(data))
+async def send_data_to_subscribers(data):
+    for websocket in subscriptions:
+        await websocket.send_json(json.dumps([data]))
 
 
 # FastAPI CRUDL endpoints
 
-
-@app.post("/processed_agent_data/")
-async def create_processed_agent_data(data: List[ProcessedAgentData], user_id: int = Body(..., embed=True)):
-    session = SessionLocal()
-    try:
-        created_data = [
+def ProcessedAgentData_to_td(data: List[ProcessedAgentData]):
+        return [
             {
                 "road_state": item.road_state,
-                "user_id": user_id,
+                "user_id": item.agent_data.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,
+                "visible": True,
             }
             for item in data
         ]
+
+
+@app.post("/processed_agent_data/")
+async def create_processed_agent_data(data: List[ProcessedAgentData], user_id: int = Body(..., embed=True)):
+    session = SessionLocal()
+    try:
+        created_data = ProcessedAgentData_to_td(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))
+        for record in sorted(created_records, key = lambda x: x['timestamp']):
+            await send_data_to_subscribers(jsonable_encoder(record))
         return created_records
     except Exception as err:
         session.rollback()
@@ -164,8 +200,12 @@ def update_processed_agent_data(processed_agent_data_id: int, data: ProcessedAge
         session.commit()
 
         updated_result = session.execute(query).fetchone()
-
         return ProcessedAgentDataInDB(**updated_result._mapping)
+        
+    except Exception as err:
+        session.rollback()
+        print(f"Database error: {err}")
+        raise HTTPException(status_code=500, detail="Internal Server Error")
 
     finally:
         session.close()
@@ -196,7 +236,12 @@ def delete_processed_agent_data(processed_agent_data_id: int):
         session.commit()
 
         return ProcessedAgentDataInDB(**result._mapping)
-
+        
+    except Exception as err:
+        session.rollback()
+        print(f"Database error: {err}")
+        raise HTTPException(status_code=500, detail="Internal Server Error")
+        
     finally:
         session.close()
 

store/schemas.py

@@ -49,3 +49,7 @@ class AgentData(BaseModel):
 class ProcessedAgentData(BaseModel):
     road_state: str
     agent_data: AgentData
+    
+class WebSocketData(BaseModel):
+    id: int
+

store/test-entry.sh

@@ -0,0 +1,3 @@
+#!/bin/sh
+
+PYTHONPATH=$PWD python3 test/main_test.py

store/test/main_test.py

@@ -0,0 +1,39 @@
+from schemas import AccelerometerData, AgentData, GpsData, ProcessedAgentData
+
+import main
+
+def _test_ProcessedAgentData_to_td():
+    processed_data_batch = [
+            ProcessedAgentData(road_state = "normal",
+                               agent_data = AgentData(user_id = 1,
+                                                      accelerometer = AccelerometerData(x = 0.1, y = 0.2, z = 0.3),
+                                                      gps = GpsData(latitude = 10.123, longitude = 20.456),
+                                                      timestamp = "2023-07-21T12:34:56Z")
+            ),
+            ProcessedAgentData(road_state = "normal",
+                               agent_data = AgentData(user_id = 2,
+                                                      accelerometer = AccelerometerData(x = 0.1, y = 0.2, z = 0.3),
+                                                      gps = GpsData(latitude = 10.123, longitude = 20.456),
+                                                      timestamp = "2023-07-21T12:34:56Z")
+            ),
+            ProcessedAgentData(road_state = "normal",
+                               agent_data = AgentData(user_id = 3,
+                                                      accelerometer = AccelerometerData(x = 0.1, y = 0.2, z = 0.3),
+                                                      gps = GpsData(latitude = 10.123, longitude = 20.456),
+                                                      timestamp = "2023-07-21T12:34:56Z")
+            ),
+    ]
+
+    res = main.ProcessedAgentData_to_td(processed_data_batch)
+
+    assert res[0]["user_id"] == 1
+    assert res[1]["user_id"] == 2
+    assert res[2]["user_id"] == 3
+
+
+if __name__ == "__main__":
+    test_functions = [i for i in dir() if i.startswith('_test_')]
+
+    for i in test_functions:
+        print(i)
+        eval(i)()

utils/check-up.py

@@ -0,0 +1,23 @@
+import sys
+import os
+
+print("Checking for dead containers...")
+
+l = [i for i in sys.stdin.read().split("\n") if i]
+header, statuses = l[0], l[1:]
+
+status_index = header.find('STATUS')
+name_index = header.find('NAMES')
+
+exit_code = 0
+
+for i in statuses:
+    if not i[status_index:].startswith("Up "):
+        service_name = i[name_index:]
+        print(f"Crash detected in {service_name}")
+        print(f"docker logs for the container:\n")
+        os.system(f"docker logs {i.split(' ')[0]}")
+        print()
+        exit_code = 1
+
+sys.exit(exit_code)