openwebrx-clone/owrx/connection.py

523 lines
17 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
2021-02-20 21:54:07 +00:00
from owrx.source import SdrSourceState, SdrBusyState, 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
2020-09-15 20:04:53 +00:00
from queue import Queue, Full, Empty
from js8py import Js8Frame
2020-05-08 22:20:38 +00:00
from abc import ABC, ABCMeta, abstractmethod
import json
import threading
import logging
logger = logging.getLogger(__name__)
PoisonPill = object()
class Client(ABC):
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",
"waterfall_auto_level_margin",
"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
self.sdr = None
self.configSubs = []
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.__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(changes=None):
cf = configProps["center_freq"]
srh = configProps["samp_rate"] / 2
frequencyRange = (cf - srh, cf + srh)
self.write_dial_frequencies(Bandplan.getSharedInstance().collectDialFrequencies(frequencyRange))
bookmarks = [b.__dict__() for b in Bookmarks.getSharedInstance().getBookmarks(frequencyRange)]
self.write_bookmarks(bookmarks)
self.configSubs.append(configProps.wire(sendConfig))
self.configSubs.append(stack.filter("center_freq", "samp_rate").wire(sendBookmarks))
# 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()
2021-02-20 21:54:07 +00:00
elif state is SdrSourceState.FAILED:
self.handleSdrFailed()
def handleSdrFailed(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-02-20 21:54:07 +00:00
def onBusyStateChange(self, state: SdrBusyState):
pass
2021-02-20 21:54:07 +00:00
def getClientClass(self) -> SdrClientClass:
return SdrClientClass.USER
2019-12-23 20:12:28 +00:00
def __sendProfiles(self):
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)
def handleTextMessage(self, conn, message):
try:
message = json.loads(message)
if "type" in message:
if message["type"] == "dspcontrol":
if "action" in message and message["action"] == "start":
self.startDsp()
if "params" in message:
dsp = self.getDsp()
if dsp is None:
logger.warning("DSP not available; discarding client data")
else:
params = message["params"]
dsp.setProperties(params)
2019-11-26 19:10:26 +00:00
elif message["type"] == "config":
if "params" in message:
self.setParams(message["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()
if self.sdr is not None:
self.sdr.removeClient(self)
if next is None:
# exit condition: no sdrs available
logger.warning("no more SDR devices available")
self.handleNoSdrsAvailable()
return
self.sdr = next
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
2019-12-23 20:12:28 +00:00
self.__sendProfiles()
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 startDsp(self):
self.getDsp().start()
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()
2019-07-01 17:49:58 +00:00
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):
config = Config.get()
# allow direct configuration only if enabled in the config
2020-05-03 15:50:37 +00:00
if "configurable_keys" not in config:
return
keys = config["configurable_keys"]
if not keys:
return
protected = self.stack.filter(*keys)
for key, value in params.items():
try:
2020-05-03 19:28:37 +00:00
protected[key] = value
except KeyError:
pass
def getDsp(self):
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):
2019-09-22 11:16:24 +00:00
self.send({"type": "smeter", "value": 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
def write_secondary_demod(self, data):
2020-01-09 12:49:38 +00:00
message = data.decode("ascii", "replace")
self.send({"type": "secondary_demod", "value": message})
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):
2019-09-22 11:16:24 +00:00
self.send({"type": "metadata", "value": metadata})
2019-07-06 18:03:17 +00:00
def write_wsjt_message(self, message):
2019-09-22 11:16:24 +00:00
self.send({"type": "wsjt_message", "value": message})
2019-07-06 18:03:17 +00:00
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_aprs_data(self, data):
2019-09-22 11:16:24 +00:00
self.send({"type": "aprs_data", "value": data})
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})
2020-01-09 14:12:51 +00:00
def write_pocsag_data(self, data):
self.send({"type": "pocsag_data", "value": data})
def write_backoff_message(self, reason):
self.send({"type": "backoff", "reason": reason})
def write_js8_message(self, frame: Js8Frame, freq: int):
2021-01-20 15:46:55 +00:00
self.send(
{
"type": "js8_message",
"value": {
"msg": str(frame),
"timestamp": frame.timestamp,
"db": frame.db,
"dt": frame.dt,
"freq": freq + frame.freq,
"thread_type": frame.thread_type,
"mode": frame.mode,
},
}
)
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
class WebSocketMessageHandler(object):
def __init__(self):
self.handshake = 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))
2020-12-13 15:31:19 +00:00
logger.debug("client connection initialized")
if "type" in self.handshake:
if self.handshake["type"] == "receiver":
client = OpenWebRxReceiverClient(conn)
2019-07-01 17:49:58 +00:00
if self.handshake["type"] == "map":
client = MapConnection(conn)
# backwards compatibility
else:
client = OpenWebRxReceiverClient(conn)
# hand off all further communication to the correspondig connection
conn.setMessageHandler(client)
return
if not self.handshake:
logger.warning("not answering client request since handshake is not complete")
return
2019-09-27 22:27:42 +00:00
def handleBinaryMessage(self, conn, data):
pass
def handleClose(self):
pass