Python Enum guide — IoT status og tilstandsmaskiner
Python Enum til IoT-platform: StrEnum, IntEnum, tilstandsmaskine med valid transitions, Pydantic-integration, SQLModel enum-kolonner og gateway-status tracking med Flag-enums.
Af M-Bus Gateway
Enum i Python giver typesikker tilstandsrepræsentation. Her er de praktiske mønstre til IoT gateway-status, afregningsstatus og tilstandsmaskiner.
StrEnum til API og database
# server/src/db/enums.py
# Python 3.11+: StrEnum er built-in
from enum import StrEnum, IntEnum, Flag, auto
class GatewayStatus(StrEnum):
"""Gateway online/offline status."""
ONLINE = "online"
OFFLINE = "offline"
DEGRADED = "degraded" # Sender men med fejl
PROVISIONING = "provisioning" # Under enrollment
RETIRED = "retired"
class SettlementStatus(StrEnum):
"""Afregnings livscyklus."""
DRAFT = "draft"
GENERATING = "generating"
GENERATED = "generated"
SENT = "sent"
VIEWED = "viewed"
PAID = "paid"
CLOSED = "closed"
DISPUTED = "disputed"
class MeterType(StrEnum):
HEAT = "heat"
WATER_COLD = "water_cold"
WATER_HOT = "water_hot"
ELECTRICITY = "electricity"
GAS = "gas"
HCA = "hca"
class AlertSeverity(StrEnum):
INFO = "info"
WARNING = "warning"
CRITICAL = "critical"
# StrEnum serialiserer til string automatisk:
status = GatewayStatus.ONLINE
print(status) # "online"
print(str(status)) # "online"
print(status.value) # "online"
# Sammenligning virker med strings:
assert status == "online" # True
assert "online" == GatewayStatus.ONLINE # True
# JSON-serialisering:
import json
json.dumps({"status": status}) # '{"status": "online"}' — ingen .value nødvendig
SQLModel enum-kolonner
# server/src/db/models.py
from sqlalchemy import Enum as SAEnum
import enum
class Gateway(SQLModel, table=True):
"""Gateway med enum-status."""
id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
tenant_id: uuid.UUID
# PostgreSQL native enum-type:
status: GatewayStatus = Field(
default=GatewayStatus.PROVISIONING,
sa_column=Column(
SAEnum(GatewayStatus, name="gateway_status_enum"),
nullable=False,
)
)
# Alternativ: TEXT-kolonne med Python validation:
# status: str = Field(default="provisioning")
def is_active(self) -> bool:
return self.status in {GatewayStatus.ONLINE, GatewayStatus.DEGRADED}
class Settlement(SQLModel, table=True):
status: SettlementStatus = Field(
default=SettlementStatus.DRAFT,
sa_column=Column(
SAEnum(SettlementStatus, name="settlement_status_enum"),
nullable=False,
)
)
Tilstandsmaskine med valid transitions
# server/src/settlements/state_machine.py
from enum import StrEnum
VALID_TRANSITIONS: dict[SettlementStatus, set[SettlementStatus]] = {
SettlementStatus.DRAFT: {SettlementStatus.GENERATING},
SettlementStatus.GENERATING: {
SettlementStatus.GENERATED,
SettlementStatus.DRAFT, # Fejl → tilbage til draft
},
SettlementStatus.GENERATED: {SettlementStatus.SENT},
SettlementStatus.SENT: {
SettlementStatus.VIEWED,
SettlementStatus.DISPUTED,
},
SettlementStatus.VIEWED: {
SettlementStatus.PAID,
SettlementStatus.DISPUTED,
SettlementStatus.CLOSED,
},
SettlementStatus.PAID: {SettlementStatus.CLOSED},
SettlementStatus.DISPUTED: {
SettlementStatus.VIEWED, # Dispute løst
SettlementStatus.CLOSED, # Dispute lukket
},
SettlementStatus.CLOSED: set(), # Terminal state
}
def transition_settlement(
settlement: Settlement,
new_status: SettlementStatus,
actor_id: uuid.UUID,
) -> Settlement:
"""
Valideret tilstandsovergang.
Kaster ValueError ved ugyldig overgang.
"""
current = settlement.status
allowed = VALID_TRANSITIONS.get(current, set())
if new_status not in allowed:
raise ValueError(
f"Ugyldig statusovergang: {current} → {new_status}. "
f"Tilladte: {', '.join(s.value for s in allowed)}"
)
old_status = settlement.status
settlement.status = new_status
settlement.updated_at = datetime.utcnow()
# Audit log:
log_audit(
action="settlement.status_change",
entity_id=settlement.id,
before={"status": old_status},
after={"status": new_status},
actor_id=actor_id,
)
return settlement
Flag enum til gateway-kapabiliteter
# Flag-enum: Bitflag til at repræsentere multiple tilstande
class GatewayCapability(Flag):
"""Gateway-kapabiliteter som bitflag."""
NONE = 0
WMBUS = auto() # wM-Bus modtagelse
CELLULAR = auto() # 4G/SIM7080G
WIFI = auto() # WiFi backup
LOCAL_UI = auto() # FastAPI lokal UI
OTA = auto() # Over-the-air opdatering
CLOUDFLARE = auto() # Cloudflare Tunnel
AES_DECRYPT = auto() # AES-128 CTR nøgler konfigureret
# Kombination:
FULL = WMBUS | CELLULAR | OTA | CLOUDFLARE
# Brug:
gw_caps = GatewayCapability.WMBUS | GatewayCapability.CELLULAR | GatewayCapability.OTA
# Check:
if GatewayCapability.AES_DECRYPT not in gw_caps:
log.warning("gateway.no_aes_keys", gateway_id=gw_id)
# Serialisering til JSON:
def caps_to_list(caps: GatewayCapability) -> list[str]:
return [c.name.lower() for c in GatewayCapability if c in caps and c != GatewayCapability.NONE]
caps_to_list(gw_caps) # ["wmbus", "cellular", "ota"]
Pydantic-integration
# Pydantic håndterer StrEnum transparent:
from pydantic import BaseModel
class GatewayOut(BaseModel):
id: uuid.UUID
status: GatewayStatus
capabilities: list[str]
model_config = {"use_enum_values": True} # Serialisér til string
# JSON output:
# {"id": "...", "status": "online", "capabilities": [...]}
# Request-validering:
class GatewayStatusUpdate(BaseModel):
status: GatewayStatus # Validerer automatisk at string er valid enum-value
# Test:
data = GatewayStatusUpdate(status="invalid") # ValidationError
data = GatewayStatusUpdate(status="online") # OK — GatewayStatus.ONLINE
data = GatewayStatusUpdate(status=GatewayStatus.DEGRADED) # OK
Enum i MQTT-payload
# gateway/src/mqtt/payloads.py
from enum import StrEnum
import msgpack
class CommandType(StrEnum):
"""MQTT-kommandoer fra server til gateway."""
REBOOT = "reboot"
OTA_UPDATE = "ota_update"
SCAN_METERS = "scan_meters"
UPDATE_CONFIG = "update_config"
TUNNEL_OPEN = "tunnel_open"
TUNNEL_CLOSE = "tunnel_close"
class AlarmCode(StrEnum):
METER_SILENT = "meter_silent"
GATEWAY_OFFLINE = "gateway_offline"
LOW_BATTERY = "low_battery"
AES_ERROR = "aes_error"
CLOCK_DRIFT = "clock_drift"
# MessagePack-serialisering (StrEnum → str automatisk):
payload = {
"command": CommandType.OTA_UPDATE,
"version": "1.4.2",
"url": "https://...",
}
packed = msgpack.packb(payload)
# StrEnum serialiseres til string — ingen speciel håndtering
# Server-side deserialisering:
data = msgpack.unpackb(packed)
cmd_type = CommandType(data["command"]) # "ota_update" → CommandType.OTA_UPDATE
Konklusion
StrEnum (Python 3.11+) er det primære valg til IoT-platform: serialiserer transparent til JSON-strings, validerer i Pydantic, fungerer i SQLAlchemy-kolonner og sammenlignes direkte med strings. Tilstandsmaskine-dict med VALID_TRANSITIONS håndhæver lovlige statusovergange ved settlement-generering. Flag-enum til gateway-kapabiliteter muliggør bitflag-baseret repræsentation af multiple samtidige egenskaber. IntEnum bruges kun til protokol-bytekoder (OMS, M-Bus).