openwebrx-clone/owrx/source/__init__.py

from owrx.config import Config
import threading
import subprocess
import os
import socket
import shlex
import time
import signal
import pkgutil
from abc import ABC, abstractmethod
from owrx.command import CommandMapper
from owrx.socket import getAvailablePort
from owrx.property import PropertyStack, PropertyLayer, PropertyFilter, PropertyCarousel, PropertyDeleted
from owrx.property.filter import ByLambda
from owrx.form.input import Input, TextInput, NumberInput, CheckboxInput, ModesInput, ExponentialInput
from owrx.form.input.converter import OptionalConverter
from owrx.form.input.device import GainInput, SchedulerInput, WaterfallLevelsInput
from owrx.form.input.validator import RequiredValidator
from owrx.form.input.converter import Converter
from owrx.form.section import OptionalSection
from owrx.feature import FeatureDetector
from owrx.log import LogPipe, HistoryHandler
from typing import List
from enum import Enum

from pycsdr.modules import TcpSource, Buffer
from pycsdr.types import Format

import logging

logger = logging.getLogger(__name__)


class SdrSourceState(Enum):
    STOPPED = "Stopped"
    STARTING = "Starting"
    RUNNING = "Running"
    STOPPING = "Stopping"
    TUNING = "Tuning"

    def __str__(self):
        return self.value


class SdrBusyState(Enum):
    IDLE = 1
    BUSY = 2


class SdrClientClass(Enum):
    INACTIVE = 1
    BACKGROUND = 2
    USER = 3


class SdrSourceEventClient(object):
    def onStateChange(self, state: SdrSourceState):
        pass

    def onBusyStateChange(self, state: SdrBusyState):
        pass

    def onFail(self):
        pass

    def onShutdown(self):
        pass

    def onDisable(self):
        pass

    def onEnable(self):
        pass

    def getClientClass(self) -> SdrClientClass:
        return SdrClientClass.INACTIVE


class SdrProfileCarousel(PropertyCarousel):
    def __init__(self, props):
        super().__init__()
        if "profiles" not in props:
            return

        for profile_id, profile in props["profiles"].items():
            self.addLayer(profile_id, profile)
        # activate first available profile
        self.switch()

        props["profiles"].wire(self.handleProfileUpdate)

    def addLayer(self, profile_id, profile):
        profile_stack = PropertyStack()
        profile_stack.addLayer(0, PropertyLayer(profile_id=profile_id).readonly())
        profile_stack.addLayer(1, profile)
        super().addLayer(profile_id, profile_stack)

    def handleProfileUpdate(self, changes):
        for profile_id, profile in changes.items():
            if profile is PropertyDeleted:
                self.removeLayer(profile_id)
            else:
                self.addLayer(profile_id, profile)

    def _getDefaultLayer(self):
        # return the first available profile, or the default empty layer if we don't have any
        if self.layers:
            return next(iter(self.layers.values()))
        return super()._getDefaultLayer()


