openwebrx-clone/owrx/source/__init__.py
2022-12-11 23:48:56 +01:00

664 lines
21 KiB
Python

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(),
)