Extensible MQTT IoT Communication Framework

A lightweight Python framework for building MQTT-based IoT systems. It provides reusable base classes for hubs, devices, sensors, and MQTT clients, making it easier to implement custom IoT workflows with a consistent architecture.

• PyPI: foresttm23-iot
• Requires: Python 3.13+

Overview

This framework is designed for object-oriented IoT development and is centered around reusable base abstractions for:

• BaseMqttClient — base class for MQTT clients
• BaseIot — shared base class for sensors and devices
• BaseDevice — base class for IoT devices
• BaseSensor — base class for IoT sensors
• BaseHub — base class for IoT hubs

It also includes a registry system that allows IoT classes to be dynamically discovered and instantiated by their sub_type.

Key Features

• MQTT-based IoT architecture
• Reusable base classes for hubs, devices, and sensors
• Class registry for dynamic IoT type lookup
• Support for custom and extended IoT implementations
• Built-in logging and helper utilities
• Designed to work with an MQTT broker

Core Concepts

Registry System

Use the @register_iot decorator to register your IoT classes:

from iot_framework import BaseDevice, register_iot


@register_iot
class LampDevice(BaseDevice):
    sub_type = "lamp"

This registers the class in the framework’s internal registry so it can be instantiated dynamically with get_iot_class().

Base Classes

• BaseMqttClient: Provides the shared MQTT client behavior.
• BaseIot: Adds shared IoT functionality for sensors and devices.
• BaseDevice / BaseSensor: Specialized implementations for device and sensor behavior.
• BaseHub: Coordinates devices and sensors, handles discovery, and manages device interactions.

Installation

Install the package from PyPI:

pip install foresttm23-iot

Or install dependencies for local development using the project’s environment setup.

Usage

Run the framework entry point

python main.py <hub_sub_type> <device_sub_type> <sensor_sub_type>

Example:

python main.py hub lamp motion

This launches the requested IoT processes and connects them to the configured MQTT broker.

Example Device

import json
import time

from iot_framework import BaseDevice, logger, register_iot


@register_iot
class LampDevice(BaseDevice):
    sub_type = "lamp"

    def __init__(self, client, device_id: str):
        super().__init__(client, device_id, self.sub_type)

    def on_message(self, client, userdata, msg) -> None:
        payload = json.loads(msg.payload.decode())
        command = payload.get("command").lower()
        transaction_id = payload.get("transaction_id")

        logger.info(f"📩 [{self.id}] Received command: {command}")
        self.set_state(command)

        lamp_payload = json.dumps(
            {
                "state": self.state,
                "transaction_id": transaction_id,
                "timestamp": time.time(),
            }
        )
        self.client.publish(self.state_topic, lamp_payload, retain=True)

Requirements

• Python 3.13+
• MQTT broker
• loguru
• paho-mqtt

Project Structure

• iot_framework/ — framework package
• main.py — launcher for IoT processes
• docker-compose.yml — container setup
• Dockerfile — image definition
• load_test.jmx — performance/load testing
• mosquitto.conf — MQTT broker configuration

License

No license has been specified for this project.