optimized control_mqtt

.vscode/settings.json

@@ -1,3 +1,4 @@
 {
-    "taipyStudio.config.mainPythonFile": "prova.py"
+    "taipyStudio.config.mainPythonFile": "prova.py",
+    "taipyStudio.gUI.elementsFilePaths": []
 }

.vscode/tasks.json

@@ -0,0 +1,12 @@
+{
+    // See https://go.microsoft.com/fwlink/?LinkId=733558
+    // for the documentation about the tasks.json format
+    "version": "2.0.0",
+    "tasks": [
+        {
+            "label": "deploy file to mqtt server",
+            "type": "shell",
+            "command": "scp -P 6229 ${file} root@mqtt:/var/lib/mosquitto/"
+        }
+    ]
+}

control_mqtt.py

@@ -0,0 +1,119 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+import paho.mqtt.subscribe as subscribe
+import paho.mqtt.publish as publish
+import subprocess
+import argparse
+import json
+import os
+import logging
+
+class CurrentClients:
+    def __init__(self, args):
+        with open(args.dyn_sec_conf, "r") as file:
+            data = json.load(file)
+        self.active_clients_list = [client["username"] for client in data["clients"]]
+        self.active_clients_pids = {}
+        for username in self.active_clients_list:
+            if username.endswith("_ase"):
+                self.start_client(username, args)
+
+    def list(self):
+        return self.active_clients_list
+
+    def pids(self):
+        return self.active_clients_pids
+
+    def add(self, client, args):
+        self.active_clients_list.append(client)
+        self.start_client(client, args) if client.endswith("_ase") else None
+
+    def remove(self, client):
+        self.active_clients_list.remove(client)
+        self.stop_client(client) if client.endswith("_ase") else None
+
+    def start_client(self, client, args):
+        process = subprocess.Popen(
+            ["python3", args.ase_receiver, client],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True
+        )
+        self.active_clients_pids[client] = process.pid
+        logging.info(f"Started process for {client}, PID: {process.pid}")
+
+    def stop_client(self, client):
+        logging.info(f"Terminating process for {client} (PID: {self.active_clients_pids[client]})")
+        os.kill(self.active_clients_pids[client], 9)
+
+def get_client_list(args, auth):
+    publish.single(args.pub_topic, '{"commands":[{"command":"listClients"}]}', hostname=args.host, port=args.port, auth=auth)
+
+def create_client(datas, userdata):
+    get_client_list(userdata['args'], userdata['auth'])
+
+def delete_client(datas, userdata):
+    get_client_list(userdata['args'], userdata['auth'])
+
+def list_clients(datas, userdata):
+    list_clients = datas['responses'][0]['data']['clients']
+
+    delta_clients_add = set(list_clients) - set(userdata['cur_clients'].list())
+    [userdata['cur_clients'].add(item, userdata['args']) for item in delta_clients_add]
+
+    delta_clients_del = set(userdata['cur_clients'].list()) - set(list_clients)
+    [userdata['cur_clients'].remove(item) for item in delta_clients_del]
+
+def ctrl_client_mod(client, userdata, message):
+
+    command_functions = {
+        "createClient": create_client,
+        "deleteClient": delete_client,
+        "listClients": list_clients,
+    }
+
+    datas = json.loads(message.payload)
+    target_commands = {"createClient", "deleteClient", "listClients"}
+    found_command = [item["command"] for item in datas['responses'] if item["command"] in target_commands]
+    if found_command:
+        command_functions[found_command[0]](datas, userdata)
+
+def main():
+    parser = argparse.ArgumentParser()
+
+    parser.add_argument('-H', '--host', default="mqtt")
+    parser.add_argument('-t', '--sub_topic', default="$CONTROL/dynamic-security/v1/response")
+    parser.add_argument('-T', '--pub_topic', default="$CONTROL/dynamic-security/v1")
+    parser.add_argument('-q', '--qos', type=int,default=0)
+    parser.add_argument('-u', '--username', default="admin")
+    parser.add_argument('-p', '--password', default="BatManu#171017")
+    parser.add_argument('-P', '--port', type=int, default=1883)
+    parser.add_argument('-l', '--log', default='/var/log/mosquitto/subscriber_manager.log')
+    parser.add_argument('-L', '--log_level', choices=['DEBUG', 'INFO', 'WARNING', 'ERROR'], default='INFO')
+    parser.add_argument('-s', '--dyn_sec_conf', default='./dynamic-security.json')
+    parser.add_argument('-r', '--ase_receiver', default='./subscribe_ase_receiver.py')
+
+    args = parser.parse_args()
+
+    logging.basicConfig(
+        format="%(asctime)s - PID: %(process)d %(levelname)8s: %(message)s ",
+        level=args.log_level,
+        filename=args.log
+    )
+
+    cur_clients = CurrentClients(args)
+    auth = {'username': args.username, 'password': args.password}
+    userdata = {'args': args, 'cur_clients': cur_clients, 'auth': auth}
+
+    try:
+        subscribe.callback(ctrl_client_mod, hostname=args.host, port=args.port, topics=args.sub_topic, auth=auth, userdata=userdata)
+    except (KeyboardInterrupt, Exception) as e:
+        logging.info("Terminating: ....")
+
+        for client in cur_clients.list():
+            cur_clients.stop_client(client) if client.endswith("_ase") else None
+
+
+if __name__ == "__main__":
+    main()

pyproject.toml

@@ -4,4 +4,6 @@ version = "0.1.0"
 description = "Add your description here"
 readme = "README.md"
 requires-python = ">=3.12"
-dependencies = []
+dependencies = [
+    "paho-mqtt>=2.1.0",
+]

subscribe_ase_receiver.py

