M-Bus Gateway
← Tilbage til blog
· Python· Enum· StrEnum· IoT· state machine· Pydantic· SQLModel· backend

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).

Se Python type hints guide eller Pydantic v2 guide.