#!/usr/bin/env python3 """ """ import paho.mqtt.client as mqtt import PyDSlog.stream as stream import time from datetime import datetime import json import sys import logging import logging.handlers import signal CONFIG_TMP = None CONFIG = None BACKEND = None is_streaming = False is_connected = False is_error = "" flag_reconfigure = False flag_start = False flag_stop = False VERSION = '1.0.0' TOPIC_DATA = 'pydslogd/0/v' TOPIC_ERROR = 'pydslogd/0/E' TOPIC_CONFIG = 'pydslogd/0/C' TOPIC_STATUS = 'pydslogd/0/S' logger = logging.getLogger('') Grun = 5 def signal_handler(signal, frame): global Grun logger.debug("Caught signal, shutting down...") Grun = 0 def set_error(msg=None): global is_error if msg == None: is_error = "" else: is_error = str(msg) def on_subscribe(mqttc, obj, mid, granted_qos): logger.debug("MQTT: subscribed topic '%s'",TOPIC_CONFIG) def on_unsubscribe(mqttc, obj, mid): logger.debug("MQTT: unsubscribed topic '%s'",TOPIC_CONFIG) def on_connect(mqttc, obj, flags, rc): global is_connected if rc == 0: logger.debug("MQTT: connection successful") mqttc.subscribe(TOPIC_CONFIG) mqttc.publish(TOPIC_STATUS, "1", retain=True) is_connected = True elif rc == 1: logger.error("MQTT: connection refused - incorrect protocol version") elif rc == 2: logger.error("MQTT: connection refused - invalid client identifier") elif rc == 3: logger.error("MQTT: connection refused - server unavailable") elif rc == 4: logger.error("MQTT: connection refused - bad username or password") elif rc == 5: logger.error("MQTT: connection refused - not authorised") else: logger.error("MQTT: connection failed ({0})".format(rc)) def on_disconnect(mqttc, obj, rc): global is_connected logger.debug("MQTT: disconnect ({0})".format(rc)) is_connected = False def on_message(mqttc, obj, msg): global CONFIG_TMP global flag_reconfigure global flag_start global flag_stop logger.debug("MQTT: message topic {0} {1}".format(msg.topic, msg.payload)) v = json.loads(msg.payload.decode("utf-8")) #print(v) if not v: set_error('json syntax error') return if "conf" in v: config = parse_config(v["conf"]) if config != None and config != CONFIG: CONFIG_TMP = config flag_reconfigure = True logger.debug("PDSL: got configuration") if "job" in v: j = v["job"] #print("got new job", j) if j == "start": flag_start = True logger.debug("PDSL: got command start") if j == "stop": flag_stop = True logger.debug("PDSL: got command stop") def parse_config(c): logger.debug("PDSL: parse configuration") if type(c) is not dict: set_error("attribute 'conf' is empty") return None for i in ["in","out"]: if i not in c: set_error("missing attribute 'conf."+i+"'") return None c_in = c["in"] c_out = c["out"] for i in ["device","channels","freq","frameSize"]: if i not in c_in: set_error("missing attribute 'conf.in."+i+"'") return None if "data" not in c_out: set_error("missing attribute 'conf.out.data'") return None if c_in["device"] not in {"mls160a", "io5640"}: set_error("unsupported device '"+c_in["device"]+"'") return None # TODO test chennels freq size return c def pdsl_restart(): global BACKEND global CONFIG global FFT global flag_start global flag_stop global flag_reconfigure global is_streaming logger.debug("PDSL: reconfigure") port="/dev/ttyR1" baudrate=115200 CONFIG = CONFIG_TMP if BACKEND != None and is_streaming and (flag_reconfigure or flag_stop): BACKEND.stop() BACKEND.disconnect() is_streaming = False mqttc.publish(TOPIC_STATUS, "1", retain=True) logger.debug("PDSL: stop streaming") if flag_reconfigure: if(CONFIG["in"]["device"] == "mls160a"): BACKEND = stream.MLS160A_stream( sz_block=CONFIG["in"]["frameSize"], channels_to_use=CONFIG["in"]["channels"], frequency=CONFIG["in"]["freq"], port=port, baudrate=baudrate) if(CONFIG["in"]["device"] == "io5640"): BACKEND = stream.IO5640_stream( sz_block=CONFIG["in"]["frameSize"], channels_to_use=CONFIG["in"]["channels"], frequency=CONFIG["in"]["freq"], port=port, baudrate=baudrate) if(CONFIG["out"]["data"] == "fft"): freq = CONFIG["in"]["freq"] signlen = CONFIG["in"]["frameSize"] period = 1.0/freq FFT = fft.FFTGenerator(period, signlen, freq) logger.debug("PDSL: create streamer '"+CONFIG["in"]["device"]+"'") if BACKEND != None and not is_streaming and flag_start: BACKEND.connect() BACKEND.start() is_streaming = True mqttc.publish(TOPIC_STATUS, "2", retain=True) logger.debug("PDSL: start streaming") # reset all flags flag_reconfigure = False flag_stop = False flag_start = False if __name__ == "__main__": loglevel = logging.INFO sys_handler = logging.handlers.SysLogHandler(address = '/dev/log') sys_formatter = logging.Formatter('%(filename)s [%(process)d]: %(levelname)s %(message)s') sys_handler.setFormatter(sys_formatter) std_handler = logging.StreamHandler(stream=sys.stdout) std_formatter = logging.Formatter("%(filename)s %(asctime)s - %(levelname)s - %(message)s") std_handler.setFormatter(std_formatter) for arg in sys.argv[1:]: if arg == '-f': logger.addHandler(std_handler) elif arg == '-d': loglevel = logging.DEBUG elif arg[0:2] == '-v': loglevel = int(arg[2:]) elif arg == '-h': print("ssvmq-pydslogd v"+VERSION+" deamon.") print("Usage: ssvmq-pydslogd [-h] [-f] [-d] [-vN]") print("\t-f log also to stdout") print("\t-d debug loglevel") print("\t-vN verbose loglevel, default INFO") print("\t-h this help") exit(1) logger.addHandler(sys_handler) logger.setLevel(loglevel) logger.info("Starting ssvmq-pydslogd v" + VERSION) signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler) mqttc = mqtt.Client() try: mqhost = "127.0.0.1" mqport = 7883 mqttc.on_connect = on_connect mqttc.on_message = on_message mqttc.on_subscribe = on_subscribe mqttc.on_unsubscribe = on_unsubscribe mqttc.will_set(TOPIC_STATUS, "0", retain=True) logger.info("MQTT: connect to mqtt %s:%s broker.", mqhost, mqport) mqttc.connect(mqhost, mqport, 60) mqttc.loop_start() logger.info("Running...") while Grun > 0: if is_connected: if flag_reconfigure or flag_stop or flag_start: pdsl_restart() if is_error: mqttc.publish(TOPIC_ERROR, json.dumps({'error':str(is_error)}), retain=False) set_error() if is_streaming: v = BACKEND.read(transpose=False) data = { "time": datetime.now().replace(microsecond=0).isoformat(), "data": {} } #print(v[0]) for i in range(0,len(CONFIG["in"]["channels"])): data["data"].update({str(CONFIG["in"]["channels"][i]):v[i]}) mqttc.publish(TOPIC_DATA, json.dumps(data), retain=False) continue time.sleep(1) #print("waiting...") mqttc.unsubscribe(TOPIC_CONFIG) if is_streaming: BACKEND.stop() BACKEND.disconnect() mqttc.loop_stop() except Exception as e: logger.exception(e) pass logger.info("Stopped ssvmq-pydslogd")