71 lines
3.0 KiB
Python
71 lines
3.0 KiB
Python
# ================================================================
|
|
# services/data/app/schemas/traffic.py
|
|
# ================================================================
|
|
"""Traffic data schemas"""
|
|
|
|
from pydantic import BaseModel, Field, validator
|
|
from datetime import datetime
|
|
from typing import Optional, List
|
|
from uuid import UUID
|
|
|
|
class TrafficDataBase(BaseModel):
|
|
"""Base traffic data schema"""
|
|
location_id: str = Field(..., max_length=100, description="Traffic monitoring location ID")
|
|
date: datetime = Field(..., description="Date and time of traffic measurement")
|
|
traffic_volume: Optional[int] = Field(None, ge=0, description="Vehicles per hour")
|
|
pedestrian_count: Optional[int] = Field(None, ge=0, description="Pedestrians per hour")
|
|
congestion_level: Optional[str] = Field(None, regex="^(low|medium|high)$", description="Traffic congestion level")
|
|
average_speed: Optional[float] = Field(None, ge=0, le=200, description="Average speed in km/h")
|
|
source: str = Field("madrid_opendata", max_length=50, description="Data source")
|
|
raw_data: Optional[str] = Field(None, description="Raw data from source")
|
|
|
|
class TrafficDataCreate(TrafficDataBase):
|
|
"""Schema for creating traffic data"""
|
|
pass
|
|
|
|
class TrafficDataUpdate(BaseModel):
|
|
"""Schema for updating traffic data"""
|
|
traffic_volume: Optional[int] = Field(None, ge=0)
|
|
pedestrian_count: Optional[int] = Field(None, ge=0)
|
|
congestion_level: Optional[str] = Field(None, regex="^(low|medium|high)$")
|
|
average_speed: Optional[float] = Field(None, ge=0, le=200)
|
|
raw_data: Optional[str] = None
|
|
|
|
class TrafficDataResponse(TrafficDataBase):
|
|
"""Schema for traffic data responses"""
|
|
id: str = Field(..., description="Unique identifier")
|
|
created_at: datetime = Field(..., description="Creation timestamp")
|
|
updated_at: datetime = Field(..., description="Last update timestamp")
|
|
|
|
@validator('id', pre=True)
|
|
def convert_uuid_to_string(cls, v):
|
|
if isinstance(v, UUID):
|
|
return str(v)
|
|
return v
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
json_encoders = {
|
|
datetime: lambda v: v.isoformat()
|
|
}
|
|
|
|
class TrafficDataList(BaseModel):
|
|
"""Schema for paginated traffic data responses"""
|
|
data: List[TrafficDataResponse]
|
|
total: int = Field(..., description="Total number of records")
|
|
page: int = Field(..., description="Current page number")
|
|
per_page: int = Field(..., description="Records per page")
|
|
has_next: bool = Field(..., description="Whether there are more pages")
|
|
has_prev: bool = Field(..., description="Whether there are previous pages")
|
|
|
|
class TrafficAnalytics(BaseModel):
|
|
"""Schema for traffic analytics"""
|
|
location_id: str
|
|
period_start: datetime
|
|
period_end: datetime
|
|
avg_traffic_volume: Optional[float] = None
|
|
avg_pedestrian_count: Optional[float] = None
|
|
peak_traffic_hour: Optional[int] = None
|
|
peak_pedestrian_hour: Optional[int] = None
|
|
congestion_distribution: dict = Field(default_factory=dict)
|
|
avg_speed: Optional[float] = None |