MQTT Device Manager & Web UI

Features

• Device interface via MQTT: The Standard for IoT Messaging

  It is designed as an extremely lightweight publish/subscribe messaging transport that is ideal for connecting remote devices with a small code footprint and minimal network bandwidth.

• Device properties are automatically discovered & subscribed from MQTT topic structure
• Commands annotated with JSON Schema
  • Input is validated according to schema with python-jsonschema
  • Schema is passed directly to the frontend for form generation with JSONForms
• Reactive Vue frontend
  • Fast, realtime updates due to pub/sub architecture
  • Property search by name, name fragment or value
  • Flash effect on property update
  • Frontend schema validation for commands with human readable error messages

Example command implementation

    @command(
        {
            "type": "object",
            "properties": {
                "measRate": {
                    "type": "integer",
                    "minimum": 50,
                    "maximum": 60000,
                    "description": "Measurement period in milliseconds",
                    "default": 1000,
                },
                "navRate": {
                    "type": "integer",
                    "minimum": 1,
                    "maximum": 127,
                    "description": "Number of measurement cycles per navigation solution",
                    "default": 1
                },
                "timeRef": {
                    "type": "integer",
                    "enum": [0, 1],
                    "description": "Time reference (0=UTC, 1=GPS)",
                },
            },
            "required": ["measRate", "navRate", "timeRef"],
            "additionalProperties": False,
        },
        description="Reconfigure the rate properties for the UBX device.",
    )
    async def configure_rate(self, args):
        message = pyubx2.UBXMessage(
            "CFG",
            "CFG-RATE",
            pyubx2.ubxtypes_core.SET,
            measRate=args["measRate"],
            navRate=args["navRate"],
            timeRef=args["timeRef"],
        )
        num_bytes = await self.serial_port.write_async(message.serialize())
        return num_bytes