feat(publishers.mqtt): connecting and publishing test to mqtt successful

Author: KC3ZVD

src/kc3zvd/iot_state/cli/__init__.py

@@ -7,27 +7,41 @@
 
 @click.group(context_settings={"help_option_names": ["-h", "--help"], "auto_envvar_prefix": "IOT"}, invoke_without_command=False)
 @click.version_option(version=__version__, prog_name="iot-state")
-@click.option('--redis-host', help="Redis instance host", default='localhost')
-@click.option('--redis-port', help="Redis instance port", default=6479, type=int)
+@click.option('--redis-username', help="Redis instance username", default='', type=str)
+@click.option('--redis-password', help="Redis instance password", default='', type=str)
+@click.option('--redis-host', help="Redis instance host", default='localhost', type=str)
+@click.option('--redis-port', help="Redis instance port", default=6379, type=int)
 @click.option('--redis-db', help="Redis instance DB number", default=0, type=int)
 @click.pass_context
-def iot_state(ctx, redis_host, redis_port, redis_db):
-  ctx.ensure_object(dict)
-  ctx.obj['redis_host'] = redis_host
-  ctx.obj['redis_port'] = redis_port
-  ctx.obj['redis_db'] = redis_db
-  click.echo("Starting IOT state platform")
+def iot_state(ctx, redis_username, redis_password, redis_host, redis_port, redis_db):
+    ctx.ensure_object(dict)
+    
+    if redis_username or redis_password:
+        if not redis_username or not redis_password:
+            click.echo("Provide both username and password for redis")
+            die()
+        ctx.obj['redis_url'] = f"redis://{redis_username}:{redis_password}@{redis_host}:{redis_port}/{redis_db}"
+    else:
+        ctx.obj['redis_url'] = f"redis://{redis_host}:{redis_port}/{redis_db}"
+    click.echo("Starting IOT state platform")
+
 @iot_state.command()
 @click.option('--platform', help="The platform to publish to", required=True, 
               type=click.Choice(['mqtt'], case_sensitive=False))
+@click.option('--mqtt-host', help="The MQTT host to connect to", default='localhost', type=str)
+@click.option('--mqtt-port', help="The port to use to connect to the MQTT host", default=1883, type=int)
 @click.pass_context
-def publisher(ctx, platform):
-  match platform:
-    case 'mqtt':
-      click.echo("mqtt platform selected")
-      mqtt.run(
-        redis_host=ctx.obj['redis_host'],
-        redis_port=ctx.obj['redis_port'],
-        redis_db=ctx.obj['redis_db'])
-    case _:
-      die()
+def publisher(ctx, platform, mqtt_host, mqtt_port):
+    mqtt_url = ""
+    match platform:
+        case 'mqtt':
+            click.echo("mqtt platform selected")
+            mqtt.run(
+                redis_url=ctx.obj['redis_url'],
+                publisher={
+                    "mqtt_host": mqtt_host,
+                    "mqtt_port": mqtt_port
+                }
+            )
+        case _:
+            die()

src/kc3zvd/iot_state/publishers/mqtt.py

@@ -1,15 +1,37 @@
 from __future__ import annotations
 from kc3zvd.iot_state.subscribers import redis
+import paho.mqtt.client as mqtt
 import asyncio
+import json
+
 # prefix/device_type/area_name/device_name/state_class
+class MQTTPublisher:
+
+    def __init__(self, host, port):
+        self.host = host
+        self.port = port
 
-async def reader(channel: redis.client.PubSub):
+async def handle_messages(channel: redis.client.PubSub, publisher: str):
+    publisher = json.loads(publisher)
+
+    mqttc = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
+    mqttc.connect(host=publisher['mqtt_host'], port=publisher['mqtt_port'])
+    mqttc.loop_start()
     while True:
         message = await channel.get_message(ignore_subscribe_messages=True)
         if message is not None:
             print(f"(Reader) Message Received: {message}")
+            mqttc.publish(topic="test/message", payload=message['data']).wait_for_publish()
+    mqttc.disconnect()
+    mqttc.loop_stop()
+
+async def subscribe(redis_url: str, publisher: str):
+    update = asyncio.create_task(redis.subscribe('device:state:update', handle_messages, redis_url, publisher))
+    create = asyncio.create_task(redis.subscribe('device:state:create', handle_messages, redis_url, publisher))
+    await update
+    await create
 
-def run(redis_host: str = 'localhost', redis_port: int = 6379, redis_db: int = 0):
-    asyncio.run(redis.subscribe('publish', reader, redis_host, redis_port, redis_db))
+def run(redis_url: str, publisher: dict):
+    asyncio.run(subscribe(redis_url, json.dumps(publisher)))
 
 

src/kc3zvd/iot_state/subscribers/redis.py

@@ -1,14 +1,12 @@
 from __future__ import annotations
 import redis.asyncio as redis
 import asyncio
-STOPWORD = "STOP"
 
+async def subscribe(channel: str, callback: Callable[[], None], redis_url: str, publisher: str):
+    client = redis.Redis.from_url(redis_url)
 
-async def subscribe(channel: str, callback: Callable[[], None], redis_host: int, redis_port: int, redis_db: int):
     print(f"Subscribing to channel: {channel}")
-    client = redis.Redis()
-
     async with client.pubsub() as pubsub:
-        await pubsub.subscribe(channel)
-        future = asyncio.create_task(callback(pubsub))
+        await pubsub.subscribe(channel, publisher)
+        future = asyncio.create_task(callback(pubsub, publisher))
         await future