Mit default zusammenführen

revpipyload/mqttserver.py

@@ -0,0 +1,199 @@
+# -*- coding: utf-8 -*-
+#
+# RevPiPyLoad
+#
+# Webpage: https://revpimodio.org/revpipyplc/
+# (c) Sven Sager, License: LGPLv3
+#
+"""Stellt die MQTT Uebertragung fuer IoT-Zwecke bereit."""
+import proginit
+from json import load as jload
+from ssl import CERT_NONE
+from paho.mqtt.client import Client
+from threading import Thread, Event
+
+
+class MqttServer(Thread):
+
+    """Server fuer die Uebertragung des Prozessabbilds per MQTT."""
+
+    def __init__(
+            self, basetopic, sendinterval, host, port=1883,
+            tls_set=False, username="", password=None, client_id=""):
+        """Init MqttServer class.
+
+        @param basetopic Basis-Topic fuer Datenaustausch
+        @param sendinterval Prozessabbild alle n Sekunden senden
+        @param host Adresse <class 'str'> des MQTT-Servers
+        @param port Portnummer <class 'int'> des MQTT-Servers
+        @param keepalive MQTT Ping bei leerlauf
+        @param tls_set TLS fuer Verbindung zum MQTT-Server verwenden
+        @param username Optional Benutzername fuer MQTT-Server
+        @param password Optional Password fuer MQTT-Server
+        @param client_id MQTT ClientID, wenn leer automatisch random erzeugung
+
+        """
+        # TODO: Parameterprüfung
+
+        super().__init__()
+
+        # Klassenvariablen
+        self.__exit = False
+        self._evt_data = Event()
+        self._host = host
+        self._procimglength = self._get_procimglength()
+        self._port = port
+        self._sendinterval = sendinterval
+
+        # Topics konfigurieren
+        self._mqtt_picontrol = "{}/picontrol".format(basetopic)
+        self._mqtt_pictory = "{}/pictory".format(basetopic)
+        self._mqtt_sendpictory = "{}/needpictory".format(basetopic)
+
+        self._mq = Client(client_id)
+        if username != "":
+            self._mq.username_pw_set(username, password)
+        if tls_set:
+            self._mq.tls_set(cert_reqs=CERT_NONE)
+            self._mq.tls_insecure_set(True)
+
+        # Handler konfigurieren
+        self._mq.on_connect = self._on_connect
+        self._mq.on_message = self._on_message
+
+    def _get_procimglength(self):
+        """Ermittelt aus piCtory Konfiguration die laenge.
+        @return Laenge des Prozessabbilds <class 'int'>"""
+        try:
+            with open(proginit.pargs.configrsc, "r") as fh:
+                rsc = jload(fh)
+        except:
+            return 4096
+
+        length = 0
+
+        # piCtory Config prüfen
+        if "Devices" not in rsc:
+            return 0
+
+        # Letzes piCtory Device laden
+        last_dev = rsc["Devices"].pop()
+        length += last_dev["offset"]
+
+        # bei mem beginnen, weil nur der höchste IO benötigt wird
+        for type_iom in ["mem", "out", "inp"]:
+            lst_iom = sorted(
+                last_dev[type_iom],
+                key=lambda x: int(x),
+                reverse=True
+            )
+
+            if len(lst_iom) > 0:
+                # Daten des letzen IOM auswerten
+                last_iom = last_dev[type_iom][str(lst_iom[0])]
+                bitlength = int(last_iom[2])
+                length += int(last_iom[3])
+                length += 1 if bitlength == 1 else int(bitlength / 8)
+                break
+
+        return length
+
+    def _on_connect(self, client, userdata, flags, rc):
+        """Verbindung zu MQTT Broker."""
+        if rc > 0:
+            proginit.warning("can not connect to mqtt broker - will retry")
+        else:
+            # piCtory übertragen um alle RevPiMqttIO zu benachrichtigen
+            self._send_pictory_conf()
+
+            # Subscribe piCtory Anforderung
+            client.subscribe(self._mqtt_sendpictory)
+
+    def _on_disconnect(self, client, userdata, rc):
+        """Wertet Verbindungsabbruch aus."""
+        if rc != 0:
+            proginit.warning(
+                "unexpected disconnection from mqtt broker - "
+                "will try to reconnect"
+            )
+
+    def _on_message(self, client, userdata, msg):
+        """Sendet piCtory Konfiguration."""
+
+        # piCtory Konfiguration senden
+        self._send_pictory_conf()
+
+        # Prozessabbild senden
+        self._evt_data.set()
+
+    def _send_pictory_conf(self):
+        """Sendet piCtory Konfiguration."""
+        with open(proginit.pargs.configrsc, "rb") as fh:
+            self._mq.publish(self._mqtt_pictory, fh.read())
+
+    def newlogfile(self):
+        """Konfiguriert die FileHandler auf neue Logdatei."""
+        pass
+
+    def run(self):
+        """Startet die Uebertragung per MQTT."""
+        proginit.logger.debug("enter MqttServer.start()")
+
+        # Prozessabbild öffnen
+        try:
+            fh_proc = open(proginit.pargs.procimg, "r+b", 0)
+        except:
+            fh_proc = None
+            self.__exit = True
+            proginit.logger.error(
+                "can not open process image {}".format(proginit.pargs.procimg)
+            )
+
+        # MQTT verbinden
+        self._mq.connect_async(self._host, self._port, keepalive=60)
+        self._mq.loop_start()
+
+        # mainloop
+        buff = b''
+        err_count = 0
+        while not self.__exit:
+            self._evt_data.clear()
+
+            # Prozessabbild lesen
+            try:
+                fh_proc.seek(0)
+                buff = fh_proc.read(self._procimglength)
+                if err_count > 0:
+                    proginit.warning(
+                        "resume mqtt publishing after {} errors on "
+                        "processimage".format(err_count)
+                    )
+                err_count = 0
+            except IOError:
+                if err_count == 0:
+                    proginit.logger.error(
+                        "could not read process image for mqtt publishing"
+                    )
+                err_count += 1
+            else:
+                # Prozessabbild übertragen
+                self._mq.publish(self._mqtt_picontrol, buff)
+
+            self._evt_data.wait(self._sendinterval)
+
+        # MQTT trennen
+        self._mq.loop_stop()
+        self._mq.disconnect()
+
+        # FileHandler schließen
+        if fh_proc is not None:
+            fh_proc.close()
+
+        proginit.logger.debug("leave MqttServer.start()")
+
+    def stop(self):
+        """Stoppt die Uebertragung per MQTT."""
+        proginit.logger.debug("enter MqttServer.stop()")
+        self.__exit = True
+        self._evt_data.set()
+        proginit.logger.debug("leave MqttServer.stop()")

