Compare commits

..

4 Commits

23 changed files with 141 additions and 306 deletions
-16
View File
@@ -1,16 +0,0 @@
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
-71
View File
@@ -1,71 +0,0 @@
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
+1 -1
View File
@@ -32,7 +32,7 @@ class MapViewApp(App):
Встановлює необхідні маркери, викликає функцію для оновлення мапи
"""
self.update()
Clock.schedule_interval(self.update, 0.1)
Clock.schedule_interval(self.update, 5)
def update(self, *args):
"""
+1 -1
View File
@@ -8,7 +8,7 @@ def try_parse(type, value: str):
return None
USER_ID = try_parse(int, os.environ.get("USER_ID")) or 1
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
-2
View File
@@ -1,7 +1,6 @@
from paho.mqtt import client as mqtt_client
from schema.aggregated_data_schema import AggregatedDataSchema
from file_datasource import FileDatasource
import logging
import config
@@ -29,7 +28,6 @@ def publish(client, topic, datasource):
data = datasource.read()
msg = AggregatedDataSchema().dumps(data)
result = client.publish(topic, msg)
logging.info(f"Published to {topic}: {msg[:50]}...")
status = result[0]
if status != 0:
print(f"Failed to send message to topic {topic}")
+30
View File
@@ -0,0 +1,30 @@
@startuml
participant Agent as agent
participant "MQTT Broker (raw\ntopic)" as mqtt1
participant "Edge Service" as edge
participant "MQTT\nBroker (processed\ntopic)" as mqtt2
participant "Hub Service" as hub
participant "Redis" as redis
participant "Store API" as store
participant "PostgreSQL" as db
participant "MapView Client" as mapview
agent -> mqtt1 : Publish raw\ntelemetry
mqtt1 -> edge : Deliver raw\nmessage
edge -> edge : Validate\nAgentData
edge -> edge : Process\ntelemetry
edge -> mqtt2 : Publish\nprocessed data
mqtt2 -> hub : Deliver processed\nmessage
hub -> hub : Validate\nProcessedAgentData
hub -> redis : LPUSH to buffer
hub -> redis : LPOP batch item
redis -> hub : Return item
hub -> store : POST batch
store -> db : INSERT records
db --> store : Return created\nrecords
store -> mapview : WebSocket push
mapview -> mapview : Sort by timestamp
mapview -> mapview : Update vehicle\nmarker
mapview -> mapview : Add pothole/bump\nmarker
store --> hub : Success response
@enduml
+23
View File
@@ -0,0 +1,23 @@
@startuml
rectangle IoT-Systems {
usecase "Collect telemetry (accelerometer + GPS)" as uc1
usecase "Send telemetry" as uc2
usecase "Process telemetry" as uc3
usecase "Determine road condition (pothole / bump /\nnormal)" as uc4
usecase "View road defect marks" as uc5
usecase "View route on map" as uc6
}
rectangle "The user is the card operator" as uc10
rectangle "Sensor Agent\n(Device/STM32/Emulator)" as uc11
uc11 - uc1
uc11 - uc2
uc10 - uc5
uc10 - uc6
uc2 -.|> uc3 : <<include>>
uc3 -.|> uc4 : <<include>>
@enduml
+2 -6
View File
@@ -22,7 +22,6 @@ services:
depends_on:
- mqtt
environment:
PYTHONUNBUFFERED: 1
MQTT_BROKER_HOST: "mqtt"
MQTT_BROKER_PORT: 1883
MQTT_TOPIC: "agent_data_topic"
@@ -40,10 +39,9 @@ services:
environment:
MQTT_BROKER_HOST: "mqtt"
MQTT_BROKER_PORT: 1883
MQTT_TOPIC: "agent_data_topic"
HUB_HOST: "hub"
MQTT_TOPIC: " "
HUB_HOST: "store"
HUB_PORT: 8000
HUB_CONNECTION_TYPE: "http"
HUB_MQTT_BROKER_HOST: "mqtt"
HUB_MQTT_BROKER_PORT: 1883
HUB_MQTT_TOPIC: "processed_data_topic"
@@ -91,7 +89,6 @@ services:
- postgres_db
restart: always
environment:
PYTHONUNBUFFERED: 1
POSTGRES_USER: user
POSTGRES_PASSWORD: pass
POSTGRES_DB: test_db
@@ -123,7 +120,6 @@ services:
- redis
- store
environment:
PYTHONUNBUFFERED: 1
STORE_API_HOST: "store"
STORE_API_PORT: 8000
REDIS_HOST: "redis"
+32 -9
View File
@@ -13,7 +13,9 @@ 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
@@ -33,21 +35,42 @@ 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)
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.")
# 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")
except Exception as e:
logging.error(f"Error processing MQTT message: {e}")
logging.info(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 loop_forever(self):
self.client.loop_forever()
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.")
-1
View File
@@ -14,7 +14,6 @@ class GpsData(BaseModel):
class AgentData(BaseModel):
user_id: int
accelerometer: AccelerometerData
gps: GpsData
timestamp: datetime
+9 -2
View File
@@ -26,8 +26,15 @@ class AgentGateway(ABC):
pass
@abstractmethod
def loop_forever(self):
def start(self):
"""
Method to await for new messages.
Method to start listening for messages from the agent.
"""
pass
@abstractmethod
def stop(self):
"""
Method to stop the agent gateway and clean up resources.
"""
pass
-4
View File
@@ -13,7 +13,3 @@ def process_agent_data(
processed_data_batch (ProcessedAgentData): Processed data containing the classified state of the road surface and agent data.
"""
# Implement it
return ProcessedAgentData(
road_state="normal",
agent_data=agent_data
)
+2 -5
View File
@@ -16,12 +16,9 @@ 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_data_topic"
HUB_MQTT_TOPIC = os.environ.get("HUB_MQTT_TOPIC") or "processed_agent_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 8000
HUB_PORT = try_parse_int(os.environ.get("HUB_PORT")) or 12000
HUB_URL = f"http://{HUB_HOST}:{HUB_PORT}"
# For choosing type of connection
HUB_CONNECTION_TYPE = os.environ.get("HUB_CONNECTION_TYPE") or "mqtt"
+21 -30
View File
@@ -10,51 +10,42 @@ 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,
level=logging.INFO, # Set the log level to INFO (you can use logging.DEBUG for more detailed logs)
format="[%(asctime)s] [%(levelname)s] [%(module)s] %(message)s",
handlers=[
logging.StreamHandler(),
logging.FileHandler("app.log"),
logging.StreamHandler(), # Output log messages to the console
logging.FileHandler("app.log"), # Save log messages to a file
],
)
# 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
# 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
agent_adapter = AgentMQTTAdapter(
broker_host=MQTT_BROKER_HOST,
broker_port=MQTT_BROKER_PORT,
topic=MQTT_TOPIC,
hub_gateway=hub_adapter,
)
try:
logging.info(f"Connecting to MQTT broker at {MQTT_BROKER_HOST}:{MQTT_BROKER_PORT}")
# Connect to the MQTT broker and start listening for messages
agent_adapter.connect()
logging.info("Broker connection success. Waiting for data...")
agent_adapter.loop_forever()
agent_adapter.start()
# Keep the system running indefinitely (you can add other logic as needed)
while True:
pass
except KeyboardInterrupt:
logging.info("Interrupt signal received. Shutting down...")
agent_adapter.disconnect()
logging.info("Disconnected from MQTT broker.")
# Stop the MQTT adapter and exit gracefully if interrupted by the user
agent_adapter.stop()
logging.info("System stopped.")
-12
View File
@@ -1,12 +0,0 @@
# 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"]
+1 -9
View File
@@ -13,7 +13,7 @@ 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]):
def save_data(self, processed_agent_data_batch: List[ProcessedAgentData]):
if not processed_agent_data_batch:
return False
@@ -25,14 +25,6 @@ class StoreApiAdapter(StoreGateway):
"user_id": user_id
}
return payload
def save_data(self, processed_agent_data_batch: List[ProcessedAgentData]):
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(
@@ -1,41 +0,0 @@
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)()
-3
View File
@@ -1,3 +0,0 @@
#!/bin/sh
PYTHONPATH=$PWD python3 app/adapters/store_api_adapter_test.py
-13
View File
@@ -1,13 +0,0 @@
# 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"]
+19 -19
View File
@@ -33,7 +33,7 @@ processed_agent_data = Table(
)
# WebSocket subscriptions
subscriptions: Set[WebSocket] = set()
subscriptions: Dict[int, Set[WebSocket]] = {}
# FastAPI WebSocket endpoint
@@ -41,7 +41,10 @@ subscriptions: Set[WebSocket] = set()
async def websocket_endpoint(websocket: WebSocket, user_id: int):
await websocket.accept()
subscriptions.add(websocket)
if user_id not in subscriptions:
subscriptions[user_id] = set()
subscriptions[user_id].add(websocket)
try:
# send already available data
@@ -52,29 +55,33 @@ async def websocket_endpoint(websocket: WebSocket, user_id: int):
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]))
await websocket.send_json(json.dumps(jsonable_data))
# receive forever
while True:
await websocket.receive_text()
except WebSocketDisconnect:
subscriptions.remove(websocket)
subscriptions[user_id].remove(websocket)
# Function to send data to subscribed users
async def send_data_to_subscribers(data):
for websocket in subscriptions:
await websocket.send_json(json.dumps([data]))
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))
# FastAPI CRUDL endpoints
def ProcessedAgentData_to_td(data: List[ProcessedAgentData]):
return [
@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 = [
{
"road_state": item.road_state,
"user_id": item.agent_data.user_id,
"user_id": user_id,
"x": item.agent_data.accelerometer.x,
"y": item.agent_data.accelerometer.y,
"z": item.agent_data.accelerometer.z,
@@ -84,20 +91,13 @@ def ProcessedAgentData_to_td(data: List[ProcessedAgentData]):
}
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(jsonable_encoder(record))
await send_data_to_subscribers(user_id, jsonable_encoder(record))
return created_records
except Exception as err:
session.rollback()
-3
View File
@@ -1,3 +0,0 @@
#!/bin/sh
PYTHONPATH=$PWD python3 test/main_test.py
-39
View File
@@ -1,39 +0,0 @@
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)()
-19
View File
@@ -1,19 +0,0 @@
import sys
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}")
exit_code = 1
sys.exit(exit_code)