Space Update Model

Whatever system you use to send parking space occupancy updates, it must send json payloads to your site's /space/update-state endpoint.

You must create a subclass of SpaceUpdateBaseModel for your specific update payload. This subclass is passed to build_app() and is used to validate requests that hit the endpoint.

I recommended using the excellent tool datamodel-code-generator to automatically generate the pydantic model from a real json payload for your system.

I like running it with these options:

datamodel-codegen \
--input "your-update-payload.json" \
--input-file-type json \
--output "your_update_class.py" \
--class-name "YourUpdateClass" \
--output-model-type pydantic_v2.BaseModel \
--target-python-version 3.13 \
--use-standard-collections \
--use-union-operator \
--reuse-model \
--use-double-quotes \
--enum-field-as-literal all \
--use-one-literal-as-default \
--collapse-root-models \
--wrap-string-literal \
--disable-timestamp

Here's a working example generated for the car sensor I'm using:

nwave_parking_sensor.py

"""
A real working example of a `SpaceUpdateBaseModel` subclass: the Nwave Smart Parking
Sensor G4 SM.
https://www.thethingsnetwork.org/device-repository/devices/nwave/nps405sm/

This is the LoRa device I'm using for the deployment that inspired this
framework. This file was autogenerated from the json payload using `datamodel-codegen`
and then slightly embellished: annotated the timestamps as `AwareDatetime`, made the
model inherit from `SpaceUpdateBaseModel` and defined its subclass methods.

I'm using the free, open-source LoRaWAN network server hosted by The Things Network
(TTN), with a webhook to send the payload on to my app endpoint. Note that it was
necessary to define an Uplink "Payload Formatter" function in TTN console to make the
TTN Application Server automatically parse the raw payload bytes into a human-interpretable
`DecodedPayload` model. TTN keeps a repository of these formatter functions for common devices.
"""

from __future__ import annotations

from typing import Literal

from pydantic import AwareDatetime, BaseModel, Field

from park_it.models.space_update import LowerStr, SpaceUpdateBaseModel


class ApplicationIds(BaseModel):
    application_id: str


class DeviceIds(BaseModel):
    device_id: str
    application_ids: ApplicationIds
    dev_eui: str
    join_eui: str
    dev_addr: str


class Identifier(BaseModel):
    device_ids: DeviceIds


class DecodedPayload(BaseModel):
    """The meat, decoded from raw payload bytes using the default TTN
    payload formatter for this sensor."""

    occupied: bool
    previous_state_duration: int
    previous_state_duration_error: int
    previous_state_duration_overflow: bool
    # TODO maybe operate on the periodic heartbeat message too
    type: Literal["parking_status"]


class GatewayIds(BaseModel):
    gateway_id: str
    eui: str


class Location(BaseModel):
    latitude: float
    longitude: float
    source: str


class RxMetadatum(BaseModel):
    gateway_ids: GatewayIds
    timestamp: int
    rssi: int
    channel_rssi: int
    snr: float
    frequency_offset: str
    location: Location
    uplink_token: str
    channel_index: int
    received_at: AwareDatetime


class Lora(BaseModel):
    bandwidth: int
    spreading_factor: int
    coding_rate: str


class DataRate(BaseModel):
    lora: Lora


class Settings(BaseModel):
    data_rate: DataRate
    frequency: str
    timestamp: int


class NetworkIds(BaseModel):
    net_id: str
    ns_id: str
    tenant_id: str
    cluster_id: str
    cluster_address: str


class UplinkMessage(BaseModel):
    session_key_id: str
    f_port: int
    f_cnt: int
    frm_payload: str
    decoded_payload: DecodedPayload
    rx_metadata: list[RxMetadatum]
    settings: Settings
    received_at: AwareDatetime
    confirmed: bool
    consumed_airtime: str
    network_ids: NetworkIds


class Data(BaseModel):
    field_type: str = Field(..., alias="@type")
    end_device_ids: DeviceIds
    correlation_ids: list[str]
    received_at: AwareDatetime
    uplink_message: UplinkMessage


class Context(BaseModel):
    tenant_id: str = Field(..., alias="tenant-id")


class Visibility(BaseModel):
    rights: list[str]


class NwaveParkingSensorUpdate(SpaceUpdateBaseModel):
    """
    Subclass generated from the full payload bundle for the Nwave Smart Parking Sensor
    G4 SM, with subclass methods implemented for grabbing required data.
    """

    name: str
    time: AwareDatetime
    identifiers: list[Identifier]
    data: Data
    correlation_ids: list[str]
    origin: str
    context: Context
    visibility: Visibility
    unique_id: str

    def sensor_id(self) -> LowerStr:
        return self.identifiers[0].device_ids.device_id

    def occupied(self) -> bool:
        return self.data.uplink_message.decoded_payload.occupied

    def update_time(self) -> AwareDatetime:
        return self.data.uplink_message.received_at