hasslesstech/IoT-Systems
1
0
Fork 0
Code Issues Pull Requests Actions Activity

[P] Use state machine approach to determine road condition

Browse Source
This commit is contained in:
ІО-23 Шмуляр Олег
2026-03-25 12:00:55 +02:00
parent d6e094e6c0
commit 0695e3d092
1 changed files with 13 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
edge/app/usecases/data_processing.py
View File

@@ -1,8 +1,7 @@
import time
from app.entities.agent_data import AgentData from app.entities.agent_data import AgentData
from app.entities.processed_agent_data import ProcessedAgentData from app.entities.processed_agent_data import ProcessedAgentData
_last_processed_times = {} _last_detection_state = {}
def process_agent_data( def process_agent_data(
agent_data: AgentData, agent_data: AgentData,
@@ -17,16 +16,19 @@ def process_agent_data(
user_id = agent_data.user_id user_id = agent_data.user_id
road_state = "normal" road_state = "normal"
curr_time = time.time() last_detection_state = _last_detection_state.get(user_id, False)
last_processed_time = _last_processed_times.get(user_id, 0)
if curr_time - last_processed_time > 1.0:
if agent_data.accelerometer.z < -1.0:
road_state = "pothole"
elif agent_data.accelerometer.z > 1.0:
road_state = "bump"
_last_processed_times[user_id] = curr_time 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( return ProcessedAgentData(
road_state=road_state, road_state=road_state,