revpipyload/plcsystem.py

@@ -221,7 +221,12 @@ class RevPiPlc(Thread):
         # Prozess beenden
         count = 0
         proginit.logger.info("term plc program {0}".format(self._program))
-        self._procplc.terminate()
+        try:
+            self._procplc.terminate()
+        except ProcessLookupError:
+            proginit.logger.error("plc program was terminated unexpectedly")
+            proginit.logger.debug("leave RevPiPlc.stop()")
+            return
 
         while self._procplc.poll() is None and count < 10:
             count += 1

revpipyload/revpipyload.py

@@ -78,6 +78,7 @@ class RevPiPyLoad():
         self.xmlrpcacl = IpAclManager(minlevel=0, maxlevel=4)
 
         # Threads/Prozesse
+        self.th_mqtt = None
         self.th_plcslave = None
         self.plc = None
 
@@ -144,6 +145,7 @@ class RevPiPyLoad():
         proginit.logger.debug("enter RevPiPyLoad._loadconfig()")
 
         # Subsysteme herunterfahren
+        self.stop_plcmqtt()
         self.stop_xmlrpcserver()
 
         # Konfigurationsdatei laden
@@ -182,6 +184,28 @@ class RevPiPyLoad():
         self.zeroonexit = \
             self.globalconfig["DEFAULT"].getboolean("zeroonexit", True)
 
