Subscriber MQTT Forge IoT¶

Statut : subscriber livré, branché sur
paho-mqtt. Le module parse les
messages en objets Measurement, les persiste dans iot_events, les
expose via une API HTTP JSON, et fournit la CLI iot:*
(doctor/init/listen/simulate). Cette page décrit la couche subscriber.

Objectif¶

Brancher Forge sur un broker MQTT pour recevoir les messages des
capteurs, strictement selon le contrat figé par
IOT-MQTT-CONTRACT-001, et délivrer chaque mesure
valide à un callback applicatif.

Architecture du module¶

Le code est volontairement scindé en deux modules :

Module	Rôle	Dépend de paho-mqtt ?
`forge_mvc_iot.mqtt.contract`	parsing et validation pure (topic + payload), construction de `Measurement`	non, testable hors ligne
`forge_mvc_iot.mqtt.subscriber`	pont vers `paho.mqtt.client.Client`, callbacks, abonnement, cycle de vie	oui

Cette séparation permet de :

tester l'intégralité du contrat sans installer ou simuler un broker ;
réutiliser parse_message ailleurs (par exemple dans un test d'API
HTTP qui injecte directement une mesure) ;
changer plus tard de bibliothèque MQTT sans toucher au contrat.

API publique¶

`contract.parse_message(topic, payload)`¶

Parse un message complet et retourne une Measurement.

from forge_mvc_iot.mqtt.contract import parse_message

m = parse_message(
    "forge/atelier/esp32-001/telemetry",
    b'{"kind":"temperature","value":22.4,"unit":"°C",'
    b'"timestamp":"2026-05-28T10:00:00Z"}',
)
# Measurement(site='atelier', device_id='esp32-001', kind='temperature',
#             value=22.4, unit='°C', timestamp='2026-05-28T10:00:00Z',
#             metadata=None)

payload accepte str ou bytes.

`contract.Measurement`¶

Dataclass immuable (frozen=True) :

@dataclass(frozen=True)
class Measurement:
    site: str
    device_id: str
    kind: str
    value: int | float
    unit: str
    timestamp: str             # ISO 8601 UTC, suffixe Z
    metadata: dict[str, str] | None

Le timestamp reste en chaîne ISO pour rester proche du payload reçu :
les consommateurs convertissent en datetime quand ils en ont besoin.

`contract.ContractError`¶

Exception levée pour toute violation du contrat. Porte un attribut
code issu de la taxonomie stable :

Constante	Valeur
`CODE_TOPIC_PATTERN`	`"TOPIC_PATTERN"`
`CODE_PAYLOAD_PARSE`	`"PAYLOAD_PARSE"`
`CODE_PAYLOAD_FIELD_MISSING`	`"PAYLOAD_FIELD_MISSING"`
`CODE_PAYLOAD_FIELD_TYPE`	`"PAYLOAD_FIELD_TYPE"`
`CODE_PAYLOAD_VALUE_FORMAT`	`"PAYLOAD_VALUE_FORMAT"`

Ces codes sont identiques à ceux documentés dans
Contrat MQTT : Erreurs.

from forge_mvc_iot.mqtt.contract import ContractError, parse_message

try:
    parse_message("bad", b"{}")
except ContractError as exc:
    print(exc.code)     # 'TOPIC_PATTERN'
    print(exc.message)  # 'Topic invalide : ...'

`subscriber.MqttSubscriber`¶

Pont avec paho-mqtt.

from forge_mvc_iot.config import load_iot_config
from forge_mvc_iot.mqtt.subscriber import MqttSubscriber

cfg = load_iot_config()  # lit os.environ

def on_measurement(m):
    print(f"{m.site}/{m.device_id} → {m.kind}={m.value}{m.unit}")

sub = MqttSubscriber(cfg, on_measurement=on_measurement)
sub.connect()
sub.loop_forever()

Paramètres du constructeur¶

Paramètre	Type	Rôle
`config`	`IotConfig`	configuration chargée par `load_iot_config`
`on_measurement`	`Callable[[Measurement], None]`	obligatoire, appelé pour chaque mesure valide
`on_contract_error`	`Callable[[ContractError, str, bytes], None] \\| None`	optionnel, appelé pour chaque violation de contrat. Si `None`, l'erreur est uniquement loguée.
`client_factory`	`Callable[[IotConfig], Any] \\| None`	optionnel, fabrique de client MQTT, utile pour les tests. Par défaut, instancie un `paho.mqtt.client.Client` configuré pour `CallbackAPIVersion.VERSION2`.

Méthodes de cycle de vie¶

