· Python· dataclasses· IoT· SaaS· Pydantic· NamedTuple· frozen· slots
Python dataclasses til IoT platform — komplet guide
Python dataclasses til IoT og SaaS: @dataclass, field(), __post_init__, frozen immutable records, slots=True, ClassVar og sammenligning med Pydantic og NamedTuple.
Af M-Bus Gateway
Python dataclasses eliminerer boilerplate-kode til dataobjekter. Her er de produktionsklare mønstre fra IoT-platformen.
Grundlæggende @dataclass
# gateway/src/wmbus/models.py
from dataclasses import dataclass, field
from datetime import datetime
from decimal import Decimal
@dataclass
class WmbusTelegram:
"""
Parsed wM-Bus telegram fra wmbusmeters.
Oprettes fra JSON output — kan ikke ændres efter parsing.
"""
meter_id: str
fab_id: str # FAB ID: KAM, ELS, APT, etc.
driver: str # wmbusmeters driver-navn
medium: str # heat, water, gas, electric
timestamp: datetime
value: Decimal
unit: str # MWh, m3, kWh
rssi: int | None = None
lqi: int | None = None
aes_decrypted: bool = False
raw_frame: bytes = field(default_factory=bytes, repr=False)
def __post_init__(self):
"""Normalisér feltværdier efter construction."""
self.fab_id = self.fab_id.upper()
self.meter_id = self.meter_id.strip()
if self.rssi is not None and self.rssi > 0:
# RSSI fra visse fabrikanter rapporteres positivt
self.rssi = -self.rssi
# Instantiering:
telegram = WmbusTelegram(
meter_id="12345678",
fab_id="kam", # normaliseres til "KAM" i __post_init__
driver="multical302",
medium="heat",
timestamp=datetime.utcnow(),
value=Decimal("15.234"),
unit="MWh",
rssi=-68,
)
field() og komplekse defaults
from dataclasses import dataclass, field
from typing import ClassVar
from decimal import Decimal
@dataclass
class GatewayConfig:
"""
Gateway-konfiguration — læses fra SQLite config-tabel.
Bruges som immutable snapshot under send-session.
"""
gateway_id: str
mqtt_host: str
mqtt_port: int = 8883
send_cron: str = "0 6 * * *"
wmbus_mode: str = "c1"
log_level: str = "INFO"
# Komplekse defaults med field(default_factory=):
allowed_meter_ids: list[str] = field(default_factory=list)
aes_keys: dict[str, str] = field(default_factory=dict, repr=False)
# ClassVar — ikke et dataclass-felt, kun klassevariabel:
VALID_WMBUS_MODES: ClassVar[frozenset[str]] = frozenset({"c1", "t1", "s1", "c1t1"})
def __post_init__(self):
if self.wmbus_mode not in self.VALID_WMBUS_MODES:
raise ValueError(
f"Ugyldig wmbus_mode: {self.wmbus_mode}. Vælg: {self.VALID_WMBUS_MODES}"
)
@property
def has_aes_keys(self) -> bool:
return len(self.aes_keys) > 0
# Fejlagtig default — ALDRIG mutable default direkte:
# aes_keys: dict = {} # TypeError: mutable default
# aes_keys: dict = field(default_factory=dict) # KORREKT
frozen=True — immutable records
from dataclasses import dataclass
from decimal import Decimal
from datetime import date
@dataclass(frozen=True)
class PeriodKey:
"""
Unik nøgle til én afregningsperiode for én lejlighed.
Frozen → hashbar → kan bruges som dict-nøgle eller i sets.
"""
property_id: str
unit_id: str
period_start: date
period_end: date
def __str__(self) -> str:
return f"{self.unit_id}:{self.period_start}→{self.period_end}"
@dataclass(frozen=True)
class AllocationResult:
"""Beregningsresultat pr. lejlighed — immutable efter beregning."""
unit_id: str
fixed_amount: Decimal
variable_amount: Decimal
@property
def total(self) -> Decimal:
return self.fixed_amount + self.variable_amount
@property
def balance(self) -> Decimal:
"""Positiv = efterbetaling, negativ = tilbagebetaling."""
return self.total # Subtraheres med aconto_paid i consumer
# Hashbart → brugbart i sets og dict-nøgler:
results: dict[PeriodKey, AllocationResult] = {}
key = PeriodKey("prop-1", "unit-A", date(2025, 6, 1), date(2026, 5, 31))
results[key] = AllocationResult("unit-A", Decimal("1200"), Decimal("4300"))
# Forsøg på mutation → FrozenInstanceError:
# result.fixed_amount = Decimal("0") # Fejl
slots=True — memory-optimering
from dataclasses import dataclass
import sys
# Uden slots: ~200 bytes pr. instans (via __dict__)
@dataclass
class ReadingBasic:
meter_id: str
timestamp: float
value: float
# Med slots: ~56 bytes pr. instans (direkte slots, ingen __dict__)
@dataclass(slots=True)
class Reading:
"""
Aflæsning — gemmes i RAM-buffer (op til 50.000 ad gangen).
slots=True sparer ~150 bytes pr. instans = ~7.5 MB pr. 50K readings.
"""
meter_installation_id: str
timestamp: float
value: float
unit: str
rssi: int | None = None
# Memory sammenligning:
r_basic = ReadingBasic("12345678", 1716500000.0, 15.234)
r_slots = Reading("inst-uuid", 1716500000.0, 15.234, "MWh")
print(sys.getsizeof(r_basic)) # ~200 bytes (inkl. __dict__)
print(sys.getsizeof(r_slots)) # ~56 bytes (slots kun)
# OBS: slots=True og arv kræver opmærksomhed:
# Subklasse skal også have slots=True eller arver __dict__
dataclass vs Pydantic vs NamedTuple
# Sammenligning af tre data-objekttyper:
# 1. dataclass — Python stdlib, hurtig, ingen validering
@dataclass
class TelegramDC:
meter_id: str
value: float # Ingen automatisk type-konvertering
# 2. Pydantic BaseModel — validering + serialisering, lidt langsommere
from pydantic import BaseModel
class TelegramPydantic(BaseModel):
meter_id: str
value: float # Konverterer automatisk "15.3" → 15.3
# model_dump(), model_validate(), JSON serialization built-in
# 3. NamedTuple — immutable, hurtig, hashbar
from typing import NamedTuple
class TelegramNT(NamedTuple):
meter_id: str
value: float
# Upackable: meter_id, value = telegram
# Index-adgang: telegram[0]
# Vores platform bruger:
# dataclass → interne beregningsresultater (AlloationResult, PeriodKey)
# Pydantic → API schemas (in/out), request/response validering
# NamedTuple → sjældent — kun ved DB query-resultater der unpackes
# Konvertering dataclass → dict:
from dataclasses import asdict, astuple
result = AllocationResult("unit-A", Decimal("1200"), Decimal("4300"))
d = asdict(result) # {"unit_id": "unit-A", "fixed_amount": Decimal("1200"), ...}
t = astuple(result) # ("unit-A", Decimal("1200"), Decimal("4300"))
post_init til validering og afledning
from dataclasses import dataclass, field
from decimal import Decimal
import uuid
@dataclass
class UnitAllocation:
"""
Beregnet fordeling til én lejlighed.
__post_init__ validerer og beregner afledte felter.
"""
unit_id: str
floor_area_m2: Decimal
hca_units: Decimal
aconto_paid: Decimal
fixed_amount: Decimal = field(init=False) # Beregnes i __post_init__
variable_amount: Decimal = field(init=False)
# Bruges internt til fordeling — ikke et offentligt felt:
_fixed_pool: Decimal = field(default=Decimal("0"), repr=False)
_variable_pool: Decimal = field(default=Decimal("0"), repr=False)
_total_area: Decimal = field(default=Decimal("1"), repr=False)
_total_hca: Decimal = field(default=Decimal("1"), repr=False)
def __post_init__(self):
if self.floor_area_m2 <= 0:
raise ValueError(f"Areal skal være positivt: {self.floor_area_m2}")
if self.hca_units < 0:
raise ValueError(f"HCA-enheder kan ikke være negative: {self.hca_units}")
# Beregn fordeling:
self.fixed_amount = (
self._fixed_pool * self.floor_area_m2 / self._total_area
).quantize(Decimal("0.01"))
self.variable_amount = (
self._variable_pool * self.hca_units / self._total_hca
).quantize(Decimal("0.01"))
@property
def total(self) -> Decimal:
return self.fixed_amount + self.variable_amount
@property
def balance(self) -> Decimal:
"""Positiv = efterbetaling, negativ = tilbagebetaling."""
return self.total - self.aconto_paid
Konklusion
@dataclass er den rette løsning til interne beregningsobjekter (aflæsninger, fordelingsnøgler, konfigurationer) hvor validering ikke er kritisk. frozen=True giver hashbare immutable records — ideelt til dict-nøgler og set-membership. slots=True reducerer memory med ~75% per instans — afgørende når IoT-platformen bufferer tusindvis af Reading-objekter i RAM. Brug Pydantic til API-grænseflader (JSON ↔ Python med validering) og dataclass til interne strukturer.