class SdrSource(ABC):
    def __init__(self, id, props):
        self.id = id

        self.commandMapper = None
        self.tcpSource = None
        self.buffer = None
        self.logger = logger.getChild(id) if id is not None else logger
        self.logger.addHandler(HistoryHandler.getHandler(self.logger.name))
        self.stdoutPipe = None
        self.stderrPipe = None

        self.props = PropertyStack()

        # layer 0 reserved for profile properties
        self.profileCarousel = SdrProfileCarousel(props)
        # prevent profile names from overriding the device name
        self.props.addLayer(0, PropertyFilter(self.profileCarousel, ByLambda(lambda x: x != "name")))

        # props from our device config
        self.props.addLayer(1, props)

        # the sdr_id is constant, so we put it in a separate layer
        # this is used to detect device changes, that are then sent to the client
        self.props.addLayer(2, PropertyLayer(sdr_id=id).readonly())

        # finally, accept global config properties from the top-level config
        self.props.addLayer(3, Config.get())

        self.sdrProps = self.props.filter(*self.getEventNames())

        self.wireEvents()

        self.port = getAvailablePort()
        self.monitor = None
        self.clients = []
        self.spectrumClients = []
        self.spectrumThread = None
        self.spectrumLock = threading.Lock()
        self.process = None
        self.modificationLock = threading.Lock()
        self.state = SdrSourceState.STOPPED
        self.enabled = "enabled" not in props or props["enabled"]
        props.filter("enabled").wire(self._handleEnableChanged)
        self.failed = False
        self.busyState = SdrBusyState.IDLE

        self.validateProfiles()

        if self.isAlwaysOn() and self.isEnabled():
            self.start()

    def isEnabled(self):
        return self.enabled

    def _handleEnableChanged(self, changes):
        if "enabled" in changes and changes["enabled"] is not PropertyDeleted:
            self.enabled = changes["enabled"]
        else:
            self.enabled = True
        if not self.enabled:
            self.stop()
        for c in self.clients.copy():
            if self.isEnabled():
                c.onEnable()
            else:
                c.onDisable()

    def isFailed(self):
        return self.failed

    def fail(self):
        self.failed = True
        for c in self.clients.copy():
            c.onFail()

    def validateProfiles(self):
        props = PropertyStack()
        props.addLayer(1, self.props)
        for id, p in self.props["profiles"].items():
            props.replaceLayer(0, p)
            if "center_freq" not in props:
                self.logger.warning('Profile "%s" does not specify a center_freq', id)
                continue
            if "samp_rate" not in props:
                self.logger.warning('Profile "%s" does not specify a samp_rate', id)
                continue
            if "start_freq" in props:
                start_freq = props["start_freq"]
                srh = props["samp_rate"] / 2
                center_freq = props["center_freq"]
                if start_freq < center_freq - srh or start_freq > center_freq + srh:
                    self.logger.warning('start_freq for profile "%s" is out of range', id)

    def isAlwaysOn(self):
        return "always-on" in self.props and self.props["always-on"]

    def getEventNames(self):
        return [
            "samp_rate",
            "center_freq",
            "ppm",
            "rf_gain",
            "lfo_offset",
        ] + list(self.getCommandMapper().keys())

    def getCommandMapper(self):
        if self.commandMapper is None:
            self.commandMapper = CommandMapper()
        return self.commandMapper

    @abstractmethod
    def onPropertyChange(self, changes):
        pass

    def wireEvents(self):
        self.sdrProps.wire(self.onPropertyChange)

    def getCommand(self):
        return [self.getCommandMapper().map(self.getCommandValues())]

    def activateProfile(self, profile_id):
        try:
            profile_name = self.getProfiles()[profile_id]["name"]
            self.logger.debug("activating profile \"%s\" for \"%s\"", profile_name, self.getName())
            self.profileCarousel.switch(profile_id)
        except KeyError:
            self.logger.warning("invalid profile %s for sdr %s. ignoring", profile_id, self.getId())

    def getId(self):
        return self.id

    def getProfileId(self):
        return self.props["profile_id"]

    def getProfiles(self):
        return self.props["profiles"]

    def getName(self):
        return self.props["name"]

    def getProps(self):
        return self.props

    def getPort(self):
        return self.port

    def _getTcpSourceFormat(self):
        return Format.COMPLEX_FLOAT

    def _getTcpSource(self):
        with self.modificationLock:
            if self.tcpSource is None:
                self.tcpSource = TcpSource(self.port, self._getTcpSourceFormat())
        return self.tcpSource

    def getBuffer(self):
        if self.buffer is None:
            self.buffer = Buffer(Format.COMPLEX_FLOAT)
            self._getTcpSource().setWriter(self.buffer)
        return self.buffer

    def getCommandValues(self):
        dict = self.sdrProps.__dict__()
        if "lfo_offset" in dict and dict["lfo_offset"] is not None:
            dict["tuner_freq"] = dict["center_freq"] + dict["lfo_offset"]
        else:
            dict["tuner_freq"] = dict["center_freq"]
        return dict

    def start(self):
        with self.modificationLock:
            if self.monitor:
                return

            if self.isFailed():
                return

            try:
                self.preStart()
            except Exception:
                self.logger.exception("Exception during preStart()")

            cmd = self.getCommand()
            cmd = [c for c in cmd if c is not None]

            self.stdoutPipe = LogPipe(logging.INFO, self.logger, "STDOUT")
            self.stderrPipe = LogPipe(logging.WARNING, self.logger, "STDERR")

            # don't use shell mode for commands without piping
            if len(cmd) > 1:
                # multiple commands with pipes
                cmd = "|".join(cmd)
                self.process = subprocess.Popen(
                    cmd,
                    shell=True,
                    start_new_session=True,
                    stdout=self.stdoutPipe,
                    stderr=self.stderrPipe
                )
            else:
                # single command
                cmd = cmd[0]
                # start_new_session can go as soon as there's no piped commands left
                # the os.killpg call must be replaced with something more reasonable at the same time
                self.process = subprocess.Popen(
                    shlex.split(cmd),
                    start_new_session=True,
                    stdout=self.stdoutPipe,
                    stderr=self.stderrPipe
                )
            self.logger.info("Started sdr source: " + cmd)

            available = False
            failed = False

            def wait_for_process_to_end():
                nonlocal failed
                rc = self.process.wait()
                self.logger.debug("shut down with RC={0}".format(rc))
                self.process = None
                self.monitor = None
                self.stdoutPipe.close()
                self.stdoutPipe = None
                self.stderrPipe.close()
                self.stderrPipe = None
                if self.getState() is SdrSourceState.RUNNING:
                    self.fail()
                else:
                    failed = True
                self.setState(SdrSourceState.STOPPED)

            self.monitor = threading.Thread(target=wait_for_process_to_end, name="source_monitor")
            self.monitor.start()

            retries = 1000
            while retries > 0 and not failed:
                retries -= 1
                if self.monitor is None:
                    break
                testsock = socket.socket()
                testsock.settimeout(1)
                try:
                    testsock.connect(("127.0.0.1", self.getPort()))
                    testsock.close()
                    available = True
                    break
                except:
                    time.sleep(0.1)

            if not available:
                failed = True

            try:
                self.postStart()
            except Exception:
                self.logger.exception("Exception during postStart()")
                failed = True

        if failed:
            self.fail()
        else:
            self.setState(SdrSourceState.RUNNING)

    def preStart(self):
        """
        override this method in subclasses if there's anything to be done before starting up the actual SDR
        """
        pass

    def postStart(self):
        """
        override this method in subclasses if there's things to do after the actual SDR has started up
        """
        pass

    def isAvailable(self):
        return self.monitor is not None

    def stop(self):
        with self.modificationLock:
            if self.process is not None:
                self.setState(SdrSourceState.STOPPING)
                try:
                    os.killpg(os.getpgid(self.process.pid), signal.SIGTERM)
                    if self.monitor:
                        # wait 10 seconds for a regular shutdown
                        self.monitor.join(10)
                        # if the monitor is still running, the process still hasn't ended, so kill it
                    if self.monitor:
                        self.logger.warning("source has not shut down normally within 10 seconds, sending SIGKILL")
                        os.killpg(os.getpgid(self.process.pid), signal.SIGKILL)
                except ProcessLookupError:
                    # been killed by something else, ignore
                    pass
                except AttributeError:
                    # self.process has been overwritten by the monitor since we checked it, which is fine
                    pass
            if self.monitor:
                self.monitor.join()
            if self.tcpSource is not None:
                self.tcpSource.stop()
                self.tcpSource = None
                self.buffer = None

    def shutdown(self):
        self.stop()
        for c in self.clients.copy():
            c.onShutdown()

    def getClients(self, *args):
        if not args:
            return self.clients
        return [c for c in self.clients if c.getClientClass() in args]

    def hasClients(self, *args):
        return len(self.getClients(*args)) > 0

    def addClient(self, c: SdrSourceEventClient):
        if c in self.clients:
            return
        self.clients.append(c)
        c.onStateChange(self.getState())
        hasUsers = self.hasClients(SdrClientClass.USER)
        hasBackgroundTasks = self.hasClients(SdrClientClass.BACKGROUND)
        if hasUsers or hasBackgroundTasks:
            self.start()
            self.setBusyState(SdrBusyState.BUSY if hasUsers else SdrBusyState.IDLE)

    def removeClient(self, c: SdrSourceEventClient):
        if c not in self.clients:
            return

        self.clients.remove(c)

        self.checkStatus()

    def checkStatus(self):
        hasUsers = self.hasClients(SdrClientClass.USER)
        self.setBusyState(SdrBusyState.BUSY if hasUsers else SdrBusyState.IDLE)

        # no need to check for users if we are always-on
        if self.isAlwaysOn():
            return

        hasBackgroundTasks = self.hasClients(SdrClientClass.BACKGROUND)
        if not hasUsers and not hasBackgroundTasks:
            self.stop()

    def addSpectrumClient(self, c):
        if c in self.spectrumClients:
            return

        # local import due to circular depencency
        from owrx.fft import SpectrumThread

        self.spectrumClients.append(c)
        with self.spectrumLock:
            if self.spectrumThread is None:
                self.spectrumThread = SpectrumThread(self)
                self.spectrumThread.start()

    def removeSpectrumClient(self, c):
        try:
            self.spectrumClients.remove(c)
        except ValueError:
            pass
        with self.spectrumLock:
            if not self.spectrumClients and self.spectrumThread is not None:
                self.spectrumThread.stop()
                self.spectrumThread = None

    def writeSpectrumData(self, data):
        for c in self.spectrumClients:
            c.write_spectrum_data(data)

    def getState(self) -> SdrSourceState:
        return self.state

    def setState(self, state: SdrSourceState):
        if state == self.state:
            return
        self.state = state
        for c in self.clients.copy():
            c.onStateChange(state)

    def setBusyState(self, state: SdrBusyState):
        if state == self.busyState:
            return
        self.busyState = state
        for c in self.clients.copy():
            c.onBusyStateChange(state)


