c085a49c8c
- Agent: Updated config and main - Edge: Implemented adapter factory in main.py to switch between MQTT and HTTP. - Edge: Updated AgentData entity and processing logic to support user_id. - Infrastructure: Configured docker-compose for dynamic protocol switching and environment management.
34 lines
810 B
Python
34 lines
810 B
Python
from datetime import datetime
|
|
from pydantic import BaseModel, field_validator
|
|
|
|
|
|
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 parse_timestamp(cls, value):
|
|
# Convert the timestamp to a datetime object
|
|
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)."
|
|
)
|