openwebrx-clone/owrx/connection.py

530 lines
18 KiB
Python
Raw Normal View History

2020-05-10 15:27:46 +00:00
from owrx.details import ReceiverDetails
from owrx.dsp import DspManager
from owrx.cpu import CpuUsageThread
from owrx.sdr import SdrService
from owrx.source import SdrSourceState, SdrClientClass, SdrSourceEventClient
from owrx.client import ClientRegistry, TooManyClientsException
from owrx.feature import FeatureDetector
from owrx.version import openwebrx_version
from owrx.bands import Bandplan
2019-09-27 22:53:58 +00:00
from owrx.bookmarks import Bookmarks
from owrx.map import Map
from owrx.property import PropertyStack, PropertyDeleted
from owrx.modes import Modes, DigitalMode
from owrx.config import Config
2021-02-16 16:34:04 +00:00
from owrx.waterfall import WaterfallOptions
from owrx.websocket import Handler
2020-09-15 20:04:53 +00:00
from queue import Queue, Full, Empty
from js8py import Js8Frame
2021-05-18 18:44:05 +00:00
from abc import ABCMeta, abstractmethod
2021-08-31 14:54:37 +00:00
from io import BytesIO
import json
import threading
2021-07-24 22:05:48 +00:00
import struct
2021-08-31 14:54:37 +00:00
import pickle
import logging
logger = logging.getLogger(__name__)
PoisonPill = object()
class Client(Handler, metaclass=ABCMeta):
2019-07-01 17:49:58 +00:00
def __init__(self, conn):
self.conn = conn
self.multithreadingQueue = Queue(100)
def mp_passthru():
run = True
while run:
try:
data = self.multithreadingQueue.get()
if data is PoisonPill:
run = False
else:
self.send(data)
self.multithreadingQueue.task_done()
except (EOFError, OSError, ValueError):
run = False
except Exception:
logger.exception("Exception on client multithreading queue")
# unset the queue object to free shared memory file descriptors
self.multithreadingQueue = None
2020-08-14 18:22:25 +00:00
threading.Thread(target=mp_passthru, name="connection_mp_passthru").start()
2019-07-01 17:49:58 +00:00
2019-09-22 11:16:24 +00:00
def send(self, data):
2020-07-19 17:00:26 +00:00
try:
self.conn.send(data)
except IOError:
self.close()
2019-07-01 17:49:58 +00:00
def close(self):
if self.multithreadingQueue is not None:
2020-09-15 20:04:53 +00:00
while True:
try:
self.multithreadingQueue.get(block=False)
except Empty:
break
try:
self.multithreadingQueue.put(PoisonPill, block=False)
except Full:
# this shouldn't happen, we just emptied the queue, but it's not worth risking the exception
logger.exception("impossible queue state: Full after Empty")
2019-07-01 17:49:58 +00:00
self.conn.close()
def mp_send(self, data):
if self.multithreadingQueue is None:
return
2019-12-08 20:11:36 +00:00
try:
self.multithreadingQueue.put(data, block=False)
2019-12-08 20:11:36 +00:00
except Full:
self.close()
2019-07-01 17:49:58 +00:00
@abstractmethod
2019-09-27 22:27:42 +00:00
def handleTextMessage(self, conn, message):
pass
def handleBinaryMessage(self, conn, data):
logger.error("unsupported binary message, discarding")
def handleClose(self):
self.close()
2019-07-01 17:49:58 +00:00
2020-05-08 22:20:38 +00:00
class OpenWebRxClient(Client, metaclass=ABCMeta):
def __init__(self, conn):
super().__init__(conn)
2020-05-10 15:27:46 +00:00
receiver_details = ReceiverDetails()
2020-05-08 22:20:38 +00:00
def send_receiver_info(*args):
receiver_info = receiver_details.__dict__()
self.write_receiver_details(receiver_info)
2021-02-15 16:25:46 +00:00
self._detailsSubscription = receiver_details.wire(send_receiver_info)
2020-05-08 22:20:38 +00:00
send_receiver_info()
def write_receiver_details(self, details):
self.send({"type": "receiver_details", "value": details})
2021-02-15 16:25:46 +00:00
def close(self):
self._detailsSubscription.cancel()
super().close()
2020-05-08 22:20:38 +00:00
class OpenWebRxReceiverClient(OpenWebRxClient, SdrSourceEventClient):
sdr_config_keys = [
"waterfall_levels",
"samp_rate",
"start_mod",
"start_freq",
"center_freq",
"initial_squelch_level",
"sdr_id",
2019-11-23 16:22:20 +00:00
"profile_id",
"squelch_auto_margin",
]
global_config_keys = [
2021-02-16 16:34:04 +00:00
"waterfall_scheme",
"waterfall_colors",
2021-03-30 22:00:38 +00:00
"waterfall_auto_levels",
"waterfall_auto_min_range",
"fft_size",
"audio_compression",
"fft_compression",
"max_clients",
"tuning_precision",
]
2019-07-01 17:49:58 +00:00
def __init__(self, conn):
2019-07-01 17:49:58 +00:00
super().__init__(conn)
self.dsp = None
2021-04-27 14:58:23 +00:00
self.dspLock = threading.Lock()
self.sdr = None
self.configSubs = []
self.bookmarkSub = None
2019-11-26 19:10:26 +00:00
self.connectionProperties = {}
try:
ClientRegistry.getSharedInstance().addClient(self)
except TooManyClientsException:
2020-01-10 20:43:21 +00:00
self.write_backoff_message("Too many clients")
self.close()
raise
2019-06-03 22:39:22 +00:00
self.setupGlobalConfig()
self.stack = self.setupStack()
self.setSdr()
2019-12-23 20:12:28 +00:00
features = FeatureDetector().feature_availability()
self.write_features(features)
modes = Modes.getModes()
self.write_modes(modes)
self.configSubs.append(SdrService.getActiveSources().wire(self._onSdrDeviceChanges))
self.configSubs.append(SdrService.getAvailableProfiles().wire(self._sendProfiles))
self._sendProfiles()
2019-12-23 20:12:28 +00:00
CpuUsageThread.getSharedInstance().add_client(self)
def setupStack(self):
stack = PropertyStack()
# stack layer 0 reserved for sdr properties
# stack.addLayer(0, self.sdr.getProps())
stack.addLayer(1, Config.get())
configProps = stack.filter(*OpenWebRxReceiverClient.sdr_config_keys)
def sendConfig(changes=None):
if changes is None:
config = configProps.__dict__()
else:
# transform deletions into Nones
config = {k: v if v is not PropertyDeleted else None for k, v in changes.items()}
2021-01-20 15:46:55 +00:00
if (
(changes is None or "start_freq" in changes or "center_freq" in changes)
and "start_freq" in configProps
and "center_freq" in configProps
):
config["start_offset_freq"] = configProps["start_freq"] - configProps["center_freq"]
if (changes is None or "profile_id" in changes) and self.sdr is not None:
config["sdr_id"] = self.sdr.getId()
self.write_config(config)
def sendBookmarks(*args):
cf = configProps["center_freq"]
srh = configProps["samp_rate"] / 2
2021-03-27 22:40:10 +00:00
dial_frequencies = []
bookmarks = []
if "center_freq" in configProps and "samp_rate" in configProps:
frequencyRange = (cf - srh, cf + srh)
dial_frequencies = Bandplan.getSharedInstance().collectDialFrequencies(frequencyRange)
bookmarks = [b.__dict__() for b in Bookmarks.getSharedInstance().getBookmarks(frequencyRange)]
self.write_dial_frequencies(dial_frequencies)
self.write_bookmarks(bookmarks)
def updateBookmarkSubscription(*args):
if self.bookmarkSub is not None:
self.bookmarkSub.cancel()
2021-03-27 22:40:10 +00:00
if "center_freq" in configProps and "samp_rate" in configProps:
cf = configProps["center_freq"]
srh = configProps["samp_rate"] / 2
frequencyRange = (cf - srh, cf + srh)
self.bookmarkSub = Bookmarks.getSharedInstance().subscribe(frequencyRange, sendBookmarks)
sendBookmarks()
self.configSubs.append(configProps.wire(sendConfig))
self.configSubs.append(stack.filter("center_freq", "samp_rate").wire(updateBookmarkSubscription))
# send initial config
sendConfig()
return stack
def setupGlobalConfig(self):
2021-02-16 16:34:04 +00:00
def writeConfig(changes):
# TODO it would be nicer to have all options available and switchable in the client
# this restores the existing functionality for now, but there is lots of potential
if "waterfall_scheme" in changes or "waterfall_colors" in changes:
scheme = WaterfallOptions(globalConfig["waterfall_scheme"]).instantiate()
2021-02-16 16:34:04 +00:00
changes["waterfall_colors"] = scheme.getColors()
self.write_config(changes)
globalConfig = Config.get().filter(*OpenWebRxReceiverClient.global_config_keys)
2021-02-16 16:34:04 +00:00
self.configSubs.append(globalConfig.wire(writeConfig))
writeConfig(globalConfig.__dict__())
2021-02-20 21:54:07 +00:00
def onStateChange(self, state: SdrSourceState):
if state is SdrSourceState.RUNNING:
self.handleSdrAvailable()
def onFail(self):
logger.warning('SDR device "%s" has failed, selecting new device', self.sdr.getName())
self.write_log_message('SDR device "{0}" has failed, selecting new device'.format(self.sdr.getName()))
self.setSdr()
2021-03-18 18:59:10 +00:00
def onDisable(self):
logger.warning('SDR device "%s" was disabled, selecting new device', self.sdr.getName())
self.write_log_message('SDR device "{0}" was disabled, selecting new device'.format(self.sdr.getName()))
self.setSdr()
def onShutdown(self):
logger.warning('SDR device "%s" is shutting down, selecting new device', self.sdr.getName())
self.write_log_message('SDR device "{0}" is shutting down, selecting new device'.format(self.sdr.getName()))
self.setSdr()
2021-02-20 21:54:07 +00:00
def getClientClass(self) -> SdrClientClass:
return SdrClientClass.USER
def _onSdrDeviceChanges(self, changes):
# restart the client if an sdr has become available
if self.sdr is None and any(s is not PropertyDeleted for s in changes.values()):
self.setSdr()
def _sendProfiles(self, *args):
profiles = [{"id": pid, "name": name} for pid, name in SdrService.getAvailableProfiles().items()]
self.write_profiles(profiles)
def handleTextMessage(self, conn, message):
try:
message = json.loads(message)
if "type" in message:
if message["type"] == "dspcontrol":
2021-04-26 23:58:20 +00:00
dsp = self.getDsp()
if dsp is None:
logger.warning("DSP not available; discarding client dspcontrol message")
else:
if "action" in message and message["action"] == "start":
dsp.start()
2021-04-26 23:58:20 +00:00
if "params" in message:
params = message["params"]
dsp.setProperties(params)
2019-11-26 19:10:26 +00:00
elif message["type"] == "setsdr":
if "params" in message:
self.setSdr(message["params"]["sdr"])
2019-11-26 19:10:26 +00:00
elif message["type"] == "selectprofile":
if "params" in message and "profile" in message["params"]:
profile = message["params"]["profile"].split("|")
self.setSdr(profile[0])
self.sdr.activateProfile(profile[1])
2019-11-26 19:10:26 +00:00
elif message["type"] == "connectionproperties":
if "params" in message:
self.connectionProperties = message["params"]
if self.dsp:
self.getDsp().setProperties(self.connectionProperties)
2019-11-26 19:10:26 +00:00
else:
logger.warning("received message without type: {0}".format(message))
except json.JSONDecodeError:
logger.warning("message is not json: {0}".format(message))
def setSdr(self, id=None):
next = None
if id is not None:
next = SdrService.getSource(id)
if next is None:
next = SdrService.getFirstSource()
# exit condition: no change
if next == self.sdr and next is not None:
return
self.stopDsp()
self.stack.removeLayerByPriority(0)
if self.sdr is not None:
self.sdr.removeClient(self)
self.sdr = next
if next is None:
# exit condition: no sdrs available
logger.warning("no more SDR devices available")
self.handleNoSdrsAvailable()
return
self.sdr.addClient(self)
def handleSdrAvailable(self):
self.getDsp().setProperties(self.connectionProperties)
self.stack.replaceLayer(0, self.sdr.getProps())
2019-11-26 19:10:26 +00:00
self.sdr.addSpectrumClient(self)
2019-12-23 20:12:28 +00:00
def handleNoSdrsAvailable(self):
self.write_sdr_error("No SDR Devices available")
2019-10-13 12:17:32 +00:00
def close(self):
if self.sdr is not None:
self.sdr.removeClient(self)
self.stopDsp()
CpuUsageThread.getSharedInstance().remove_client(self)
2019-05-12 16:10:24 +00:00
ClientRegistry.getSharedInstance().removeClient(self)
while self.configSubs:
self.configSubs.pop().cancel()
if self.bookmarkSub is not None:
self.bookmarkSub.cancel()
2021-03-27 22:40:10 +00:00
self.bookmarkSub = None
2019-07-01 17:49:58 +00:00
super().close()
def stopDsp(self):
2021-04-27 14:58:23 +00:00
with self.dspLock:
if self.dsp is not None:
self.dsp.stop()
self.dsp = None
if self.sdr is not None:
self.sdr.removeSpectrumClient(self)
def getDsp(self):
2021-04-27 14:58:23 +00:00
with self.dspLock:
if self.dsp is None and self.sdr is not None:
self.dsp = DspManager(self, self.sdr)
return self.dsp
def write_spectrum_data(self, data):
self.mp_send(bytes([0x01]) + data)
2019-07-01 17:49:58 +00:00
def write_dsp_data(self, data):
2019-09-22 11:16:24 +00:00
self.send(bytes([0x02]) + data)
2019-07-01 17:49:58 +00:00
2020-08-08 19:29:25 +00:00
def write_hd_audio(self, data):
self.send(bytes([0x04]) + data)
def write_s_meter_level(self, level):
2021-07-24 22:05:48 +00:00
if isinstance(level, memoryview):
2021-08-23 12:25:28 +00:00
# may contain more than one sample, so only take the last 4 bytes = 1 float
level, = struct.unpack('f', level[-4:])
2021-07-24 22:05:48 +00:00
if not isinstance(level, float):
logger.warning("s-meter value has unexpected type")
return
try:
self.send({"type": "smeter", "value": level})
except ValueError:
logger.warning("unable to send smeter value: %s", str(level))
2019-07-01 17:49:58 +00:00
def write_cpu_usage(self, usage):
self.mp_send({"type": "cpuusage", "value": usage})
2019-07-01 17:49:58 +00:00
def write_clients(self, clients):
self.mp_send({"type": "clients", "value": clients})
2019-07-01 17:49:58 +00:00
def write_secondary_fft(self, data):
2019-09-22 11:16:24 +00:00
self.send(bytes([0x03]) + data)
2019-07-01 17:49:58 +00:00
2021-09-06 13:05:33 +00:00
def write_secondary_demod(self, message):
io = BytesIO(message.tobytes())
try:
while True:
self.send({"type": "secondary_demod", "value": pickle.load(io)})
except EOFError:
pass
2019-07-01 17:49:58 +00:00
def write_secondary_dsp_config(self, cfg):
2019-09-22 11:16:24 +00:00
self.send({"type": "secondary_config", "value": cfg})
2019-07-01 17:49:58 +00:00
def write_config(self, cfg):
2019-09-22 11:16:24 +00:00
self.send({"type": "config", "value": cfg})
2019-07-01 17:49:58 +00:00
def write_profiles(self, profiles):
2019-09-22 11:16:24 +00:00
self.send({"type": "profiles", "value": profiles})
2019-07-01 17:49:58 +00:00
def write_features(self, features):
2019-09-22 11:16:24 +00:00
self.send({"type": "features", "value": features})
2019-07-01 17:49:58 +00:00
def write_metadata(self, metadata):
io = BytesIO(metadata.tobytes())
try:
while True:
self.send({"type": "metadata", "value": pickle.load(io)})
except EOFError:
pass
def write_dial_frequencies(self, frequencies):
2019-09-22 11:16:24 +00:00
self.send({"type": "dial_frequencies", "value": frequencies})
2019-09-27 22:53:58 +00:00
def write_bookmarks(self, bookmarks):
self.send({"type": "bookmarks", "value": bookmarks})
def write_log_message(self, message):
self.send({"type": "log_message", "value": message})
2019-10-12 18:46:32 +00:00
def write_sdr_error(self, message):
self.send({"type": "sdr_error", "value": message})
def write_backoff_message(self, reason):
self.send({"type": "backoff", "reason": reason})
def write_modes(self, modes):
def to_json(m):
res = {
"modulation": m.modulation,
"name": m.name,
"type": "digimode" if isinstance(m, DigitalMode) else "analog",
"requirements": m.requirements,
"squelch": m.squelch,
}
if m.bandpass is not None:
2021-01-20 15:46:55 +00:00
res["bandpass"] = {"low_cut": m.bandpass.low_cut, "high_cut": m.bandpass.high_cut}
if isinstance(m, DigitalMode):
res["underlying"] = m.underlying
return res
self.send({"type": "modes", "value": [to_json(m) for m in modes]})
2019-07-01 17:49:58 +00:00
2020-05-08 22:20:38 +00:00
class MapConnection(OpenWebRxClient):
2019-07-01 17:49:58 +00:00
def __init__(self, conn):
super().__init__(conn)
pm = Config.get()
filtered_config = pm.filter(
"google_maps_api_key",
"receiver_gps",
"map_position_retention_time",
"receiver_name",
2021-01-20 15:46:55 +00:00
)
filtered_config.wire(self.write_config)
self.write_config(filtered_config.__dict__())
2019-07-01 17:49:58 +00:00
Map.getSharedInstance().addClient(self)
def handleTextMessage(self, conn, message):
pass
def close(self):
Map.getSharedInstance().removeClient(self)
super().close()
2019-07-01 17:49:58 +00:00
def write_config(self, cfg):
2019-09-22 11:16:24 +00:00
self.send({"type": "config", "value": cfg})
2019-07-01 17:49:58 +00:00
def write_update(self, update):
self.mp_send({"type": "update", "value": update})
2019-07-01 17:49:58 +00:00
2021-05-18 14:00:15 +00:00
class HandshakeMessageHandler(Handler):
"""
This handler receives text messages, but will only respond to the second handshake string.
As soon as a valid handshake is received, the handler replaces itself with the corresponding handler type.
"""
def handleTextMessage(self, conn, message):
if message[:16] == "SERVER DE CLIENT":
meta = message[17:].split(" ")
2021-05-18 14:00:15 +00:00
handshake = {v[0]: "=".join(v[1:]) for v in map(lambda x: x.split("="), meta)}
2020-12-13 15:31:19 +00:00
logger.debug("client connection initialized")
2021-05-18 14:00:15 +00:00
client = None
if "type" in handshake:
if handshake["type"] == "receiver":
client = OpenWebRxReceiverClient(conn)
2021-05-18 14:00:15 +00:00
elif handshake["type"] == "map":
client = MapConnection(conn)
2021-05-18 14:00:15 +00:00
else:
logger.warning("invalid connection type: %s", handshake["type"])
2021-05-18 14:00:15 +00:00
if client is not None:
2021-05-18 18:46:33 +00:00
logger.debug("handshake complete, handing off to %s", type(client).__name__)
2021-05-18 14:00:15 +00:00
# hand off all further communication to the correspondig connection
conn.send("CLIENT DE SERVER server=openwebrx version={version}".format(version=openwebrx_version))
conn.setMessageHandler(client)
else:
logger.warning('invalid handshake received')
else:
logger.warning("not answering client request since handshake is not complete")
2019-09-27 22:27:42 +00:00
def handleBinaryMessage(self, conn, data):
pass
def handleClose(self):
pass