class SdrDeviceDescriptionMissing(Exception):
    pass


class SdrDeviceTypeConverter(Converter):
    def convert_to_form(self, value):
        # local import due to circular dependendies
        types = SdrDeviceDescription.getTypes()
        if value in types:
            return types[value]
        return value

    def convert_from_form(self, value):
        return None


class SdrDeviceTypeDisplay(Input):
    """
    Not an input per se, just an element that can display the SDR device type in the web config
    """
    def __init__(self, id, label):
        super().__init__(id, label, disabled=True)

    def defaultConverter(self):
        return SdrDeviceTypeConverter()

    def parse(self, data):
        return {}


class SdrDeviceDescription(object):
    @staticmethod
    def getByType(sdr_type: str) -> "SdrDeviceDescription":
        try:
            className = "".join(x for x in sdr_type.title() if x.isalnum()) + "DeviceDescription"
            module = __import__("owrx.source.{0}".format(sdr_type), fromlist=[className])
            cls = getattr(module, className)
            return cls()
        except (ImportError, AttributeError):
            raise SdrDeviceDescriptionMissing("Device description for type {} not available".format(sdr_type))

    @staticmethod
    def getTypes():
        def get_description(module_name):
            try:
                description = SdrDeviceDescription.getByType(module_name)
                return description.getName()
            except SdrDeviceDescriptionMissing:
                return None

        descriptions = {
            module_name: get_description(module_name) for _, module_name, _ in pkgutil.walk_packages(__path__)
        }
        # filter out empty names and unavailable types
        fd = FeatureDetector()
        return {k: v for k, v in descriptions.items() if v is not None and fd.is_available(k)}

    def getName(self):
        """
        must be overridden with a textual representation of the device, to be used for device type selection

        :return: str
        """
        return None

    def supportsPpm(self):
        """
        can be overridden if the device does not support configuring PPM correction

        :return: bool
        """
        return True

    def getDeviceInputs(self) -> List[Input]:
        keys = self.getDeviceMandatoryKeys() + self.getDeviceOptionalKeys()
        return [TextInput("name", "Device name", validator=RequiredValidator())] + [
            i for i in self.getInputs() if i.id in keys
        ]

    def getProfileInputs(self) -> List[Input]:
        keys = self.getProfileMandatoryKeys() + self.getProfileOptionalKeys()
        return [TextInput("name", "Profile name", validator=RequiredValidator())] + [
            i for i in self.getInputs() if i.id in keys
        ]

    def getInputs(self) -> List[Input]:
        return [
            SdrDeviceTypeDisplay("type", "Device type"),
            CheckboxInput("enabled", "Enable this device", converter=OptionalConverter(defaultFormValue=True)),
            GainInput("rf_gain", "Device gain", self.hasAgc()),
            NumberInput(
                "ppm",
                "Frequency correction",
                append="ppm",
            ),
            CheckboxInput(
                "always-on",
                "Keep device running at all times",
                infotext="Prevents shutdown of the device when idle. Useful for devices with unreliable startup.",
            ),
            CheckboxInput(
                "services",
                "Run background services on this device",
            ),
            ExponentialInput(
                "lfo_offset",
                "Oscillator offset",
                "Hz",
                infotext="Use this when the actual receiving frequency differs from the frequency to be tuned on the"
                + " device. <br/> Formula: Center frequency + oscillator offset = sdr tune frequency",
            ),
            WaterfallLevelsInput("waterfall_levels", "Waterfall levels"),
            SchedulerInput("scheduler", "Scheduler"),
            ExponentialInput("center_freq", "Center frequency", "Hz"),
            ExponentialInput("samp_rate", "Sample rate", "S/s"),
            ExponentialInput("start_freq", "Initial frequency", "Hz"),
            ModesInput("start_mod", "Initial modulation"),
            NumberInput("initial_squelch_level", "Initial squelch level", append="dBFS"),
        ]

    def hasAgc(self):
        # default is True since most devices have agc. override in subclasses if agc is not available
        return True

    def getDeviceMandatoryKeys(self):
        return ["name", "type", "enabled"]

    def getDeviceOptionalKeys(self):
        keys = [
            "always-on",
            "services",
            "rf_gain",
            "lfo_offset",
            "waterfall_levels",
            "scheduler",
        ]
        if self.supportsPpm():
            keys += ["ppm"]
        return keys

    def getProfileMandatoryKeys(self):
        return ["name", "center_freq", "samp_rate", "start_freq", "start_mod"]

    def getProfileOptionalKeys(self):
        return ["initial_squelch_level", "rf_gain", "lfo_offset", "waterfall_levels"]

    def getDeviceSection(self):
        return OptionalSection(
            "Device settings", self.getDeviceInputs(), self.getDeviceMandatoryKeys(), self.getDeviceOptionalKeys()
        )

    def getProfileSection(self):
        return OptionalSection(
            "Profile settings",
            self.getProfileInputs(),
            self.getProfileMandatoryKeys(),
            self.getProfileOptionalKeys(),
        )