openwebrx-clone/owrx/connection.py

from owrx.config import PropertyManager
from owrx.source import DspManager, CpuUsageThread, SdrService, ClientRegistry
from owrx.feature import FeatureDetector
from owrx.version import openwebrx_version
from owrx.bands import Bandplan
from owrx.map import Map
from multiprocessing import Queue
import json
import threading

import logging

logger = logging.getLogger(__name__)


class Client(object):
    def __init__(self, conn):
        self.conn = conn
        self.multiprocessingPipe = Queue()

        def mp_passthru():
            run = True
            while run:
                try:
                    data = self.multiprocessingPipe.get()
                    self.send(data)
                except (EOFError, OSError):
                    run = False

        threading.Thread(target=mp_passthru).start()

    def send(self, data):
        self.conn.send(data)

    def close(self):
        self.conn.close()
        self.multiprocessingPipe.close()

    def mp_send(self, data):
        self.multiprocessingPipe.put(data, block=False)


class OpenWebRxReceiverClient(Client):
    config_keys = [
        "waterfall_colors",
        "waterfall_min_level",
        "waterfall_max_level",
        "waterfall_auto_level_margin",
        "lfo_offset",
        "samp_rate",
        "fft_size",
        "fft_fps",
        "audio_compression",
        "fft_compression",
        "max_clients",
        "start_mod",
        "client_audio_buffer_size",
        "start_freq",
        "center_freq",
        "mathbox_waterfall_colors",
        "mathbox_waterfall_history_length",
        "mathbox_waterfall_frequency_resolution",
    ]

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

        self.dsp = None
        self.sdr = None
        self.configSub = None

        ClientRegistry.getSharedInstance().addClient(self)

        pm = PropertyManager.getSharedInstance()

        self.setSdr()

        # send receiver info
        receiver_keys = [
            "receiver_name",
            "receiver_location",
            "receiver_qra",
            "receiver_asl",
            "receiver_gps",
            "photo_title",
            "photo_desc",
        ]
        receiver_details = dict((key, pm.getPropertyValue(key)) for key in receiver_keys)
        self.write_receiver_details(receiver_details)

        profiles = [
            {"name": s.getName() + " " + p["name"], "id": sid + "|" + pid}
            for (sid, s) in SdrService.getSources().items()
            for (pid, p) in s.getProfiles().items()
        ]
        self.write_profiles(profiles)

        features = FeatureDetector().feature_availability()
        self.write_features(features)

        CpuUsageThread.getSharedInstance().add_client(self)

    def setSdr(self, id=None):
        next = SdrService.getSource(id)
        if next == self.sdr:
            return

        self.stopDsp()

        if self.configSub is not None:
            self.configSub.cancel()
            self.configSub = None

        self.sdr = next

        # send initial config
        configProps = (
            self.sdr.getProps()
            .collect(*OpenWebRxReceiverClient.config_keys)
            .defaults(PropertyManager.getSharedInstance())
        )

        def sendConfig(key, value):
            config = dict((key, configProps[key]) for key in OpenWebRxReceiverClient.config_keys)
            # TODO mathematical properties? hmmmm
            config["start_offset_freq"] = configProps["start_freq"] - configProps["center_freq"]
            self.write_config(config)

            cf = configProps["center_freq"]
            srh = configProps["samp_rate"] / 2
            frequencyRange = (cf - srh, cf + srh)
            self.write_dial_frequendies(Bandplan.getSharedInstance().collectDialFrequencies(frequencyRange))

        self.configSub = configProps.wire(sendConfig)
        sendConfig(None, None)

        self.sdr.addSpectrumClient(self)

    def startDsp(self):
        if self.dsp is None:
            self.dsp = DspManager(self, self.sdr)
            self.dsp.start()

    def close(self):
        self.stopDsp()
        CpuUsageThread.getSharedInstance().remove_client(self)
        ClientRegistry.getSharedInstance().removeClient(self)
        if self.configSub is not None:
            self.configSub.cancel()
            self.configSub = None
        super().close()

    def stopDsp(self):
        if self.dsp is not None:
            self.dsp.stop()
            self.dsp = None
        if self.sdr is not None:
            self.sdr.removeSpectrumClient(self)

    def setParams(self, params):
        # only the keys in the protected property manager can be overridden from the web
        protected = (
            self.sdr.getProps()
            .collect("samp_rate", "center_freq", "rf_gain", "type", "if_gain")
            .defaults(PropertyManager.getSharedInstance())
        )
        for key, value in params.items():
            protected[key] = value

    def setDspProperties(self, params):
        for key, value in params.items():
            self.dsp.setProperty(key, value)

    def write_spectrum_data(self, data):
        self.send(bytes([0x01]) + data)

    def write_dsp_data(self, data):
        self.send(bytes([0x02]) + data)

    def write_s_meter_level(self, level):
        self.send({"type": "smeter", "value": level})

    def write_cpu_usage(self, usage):
        self.mp_send({"type": "cpuusage", "value": usage})

    def write_clients(self, clients):
        self.mp_send({"type": "clients", "value": clients})

    def write_secondary_fft(self, data):
        self.send(bytes([0x03]) + data)

    def write_secondary_demod(self, data):
        self.send(bytes([0x04]) + data)

    def write_secondary_dsp_config(self, cfg):
        self.send({"type": "secondary_config", "value": cfg})

    def write_config(self, cfg):
        self.send({"type": "config", "value": cfg})

    def write_receiver_details(self, details):
        self.send({"type": "receiver_details", "value": details})

    def write_profiles(self, profiles):
        self.send({"type": "profiles", "value": profiles})

    def write_features(self, features):
        self.send({"type": "features", "value": features})

    def write_metadata(self, metadata):
        self.send({"type": "metadata", "value": metadata})

    def write_wsjt_message(self, message):
        self.send({"type": "wsjt_message", "value": message})

    def write_dial_frequendies(self, frequencies):
        self.send({"type": "dial_frequencies", "value": frequencies})

    def write_aprs_data(self, data):
        self.send({"type": "aprs_data", "value": data})