+        # Konfiguration verarbeiten [MQTT]
+        self.mqtt = 0
+        if "MQTT" in self.globalconfig:
+            self.mqtt = \
+                int(self.globalconfig["MQTT"].get("mqtt", 0))
+            self.mqttbasetopic = \
+                self.globalconfig["MQTT"].get("basetopic", "")
+            self.mqttsendinterval = \
+                int(self.globalconfig["MQTT"].get("sendinterval", 15))
+            self.mqtthost = \
+                self.globalconfig["MQTT"].get("host", "")
+            self.mqttport = \
+                int(self.globalconfig["MQTT"].get("port", 1883))
+            self.mqtttls_set = \
+                int(self.globalconfig["MQTT"].get("tls_set", 0))
+            self.mqttusername = \
+                self.globalconfig["MQTT"].get("username", "")
+            self.mqttpassword = \
+                self.globalconfig["MQTT"].get("password", "")
+            self.mqttclient_id = \
+                self.globalconfig["MQTT"].get("client_id", "")
+
         # Konfiguration verarbeiten [PLCSLAVE]
         self.plcslave = False
         if "PLCSLAVE" in self.globalconfig:
@@ -238,6 +262,12 @@ class RevPiPyLoad():
             )
         os.chdir(self.plcworkdir)
 
+        # MQTT konfigurieren
+        self.th_mqtt = self._plcmqtt()
+        if self.th_mqtt is not None and not self._exit:
+            proginit.logger.info("start mqtt publisher")
+            self.th_mqtt.start()
+
         # PLC Programm konfigurieren
         if restart_plcprogram:
             self.stop_plcprogram()
@@ -303,6 +333,8 @@ class RevPiPyLoad():
                 0, self.xml_plcstop, "plcstop")
             self.xsrv.register_function(
                 0, self.xml_reload, "reload")
+            self.xsrv.register_function(
+                0, self.xml_mqttrunning, "mqttrunning")
             self.xsrv.register_function(
                 0, self.xml_plcslaverunning, "plcslaverunning")
 
@@ -343,6 +375,10 @@ class RevPiPyLoad():
                 lambda: os.system(proginit.picontrolreset),
                 "resetpicontrol"
             )