Méthode	Effet
`connect()`	ouvre la connexion vers le broker
`disconnect()`	ferme la connexion
`loop_forever()`	boucle réseau bloquante
`loop_start()`	boucle réseau en thread séparé
`loop_stop()`	arrête la boucle démarrée par `loop_start`
`handle_message(topic, payload)`	point d'entrée testable, parse + dispatch

Exemple complet : script de réception¶

import logging

from forge_mvc_iot.config import load_iot_config
from forge_mvc_iot.mqtt.contract import ContractError, Measurement
from forge_mvc_iot.mqtt.subscriber import MqttSubscriber

logging.basicConfig(level=logging.INFO)
log = logging.getLogger("forge-iot-demo")


def on_measurement(m: Measurement) -> None:
    log.info(
        "%s/%s - %s = %s %s @ %s",
        m.site, m.device_id, m.kind, m.value, m.unit, m.timestamp,
    )


def on_contract_error(exc: ContractError, topic: str, payload: bytes) -> None:
    log.warning("Refusé [%s] sur %r : %s", exc.code, topic, exc.message)


def main() -> None:
    cfg = load_iot_config()
    sub = MqttSubscriber(cfg, on_measurement, on_contract_error)
    sub.connect()
    sub.loop_forever()


if __name__ == "__main__":
    main()

Avec un Mosquitto local qui tourne, on peut publier depuis un autre
terminal :

mosquitto_pub \
  -h localhost \
  -t forge/atelier/esp32-001/telemetry \
  -m '{"kind":"temperature","value":22.4,"unit":"°C","timestamp":"2026-05-28T10:00:00Z"}'

Le script affiche :

INFO forge-iot-demo: atelier/esp32-001 - temperature = 22.4 °C @ 2026-05-28T10:00:00Z

Tester sans broker¶

Tout le contrat est testable sans paho :

from forge_mvc_iot.mqtt.contract import parse_message

m = parse_message("forge/atelier/esp32-001/telemetry", b'{...}')
assert m.kind == "temperature"

Le subscriber lui-même est testable en injectant un client_factory :

from unittest.mock import MagicMock
from forge_mvc_iot.config import IotConfig
from forge_mvc_iot.mqtt.subscriber import MqttSubscriber

cfg = IotConfig(
    mqtt_host="localhost", mqtt_port=1883,
    mqtt_topic="forge/+/+/telemetry",
    mqtt_client_id="t",
    mqtt_username=None, mqtt_password=None,
)
sub = MqttSubscriber(
    cfg,
    on_measurement=lambda m: print(m),
    client_factory=lambda c: MagicMock(),
)
sub.handle_message(
    "forge/atelier/esp32-001/telemetry",
    b'{"kind":"temperature","value":22.4,"unit":"°C","timestamp":"2026-05-28T10:00:00Z"}',
)

C'est le pattern utilisé par la suite de tests
tests/test_iot_mqtt_subscriber_001.py.

Comportement face aux erreurs¶

Pour chaque message reçu, le subscriber tente
parse_message(topic, payload) :

Succès → on_measurement(Measurement) est appelé.
ContractError :
un message WARNING est logué sur forge_mvc_iot.mqtt.subscriber
avec le code d'erreur et le topic en cause ;
si on_contract_error est fourni, il est appelé avec
(exc, topic, payload) ;
le subscriber ne se déconnecte pas : un message invalide ne
casse pas la session MQTT.

Les erreurs de connexion broker (broker injoignable, identifiants
refusés) restent gérées par paho-mqtt. La reconnexion automatique se
fait via loop_forever() / loop_start(). Un raffinement (backoff,
politique de reconnexion personnalisée) sera abordé dans
IOT-DOCTOR-001.

Décisions verrouillées¶

site et device_id viennent du topic. Tout champ site ou
device_id présent dans le payload est ignoré silencieusement
(couvert par les tests).
Codes d'erreur stables. La taxonomie est figée : un futur test
du subscriber peut s'appuyer sur exc.code == "PAYLOAD_PARSE" sans
craindre un renommage.
Pas d'imports paho dans contract.py. La séparation est testée :
parse_message reste utilisable même si paho-mqtt n'est pas installé.

Hors périmètre du module¶

Le module couvre l'ingestion (subscriber), le stockage iot_events, l'API HTTP
JSON et la CLI iot:*. Restent hors périmètre :

pas de dashboard, pas de Forge Design ;
pas d'ACL Mosquitto gérée par Forge (côté broker) ;
pas de downlink Forge → capteur (commande descendante).

Voir Architecture Forge IoT pour la
liste complète des jalons.