class MapConnection(Client):
    def __init__(self, conn):
        super().__init__(conn)

        pm = PropertyManager.getSharedInstance()
        self.write_config(pm.collect("google_maps_api_key", "receiver_gps", "map_position_retention_time").__dict__())

        Map.getSharedInstance().addClient(self)

    def close(self):
        Map.getSharedInstance().removeClient(self)
        super().close()

    def write_config(self, cfg):
        self.send({"type": "config", "value": cfg})

    def write_update(self, update):
        self.mp_send({"type": "update", "value": update})


class WebSocketMessageHandler(object):
    def __init__(self):
        self.handshake = None
        self.client = None
        self.dsp = None

    def handleTextMessage(self, conn, message):
        if message[:16] == "SERVER DE CLIENT":
            meta = message[17:].split(" ")
            self.handshake = {v[0]: "=".join(v[1:]) for v in map(lambda x: x.split("="), meta)}

            conn.send("CLIENT DE SERVER server=openwebrx version={version}".format(version=openwebrx_version))
            logger.debug("client connection intitialized")

            if "type" in self.handshake:
                if self.handshake["type"] == "receiver":
                    self.client = OpenWebRxReceiverClient(conn)
                if self.handshake["type"] == "map":
                    self.client = MapConnection(conn)
            # backwards compatibility
            else:
                self.client = OpenWebRxReceiverClient(conn)

            return

        if not self.handshake:
            logger.warning("not answering client request since handshake is not complete")
            return

        try:
            message = json.loads(message)
            if "type" in message:
                if message["type"] == "dspcontrol":
                    if "action" in message and message["action"] == "start":
                        self.client.startDsp()

                    if "params" in message:
                        params = message["params"]
                        self.client.setDspProperties(params)

                if message["type"] == "config":
                    if "params" in message:
                        self.client.setParams(message["params"])
                if message["type"] == "setsdr":
                    if "params" in message:
                        self.client.setSdr(message["params"]["sdr"])
                if message["type"] == "selectprofile":
                    if "params" in message and "profile" in message["params"]:
                        profile = message["params"]["profile"].split("|")
                        self.client.setSdr(profile[0])
                        self.client.sdr.activateProfile(profile[1])
            else:
                logger.warning("received message without type: {0}".format(message))

        except json.JSONDecodeError:
            logger.warning("message is not json: {0}".format(message))

    def handleBinaryMessage(self, conn, data):
        logger.error("unsupported binary message, discarding")

    def handleClose(self):
        if self.client:
            self.client.close()