@@ -0,0 +1,19 @@
+import sys
+import time
+import os
+
+# Verifica degli argomenti passati
+if len(sys.argv) != 2:
+    print("Uso: python programma.py <username>")
+    sys.exit(1)
+
+# Recupera il nome del client dall'argomento
+username = sys.argv[1]
+
+# Simula un processo in esecuzione
+print(f"Inizio elaborazione per {username} (PID: {os.getpid()})")
+try:
+    while True:
+        time.sleep(5)  # Simula un lavoro in corso
+except KeyboardInterrupt:
+    print(f"Processo per {username} terminato")

subscribe_callback.py

@@ -1,110 +0,0 @@
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
-
-import paho.mqtt.subscribe as subscribe
-import paho.mqtt.publish as publish
-import subprocess
-import argparse
-import json
-import os
-import logging
-
-class CurrentClients:
-    def __init__(self, args):
-        with open(args.config_json, "r") as file:
-            data = json.load(file)
-        self.active_clients_list = [client["username"] for client in data["clients"]]
-        self.active_clients_pids = {}
-        for username in self.active_clients_list:
-            if username.endswith("_ase"):
-                self.start_client(username, args)
-
-    def list(self):
-        return self.active_clients_list
-
-    def pids(self):
-        return self.active_clients_pids
-
-    def add(self, client, args):
-        self.active_clients_list.append(client)
-        self.start_client(client, args) if client.endswith("_ase") else None
-
-    def remove(self, client):
-        self.active_clients_list.remove(client)
-        self.stop_client(client) if client.endswith("_ase") else None
-
-    def start_client(self, client, args):
-        process = subprocess.Popen(
-            ["python3", args.sub_process, client],
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE,
-            text=True
-        )
-        self.active_clients_pids[client] = process.pid
-        logging.info(f"Started process for {client}, PID: {process.pid}")
-
-    def stop_client(self, client):
-        logging.info(f"Terminating process for {client} (PID: {self.active_clients_pids[client]})")
-        os.kill(self.active_clients_pids[client], 9)
-
-def get_client_list(args):
-    publish.single(args.pub_topic, args.list_clients_cmd, hostname=args.host, port=args.port, auth={'username': args.username, 'password': args.password})
-
-def create_client(datas, args, cur_clients):
-    get_client_list(args)
-
-def delete_client(datas, args, cur_clients):
-    get_client_list(args)
-
-def list_client(datas, args, cur_clients):
-    list_clients = datas['responses'][0]['data']['clients']
-
-    delta_clients_add = set(list_clients) - set(cur_clients.list())
-    [cur_clients.add(item,args) for item in delta_clients_add]
-
-    delta_clients_del = set(cur_clients.list()) - set(list_clients)
-    [cur_clients.remove(item) for item in delta_clients_del]
-
-def ctrl_client_mod(client, userdata, message):
-
-    command_functions = {
-        "createClient": create_client,
-        "deleteClient": delete_client,
-        "listClients": list_client,
-    }
-
-    datas = json.loads(message.payload)
-    target_commands = {"createClient", "deleteClient", "listClients"}
-    found_command = [item["command"] for item in datas['responses'] if item["command"] in target_commands]
-    if found_command:
-        command_functions[found_command[0]](datas, args, userdata)
-
-parser = argparse.ArgumentParser()
-
-parser.add_argument('-H', '--host', required=False, default="mqtt")
-parser.add_argument('-t', '--sub_topic', required=False, default="$CONTROL/dynamic-security/v1/response")
-parser.add_argument('-T', '--pub_topic', required=False, default="$CONTROL/dynamic-security/v1")
-parser.add_argument('-q', '--qos', required=False, type=int,default=0)
-parser.add_argument('-u', '--username', required=False, default="admin")
-parser.add_argument('-p', '--password', required=False, default="BatManu#171017")
-parser.add_argument('-P', '--port', required=False, type=int, default=1883)
-
-args = parser.parse_args()
-args.list_clients_cmd = '{"commands":[{"command":"listClients"}]}'
-args.config_json = './dynamic-security.json'
-args.sub_process = './subscribe_ase_receiver.py'
-
-# Configure logging
-logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
-
-cur_clients = CurrentClients(args)
-
-try:
-    subscribe.callback(ctrl_client_mod, hostname=args.host, port=args.port, topics=args.sub_topic, auth={'username': args.username, 'password': args.password}, userdata=cur_clients)
-except (KeyboardInterrupt, Exception) as e:
-    logging.info("Terminating: ....")
-
-    for client in cur_clients.list():
-        cur_clients.stop_client(client) if client.endswith("_ase") else None
-
-

uv.lock

@@ -0,0 +1,22 @@
+version = 1
+requires-python = ">=3.12"
+
+[[package]]
+name = "ase-mqtt"
+version = "0.1.0"
+source = { virtual = "." }
+dependencies = [
+    { name = "paho-mqtt" },
+]
+
+[package.metadata]
+requires-dist = [{ name = "paho-mqtt", specifier = ">=2.1.0" }]
+
+[[package]]
+name = "paho-mqtt"
+version = "2.1.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/39/15/0a6214e76d4d32e7f663b109cf71fb22561c2be0f701d67f93950cd40542/paho_mqtt-2.1.0.tar.gz", hash = "sha256:12d6e7511d4137555a3f6ea167ae846af2c7357b10bc6fa4f7c3968fc1723834", size = 148848 }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/c4/cb/00451c3cf31790287768bb12c6bec834f5d292eaf3022afc88e14b8afc94/paho_mqtt-2.1.0-py3-none-any.whl", hash = "sha256:6db9ba9b34ed5bc6b6e3812718c7e06e2fd7444540df2455d2c51bd58808feee", size = 67219 },
+]