rtl.py

import sys, logging, psutil, subprocess, json, time, platform
from PyQt5.QtCore import QThread
from PyQt5.QtCore import pyqtSignal as Signal
from datetime import datetime, timezone

from definitions import Definition

logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
logger = logging.getLogger()

class RTL433(QThread):
    signalUpdateGUI = Signal(dict)
    signalUpdateInfo = Signal(str)
    signalUpdateDevice = Signal(dict)

    def __init__(self):
        super().__init__()

    def setup(self):
        logger.debug("checking if rtl_433 is already running")
        for proc in psutil.process_iter():
            try:
                # Get process name & pid & cmd line from process object.
                processName = proc.name()
                processID = proc.pid
                if 'rtl_433' in processName:
                    logger.debug("rtl_433 is already running")
                    logger.debug(proc.cmdline())
                    logger.debug(processID)
                    logger.debug("stopping rtl_433")
                    proc.kill()
                    break
            except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
                pass
        logger.debug("starting rtl_433 executable")
        operating_system = platform.system()
        if operating_system == 'Windows':
            self.process = subprocess.Popen(['rtl_433', '-C', 'si', '-M', 'time:iso', '-M', 'utc', '-F', 'json'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        else:
            self.process = subprocess.Popen(['/usr/local/bin/rtl_433', '-f', '868M', '-C', 'si', '-M', 'time:iso', '-M', 'utc', '-F', 'json'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        #self.process = subprocess.Popen(['cat', 'data.txt'], stdout=subprocess.PIPE, stderr=None)
        #return True

        logger.debug("rtl_433 executable started")
        deviceFound = False
        if self.process.stderr:
            for data in self.process.stderr:
                logger.debug(data)
                info = data.decode('utf-8')
                if "Found" in info or "found" in info:
                    deviceFound = True
                    self.signalUpdateInfo.emit(f'<b>{info}</b>')
                    break

        if not deviceFound:
            logger.error("no RTL device found!")
            self.signalUpdateInfo.emit('<b style="color:red;">no RTL device found!</b>')
            return False
        else:
            return True

    def run(self):
        logger.debug("RTL433 run()")
        if not self.setup():
            logger.debug("RTL433 exiting")
            return

        logger.debug("RTL433 processing...")
        timestamp = '';
        for data in self.process.stdout:
            logger.debug("reading data...")

            # process RTL data for display update
            try:
                readings = {}
                logger.debug(f"raw data: {data}")
                data = json.loads(data.rstrip())
                logger.debug(f"JSON formatted data: {data}")

                if not ('model' in data and 'id' in data):
                    logger.debug(f"not processing data: {data}")
                    continue

                if ('time' in data and data['time'] == timestamp):
                    logger.info("discarding duplicate reading reported by the RTL app")
                    continue

                timestamp = data['time']

                #if not ('Cotech-367959' == data['model'] and 93 == data['id']):
                    #logger.debug(f"not processing data from sensor {data['model']}, {data['id']}")
                    #continue

                if 'temperature_C' in data:
                    readings[Definition.TEMP] = data['temperature_C']
                if 'humidity' in data:
                    readings[Definition.HUMIDITY] = data['humidity']
                if 'rain_mm' in data:
                    readings[Definition.RAIN] = data['rain_mm']
                if 'light_lux' in data:
                    readings[Definition.LIGHT] = data['light_lux']
                if 'uv' in data:
                    readings[Definition.UV] = data['uv']
                if 'wind_dir_deg' in data:
                    readings[Definition.WIND_DIR] = data['wind_dir_deg']
                if 'wind_avg_m_s' in data and 'wind_max_m_s' in data:
                    readings[Definition.WIND_AVG] = data['wind_avg_m_s']
                    readings[Definition.WIND_MAX] = data['wind_max_m_s']


                readings[Definition.TIME] = data['time'] if Definition.TIME in data else datetime.now(timezone.utc).replace(microsecond=0).isoformat()
                readings[Definition.MODEL] = data['model'] if Definition.MODEL in data else ''
                readings[Definition.ID] = str(data['id']) if Definition.ID in data else ''
                readings[Definition.TYPE] = data['type'] if Definition.TYPE in data else ''
                readings[Definition.SUBTYPE] = str(data['subtype']) if Definition.SUBTYPE in data else ''
                readings[Definition.CHANNEL] = str(data['channel']) if Definition.CHANNEL in data else ''

                self.signalUpdateDevice.emit({
                    Definition.MODEL: readings[Definition.MODEL],
                    Definition.ID: readings[Definition.ID],
                    Definition.TYPE: readings[Definition.TYPE],
                    Definition.SUBTYPE: readings[Definition.SUBTYPE],
                    Definition.CHANNEL: readings[Definition.CHANNEL]
                })

                logger.debug("emitting updateReadings")
                self.signalUpdateGUI.emit(readings)

            except Exception as e:
                logger.error(f"Processing error: {e}")
                continue



            #time.sleep(10)