openwebrx-clone/owrx/connection.py

465 lines
15 KiB
Python
Raw Normal View History

from owrx.config import Config
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
2019-12-23 20:12:28 +00:00
from owrx.source import SdrSource
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
from owrx.modes import Modes, DigitalMode
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)
# TODO unsubscribe
receiver_details.wire(send_receiver_info)
send_receiver_info()
def write_receiver_details(self, details):
self.send({"type": "receiver_details", "value": details})
class OpenWebRxReceiverClient(OpenWebRxClient):
config_keys = [
"waterfall_colors",
"waterfall_min_level",
"waterfall_max_level",
"waterfall_auto_level_margin",
"samp_rate",
"fft_size",
"audio_compression",
"fft_compression",
"max_clients",
"start_mod",
"start_freq",
"center_freq",
"initial_squelch_level",
2019-11-23 16:22:20 +00:00
"profile_id",
]
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.configSub = 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.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 __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):
while True:
next = None
if id is not None:
next = SdrService.getSource(id)
if next is None:
next = SdrService.getFirstSource()
if next is None:
# exit condition: no sdrs available
logger.warning("no more SDR devices available")
self.handleNoSdrsAvailable()
return
2019-10-13 12:17:32 +00:00
# exit condition: no change
if next == self.sdr:
return
self.stopDsp()
if self.configSub is not None:
self.configSub.cancel()
self.configSub = None
self.sdr = next
self.getDsp()
# found a working sdr, exit the loop
if self.sdr.getState() != SdrSource.STATE_FAILED:
break
logger.warning('SDR device "%s" has failed, selecing new device', self.sdr.getName())
self.write_log_message('SDR device "{0}" has failed, selecting new device'.format(self.sdr.getName()))
# send initial config
self.getDsp().setProperties(self.connectionProperties)
2019-11-26 19:10:26 +00:00
stack = PropertyStack()
stack.addLayer(0, self.sdr.getProps())
stack.addLayer(1, Config.get())
2020-03-24 21:16:11 +00:00
configProps = stack.filter(*OpenWebRxReceiverClient.config_keys)
def sendConfig(key, value):
config = configProps.__dict__()
# TODO mathematical properties? hmmmm
config["start_offset_freq"] = configProps["start_freq"] - configProps["center_freq"]
2019-11-23 16:22:20 +00:00
# TODO this is a hack to support multiple sdrs
config["sdr_id"] = self.sdr.getId()
self.write_config(config)
cf = configProps["center_freq"]
srh = configProps["samp_rate"] / 2
frequencyRange = (cf - srh, cf + srh)
2019-07-21 21:39:11 +00:00
self.write_dial_frequendies(Bandplan.getSharedInstance().collectDialFrequencies(frequencyRange))
2019-09-27 22:53:58 +00:00
bookmarks = [b.__dict__() for b in Bookmarks.getSharedInstance().getBookmarks(frequencyRange)]
self.write_bookmarks(bookmarks)
self.configSub = configProps.wire(sendConfig)
sendConfig(None, None)
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):
self.stopDsp()
CpuUsageThread.getSharedInstance().remove_client(self)
2019-05-12 16:10:24 +00:00
ClientRegistry.getSharedInstance().removeClient(self)
if self.configSub is not None:
self.configSub.cancel()
self.configSub = None
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
# only the keys in the protected property manager can be overridden from the web
stack = PropertyStack()
stack.addLayer(0, self.sdr.getProps())
stack.addLayer(1, config)
2020-03-24 21:16:11 +00:00
protected = 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_frequendies(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):
self.send({"type": "js8_message", "value": {
"msg": str(frame),
"timestamp": frame.timestamp,
"db": frame.db,
"dt": frame.dt,
"freq": freq + frame.freq,
2020-04-24 21:47:05 +00:00
"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:
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()
2020-03-24 21:16:11 +00:00
self.write_config(pm.filter("google_maps_api_key", "receiver_gps", "map_position_retention_time").__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))
logger.debug("client connection intitialized")
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