iotsim 1.0.1

IoT Device Simulator for developing and testing downstream logic relying on IoT physical devices

IoT Device Simulator

Python IoT device simulator consisting of an IoT Container running units representing IoT devices defined in a json file (see iotunits.json), a smart elevator and a temperature sensor implementation can be found in the examples folder.

Usage guide - Run examples

If you don't know what's uv -> astral-uv

Open folder in devcontainer, then you should be good to go

uv sync (optional)

create json config, use config-default.json as template

{
    "logger": {
        "file_path": "./iotsim.log",
        "verbosity": "DEBUG"
    },
    "client": {
        "name": "test_client",
        "type": "mqtt",
        "host": "localhost",
        "port": 1883,
        "root_ca_path": "",
        "client_certificate_path": "",
        "client_key_path": ""
    },
    "units": {
        "units_list_file_path": "", #leaving it blank will load examples
        "units_py_module_path": ""  #leaving it blank will load examples
    }
}

Run simulator

uv run iotsim/main.py --config <path-to-config.json> #if no argument is set the app will make use of the default config-default.json

Docker

docker build -t iotsim:local .
docker run --network=host iotsim:local

or

docker pull ghcr.io/mockasort-studio/iotsim:latest

Implement your own IoT Units

An IoT unit can be anything you wish, the package makes use of importlib to import user defined module at runtime just by declaring them in the iotunits.json.

The Units make use of user defined registers (simulating the volatile memory of the device) to manipulate and share data between control_loop, publishers and subscribers. Register works in a dictionary fashion and the keys and init value must be defined in the json definition of your unit.

{
    "name": "iot-unit", #name of the unit
    "control_loop_module": "user-define-unit.control-loop-module", #null if periodic activity is not needed
    "control_loop_sleep_ms": cyle_time, #cycle time of the periodic activity
    "registers": {}, #register
    "publishers": [{...}], #publishers list
    "subscribers": [{...}] #subscribers list
}

Control loop

Unit package structure

|-- user-defined-unit/
    |-- __init__.py
    |-- control-loop-module.py

Control loop function must have the signature below:

        iotsim.core.stateregistry.StateRegistry
            |
def run(registers,):
    #TODO Implement your logic
    #registers.update('key', some-value)
    #registers.get_value('key')

Publishers

Publishers are mapped to a register key and publish, with a given frequency, the value corresponding to the key.

Json definition

{
    "id": "publisher-id",
    "publish_frequency_ms",       #publish frequency in ms
    "read": "register-key",       #register key accessed by publisher
    "topic": "mqtt-topic"         #mqtt-topic of the subscription
}

Subscribers

Subscribers are triggered by a message received on their mqtt-topic and write some data received on topic to the register value they're mapped to

Json definition

{
    "id": "subscriber-id",
    "write": "register-key",         #register key accessed by subscriber
    "topic": "unit/robot_request"    #mqtt-topic of the subscription
}

TODO LIST

• Code clean up
• Unit testing
• Improve documentation
• Improve overall abstraction
• Improve Python packaging
• Refactor project in order to make use of an arbitrary communication protocol

Contributors

• Michelangelo Setaro

Contribution

If you wish to contribute get in touch with MockaSort-Studio

• [e-mail]
• MockaSort Studio - Discord