+            self.xsrv.register_function(
+                3, self.xml_mqttstart, "mqttstart")
+            self.xsrv.register_function(
+                3, self.xml_mqttstop, "mqttstop")
             self.xsrv.register_function(
                 3, self.xml_plcslavestart, "plcslavestart")
             self.xsrv.register_function(
@@ -366,6 +402,32 @@ class RevPiPyLoad():
 
         proginit.logger.debug("leave RevPiPyLoad._loadconfig()")
 
+    def _plcmqtt(self):
+        """Konfiguriert den MQTT-Thread fuer die Ausfuehrung.
+        @return MQTT-Thread Object or None"""
+        proginit.logger.debug("enter RevPiPyLoad._plcmqtt()")
+
+        th_plc = None
+        if self.mqtt:
+            try:
+                from mqttserver import MqttServer
+                th_plc = MqttServer(
+                    self.mqttbasetopic,
+                    self.mqttsendinterval,
+                    self.mqtthost,
+                    self.mqttport,
+                    self.mqtttls_set,
+                    self.mqttusername,
+                    self.mqttpassword,
+                    self.mqttclient_id
+                )
+            except:
+                # TODO: Fehlermeldung ausgeben bezüglich paho.mqtt
+                pass
+
+        proginit.logger.debug("leave RevPiPyLoad._plcmqtt()")
+        return th_plc
+
     def _plcthread(self):
         """Konfiguriert den PLC-Thread fuer die Ausfuehrung.
         @return PLC-Thread Object or None"""
@@ -501,8 +563,12 @@ class RevPiPyLoad():
             proginit.logger.info("start xmlrpc-server")
             self.xsrv.start()
 
-        if self.plcslave:
-            # Slaveausfuehrung übergeben
+        # MQTT Uebertragung starten
+        if self.th_mqtt is not None:
+            self.th_mqtt.start()
+
+        # Slaveausfuehrung übergeben
+        if self.th_plcslave is not None:
             self.th_plcslave.start()
 
         # PLC Programm automatisch starten
@@ -517,6 +583,8 @@ class RevPiPyLoad():
                 proginit.logger.info("got reqeust to reload config")
                 self._loadconfig()
 
+            # TODO: MQTT prüfen und neu starten
+
             # PLC Server Thread prüfen
             if self.plcslave and self.th_plcslave is not None \
                     and not self.th_plcslave.is_alive():
@@ -541,8 +609,9 @@ class RevPiPyLoad():
         proginit.logger.info("stopping revpipyload")
 
         # Alle Sub-Systeme beenden
-        self.stop_plcslave()
         self.stop_plcprogram()
+        self.stop_plcmqtt()
+        self.stop_plcslave()
         self.stop_xmlrpcserver()
 
         # Logreader schließen
@@ -556,6 +625,18 @@ class RevPiPyLoad():
         self._exit = True
         proginit.logger.debug("leave RevPiPyLoad.stop()")
 
+    def stop_plcmqtt(self):
+        """Beendet MQTT Sender."""
+        proginit.logger.debug("enter RevPiPyLoad.stop_plcmqtt()")
+
+        if self.th_mqtt is not None and self.th_mqtt.is_alive():
+            proginit.logger.info("stopping mqtt thread")
+            self.th_mqtt.stop()
+            self.th_mqtt.join()
+            proginit.logger.debug("mqtt thread successfully closed")
+
+        proginit.logger.debug("leave RevPiPyLoad.stop_plcmqtt()")
+
     def stop_plcprogram(self):
         """Beendet PLC Programm."""
         proginit.logger.debug("enter RevPiPyLoad.stop_plcprogram()")
@@ -610,6 +691,17 @@ class RevPiPyLoad():
         dc["zeroonerror"] = self.zeroonerror
         dc["zeroonexit"] = self.zeroonexit
 
+        # MQTT Sektion
+        dc["mqtt"] = self.mqtt
+        dc["basetopic"] = self.mqttbasetopic
+        dc["sendinterval"] = self.mqttsendinterval
+        dc["host"] = self.mqtthost
+        dc["port"] = self.mqttport
+        dc["tls_set"] = self.mqtttls_set
+        dc["username"] = self.mqttusername
+        dc["password"] = self.mqttpassword
+        dc["client_id"] = self.mqttclient_id
+
         # PLCSLAVE Sektion
         dc["plcslave"] = self.plcslave
         dc["plcslaveacl"] = self.plcslaveacl.acl
@@ -650,6 +742,42 @@ class RevPiPyLoad():
             buff = fh.read()
         return Binary(buff)
 
+    def xml_mqttrunning(self):
+        """Prueft ob MQTT Uebertragung noch lauft.
+        @return True, wenn MQTT Uebertragung noch lauft"""
+        proginit.logger.debug("xmlrpc call mqttrunning")
+        return False if self.th_mqtt is None \
+            else self.th_mqtt.is_alive()
+
+    def xml_mqttstart(self):
+        """Startet die MQTT Uebertragung.
+
+        @return Statuscode:
+            0: erfolgreich gestartet
+            -1: Nicht aktiv in Konfiguration
+            -2: Laeuft bereits
+
+        """
+        if self.th_mqtt is not None and self.th_mqtt.is_alive():
+            return -2
+        else:
+            self.th_mqtt = self._plcmqtt()
+            if self.th_mqtt is None:
+                return -1
+            else:
+                self.th_mqtt.start()
+                return 0
+
+    def xml_mqttstop(self):
+        """Stoppt  die MQTT Uebertragung.
+        @return True, wenn stop erfolgreich"""
+        if self.th_mqtt is not None:
+            self.stop_plcmqtt()
+            self.th_mqtt = None
+            return True
+        else:
+            return False
+
     def xml_plcdownload(self, mode="tar", pictory=False):
         """Uebertraegt ein Archiv vom plcworkdir.
 
@@ -796,6 +924,17 @@ class RevPiPyLoad():
                 "zeroonerror": "[01]",
                 "zeroonexit": "[01]",
             },
+            "MQTT": {
+                "mqtt": "[01]",
+                "mqttbasetopic": ".*",
+                "mqttsendinterval": "[0-9]+",
+                "mqtthost": ".+",
+                "mqttport": "[0-9]+",
+                "mqtttls_set": "[01]",
+                "mqttusername": ".*",
+                "mqttpassword": ".*",
+                "mqttclient_id": ".+",
+            },
             "PLCSLAVE": {
                 "plcslave": "[01]",
                 "plcslaveacl": self.plcslaveacl.regex_acl,