openwebrx-clone/owrx/connection.py

481 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
from owrx.source import SdrSource, 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
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, SdrSourceEventClient):
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",
"squelch_auto_margin",
]
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 onStateChange(self, state):
if state == SdrSource.STATE_RUNNING:
self.handleSdrAvailable()
elif state == SdrSource.STATE_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()
def onBusyStateChange(self, state):
pass
def getClientClass(self):
return SdrSource.CLIENT_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.configSub is not None:
self.configSub.cancel()
self.configSub = None
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):
# 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):
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)
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