395 lines
14 KiB
Python
395 lines
14 KiB
Python
from driver.tef import TEF6686
|
|
from driver.protocol import I2CPCClient # cpcc? so close (cccp?), hammer and sicle?
|
|
import os
|
|
from multiprocessing import Queue, Process, freeze_support
|
|
from queue import Empty
|
|
import libtimer
|
|
import asyncio
|
|
import websockets
|
|
from websockets import Headers, Response
|
|
import json
|
|
from cliserver_rdsDecoder import decodeRds
|
|
|
|
NAME = "CLI Server"
|
|
DESCRIPTION = ""
|
|
|
|
INITIAL_FREQ = 95.0
|
|
|
|
LATITUDE = 0
|
|
LONGITUDE = 0
|
|
|
|
PRESET0 = 87.5
|
|
PRESET1 = 87.5
|
|
PRESET2 = 87.5
|
|
PRESET3 = 87.5
|
|
|
|
# Mint - 1
|
|
# Cappucino - 2
|
|
# Nature - 3
|
|
# Ocean - 4
|
|
# Terminal - 5
|
|
# Nightlife - 6
|
|
# Blurple - 7
|
|
# Construction - 8
|
|
# Amoled - 9
|
|
# Dark Red - 10
|
|
# Ocean Blue - 11
|
|
DEFAULT_THEME = 1
|
|
|
|
BG_IMAGE = None
|
|
|
|
RBDS = False
|
|
RDS_TIMEOUT = 0
|
|
|
|
# "af":[99800,102600,90300],
|
|
# "txInfo":{"tx":"Radio Plus","pol":"V","erp":0.2,"city":"Pila","itu":"POL","dist":"3","azi":"238","id":6400902,"pi":"321C","reg":false,"otherMatches":[],"score":1},
|
|
|
|
def serialpart(inq: Queue, outq: Queue, rds: Queue):
|
|
p = I2CPCClient(os.getenv("PORT", "COM17"), int(os.getenv("BAUD", 115200)))
|
|
tef = TEF6686(p)
|
|
tef.init(clock=12_000_000)
|
|
tef.FM_Set_RDS(1)
|
|
tef.AUDIO_Set_Mute(False)
|
|
tef.FM_Set_ChannelEqualizer()
|
|
tef.FM_Set_MphSuppression()
|
|
tef.FM_Set_Stereo_Max(False)
|
|
tef.FM_Tune_To(TEF6686.TuneTo_Mode.Preset, int(INITIAL_FREQ * 100))
|
|
|
|
signal_timer = libtimer.Timer()
|
|
rds_timer = libtimer.Timer()
|
|
data_timer = libtimer.Timer()
|
|
|
|
rds_dropout_time = libtimer.Timer()
|
|
|
|
fm = True
|
|
|
|
wsdata = {
|
|
"pi": "?",
|
|
"freq": f"{INITIAL_FREQ:.3f}",
|
|
"sig": 11.25, # dBf
|
|
"sigRaw": "", # Raw signal from xrd proto
|
|
"sigTop": -1000,
|
|
"bw": "",
|
|
"st": False,
|
|
"stForced": False, # its forced for mono, lol
|
|
"rds": False,
|
|
"ps": "".ljust(8),
|
|
"tp": 0,
|
|
"ta": 0,
|
|
"ms": -1,
|
|
"pty": 0,
|
|
"ecc": None,
|
|
"af": [],
|
|
"rt0": "".ljust(64),
|
|
"rt1": "".ljust(64),
|
|
"rt_flag": 0,
|
|
"ims": 0,
|
|
"eq": 0,
|
|
"agc": "0", # static
|
|
"ant": "1", # static
|
|
"txInfo": {},
|
|
"country_name": "",
|
|
"country_iso": "UN",
|
|
"users": 0,
|
|
"ps_errors": "0,0,0,0,0,0,0,0",
|
|
"rt1_errors":"0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0",
|
|
"rt0_errors":"0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0"
|
|
}
|
|
|
|
def resetRds():
|
|
wsdata["rds"] = False
|
|
wsdata["pi"] = "?"
|
|
wsdata["ps"] = "".ljust(8)
|
|
wsdata["tp"] = 0
|
|
wsdata["ta"] = 0
|
|
wsdata["ms"] = -1
|
|
wsdata["pty"] = 0
|
|
wsdata["ecc"] = None
|
|
wsdata["af"] = []
|
|
wsdata["rt0"] = "".ljust(64)
|
|
wsdata["rt1"] = "".ljust(64)
|
|
wsdata["rt_flag"] = 0
|
|
wsdata["country_name"] = ""
|
|
wsdata["country_iso"] = "UN"
|
|
wsdata["ps_errors"] = "0,0,0,0,0,0,0,0"
|
|
wsdata["rt0_errors"] = "0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0"
|
|
wsdata["rt1_errors"] = "0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0"
|
|
rds.put("G:\r\nRESET-------\r\n\r\n")
|
|
|
|
while not p.ser.closed:
|
|
try:
|
|
msg = str(inq.get(False))
|
|
if msg == "!x":
|
|
tef.close()
|
|
break
|
|
elif msg == ".": raise Empty
|
|
elif msg == "+":
|
|
wsdata["users"] += 1
|
|
raise Empty
|
|
elif msg == "-":
|
|
wsdata["users"] -= 1
|
|
raise Empty
|
|
out = ""
|
|
if msg.startswith("M"):
|
|
mod = int(msg.removeprefix("M").strip(), 2)
|
|
if mod: fm = False
|
|
else: fm = True
|
|
out += f"M{mod}\n"
|
|
elif msg.startswith("T"):
|
|
freq = int(msg.removeprefix("T").strip()) // 10
|
|
if fm:
|
|
tef.FM_Tune_To(TEF6686.TuneTo_Mode.Preset, freq)
|
|
wsdata["freq"] = f"{(freq / 100):.3f}"
|
|
else:
|
|
tef.AM_Tune_To(TEF6686.TuneTo_Mode.Preset, freq)
|
|
wsdata["freq"] = f"{(freq / 1000):.3f}"
|
|
out += f"T{freq*10}\n"
|
|
wsdata["sigTop"] = -1000
|
|
resetRds()
|
|
elif msg.startswith("G"):
|
|
eqims = int(msg.removeprefix("G").strip(), 2)
|
|
tef.FM_Set_ChannelEqualizer((eqims & 1) == 1)
|
|
tef.FM_Set_MphSuppression((eqims & 2) == 2)
|
|
out += f"G{bin(eqims).removeprefix('0b').zfill(2)}\n"
|
|
wsdata["ims"] = (eqims & 1)
|
|
wsdata["eq"] = (eqims & 2) >> 1
|
|
elif msg.startswith("B"):
|
|
wsdata["stForced"] = bool(int(msg.removeprefix("B").strip(), 2))
|
|
tef.FM_Set_Stereo_Min(2 if wsdata["stForced"] else 0)
|
|
out += f"B{int(wsdata["stForced"])}\n"
|
|
elif msg.startswith("D"):
|
|
deemp = int(msg.removeprefix("D").strip())
|
|
dtime = 500 if deemp == 0 else (750 if deemp == 1 else 0)
|
|
tef.FM_Set_Deemphasis(dtime)
|
|
out += f"D{deemp}\n"
|
|
elif msg.startswith("W"):
|
|
bw = int(msg.removeprefix("W").strip())
|
|
auto = (bw == 0)
|
|
tef.FM_Set_Bandwidth(auto, 2360 if auto else (bw // 100))
|
|
out += f"W{bw}\n"
|
|
if out.strip(): outq.put(out)
|
|
except Empty:
|
|
if (RDS_TIMEOUT and RDS_TIMEOUT != 0) and rds_dropout_time.get_time() > RDS_TIMEOUT: resetRds()
|
|
|
|
if signal_timer.get_time() > 0.1:
|
|
signal_timer.reset()
|
|
res = None
|
|
if fm: res = tef.FM_Get_Quality_Data()
|
|
else: res = tef.AM_Get_Quality_Data()
|
|
|
|
if res:
|
|
if fm:
|
|
stereo_pilot, _ = d if (d := tef.FM_Get_Signal_Status()) else (None, None)
|
|
else: stereo_pilot = False
|
|
_, level, usn, wam, _, bandwidth, *_ = res
|
|
|
|
level /= 10
|
|
bandwidth /= 10
|
|
|
|
data = "S"
|
|
if wsdata["stForced"]: data += "M"
|
|
elif stereo_pilot: data += "s"
|
|
else: data += "m"
|
|
|
|
wsdata["st"] = stereo_pilot
|
|
|
|
level += 11.25 # to dBf from dBμV
|
|
|
|
data += f"{level},{wam//10},{usn//10},{bandwidth}"
|
|
wsdata["bw"] = str(bandwidth)
|
|
wsdata["sigRaw"] = data
|
|
wsdata["sig"] = level
|
|
wsdata["sigTop"] = wsdata["sigTop"] if level < wsdata["sigTop"] else level
|
|
if rds_timer.get_time() > 0.086 and fm:
|
|
rds_timer.reset()
|
|
|
|
res = tef.FM_Get_RDS_Data__decoder()
|
|
if res is None: continue
|
|
status, A, B, C, D, dec_error = res
|
|
dec_error >>= 8
|
|
|
|
if (status & (1 << 9) == 0) or (status & (1 << 15) == 0): continue
|
|
|
|
gdata = "G:\r\n"
|
|
a_error = dec_error >> 6
|
|
b_error = (dec_error >> 4) & 0b11
|
|
c_error = (dec_error >> 2) & 0b11
|
|
d_error = dec_error & 0b11
|
|
|
|
wsdata["rds"] = True
|
|
pi_error = decodeRds(wsdata, status, A, B, C, D, dec_error)
|
|
|
|
def threshold(group: str, error: int): return group if (error < 3) else "----"
|
|
|
|
if (status & (1 << 13) == 0):
|
|
if (status & (1 << 12) == 0):
|
|
# Type A, PI in A
|
|
gdata += threshold(f"{A:04X}", a_error)
|
|
gdata += threshold(f"{B:04X}", b_error)
|
|
gdata += threshold(f"{C:04X}", c_error)
|
|
elif (status >> 12) & 1:
|
|
# Type B, PI in A and C
|
|
pi = wsdata["pi"]
|
|
gdata += threshold(pi, pi_error)
|
|
gdata += threshold(f"{B:04X}", b_error)
|
|
gdata += threshold(pi, pi_error)
|
|
gdata += threshold(f"{D:04X}", d_error)
|
|
gdata += "\r\n\r\n"
|
|
|
|
rds.put(gdata)
|
|
rds_dropout_time.reset()
|
|
|
|
if data_timer.get_time() > 0.1:
|
|
data_timer.reset()
|
|
outq.put(json.dumps(wsdata))
|
|
|
|
inq = Queue()
|
|
outq = Queue()
|
|
rds = Queue()
|
|
|
|
wsc: list[websockets.ServerConnection] = []
|
|
rds_wsc: list[websockets.ServerConnection] = []
|
|
|
|
async def worker():
|
|
while True:
|
|
try:
|
|
msg = await asyncio.to_thread(outq.get, False)
|
|
if msg == "!x": break
|
|
for websocket in wsc[:]:
|
|
try: await websocket.send(msg)
|
|
except websockets.exceptions.ConnectionClosed: pass
|
|
except Empty: await asyncio.sleep(0.015)
|
|
async def rds_worker():
|
|
while True:
|
|
try:
|
|
msg = await asyncio.to_thread(rds.get, False)
|
|
if msg == "!x": break
|
|
for websocket in rds_wsc[:]:
|
|
try: await websocket.send(msg)
|
|
except websockets.exceptions.ConnectionClosed: pass
|
|
except Empty: await asyncio.sleep(0.015)
|
|
|
|
async def handle_text(websocket: websockets.ServerConnection):
|
|
wsc.append(websocket)
|
|
await asyncio.to_thread(inq.put, "+")
|
|
try:
|
|
async for message in websocket:
|
|
if message not in ("+", "-"):
|
|
print(message)
|
|
await asyncio.to_thread(inq.put, message)
|
|
await websocket.recv() # For loop exits normally if closed normally. This should raise an ConnectionClosed if the loop exited, because the connection was closed
|
|
except websockets.exceptions.ConnectionClosed:
|
|
wsc.remove(websocket)
|
|
await asyncio.to_thread(inq.put, "-")
|
|
print("Client disconnected")
|
|
|
|
async def handle_rds(websocket: websockets.ServerConnection):
|
|
rds_wsc.append(websocket)
|
|
await websocket.wait_closed()
|
|
rds_wsc.remove(websocket)
|
|
|
|
async def handler(websocket: websockets.ServerConnection):
|
|
if not websocket.request:
|
|
await websocket.close(1011)
|
|
return
|
|
print(f"Client connected: {websocket.remote_address} path: {websocket.request.path}")
|
|
if websocket.request.path == "/text": await handle_text(websocket)
|
|
elif websocket.request.path == "/rds": await handle_rds(websocket)
|
|
else: await websocket.close(1013)
|
|
|
|
async def handle_http(websocket: websockets.ServerConnection, reqeust: websockets.Request) -> Response | None:
|
|
tosend = None
|
|
mimetype = "text/text"
|
|
match reqeust.path:
|
|
case "/static_data":
|
|
json_data = {
|
|
"qthLatitude": f"{LATITUDE:.6f}",
|
|
"qthLongitude": f"{LONGITUDE:.6f}",
|
|
"presets": [
|
|
f"{PRESET0:.1f}",
|
|
f"{PRESET1:.1f}",
|
|
f"{PRESET2:.1f}",
|
|
f"{PRESET3:.1f}"
|
|
],
|
|
"defaultTheme": f"theme{DEFAULT_THEME}",
|
|
"bgImage": BG_IMAGE,
|
|
"rdsMode": RBDS,
|
|
"rdsTimeout": f"{RDS_TIMEOUT}",
|
|
"tunerName": NAME,
|
|
"tunerDesc": DESCRIPTION,
|
|
"ant": {
|
|
"enabled": False,
|
|
"ant1": {
|
|
"enabled": False,
|
|
"name": "Ant H"
|
|
},
|
|
"ant2": {
|
|
"enabled": False,
|
|
"name": "Ant V"
|
|
},
|
|
"ant3": {
|
|
"enabled": False,
|
|
"name": "Ant C"
|
|
},
|
|
"ant4": {
|
|
"enabled": False,
|
|
"name": "Ant D"
|
|
}
|
|
}
|
|
}
|
|
tosend = json.dumps(json_data).encode()
|
|
mimetype = "application/json"
|
|
case "/ping":
|
|
tosend = b"pong"
|
|
if tosend is not None:
|
|
return Response(200, "OK",
|
|
Headers([("Content-Type", mimetype), ("Content-Length", f"{len(tosend)}")]),
|
|
tosend)
|
|
return None
|
|
|
|
async def process_requests(websocket: websockets.ServerConnection, request: websockets.Request):
|
|
print(request)
|
|
if request.path not in ("/text", "/rds"):
|
|
handle = await handle_http(websocket, request)
|
|
if not handle:
|
|
data = b"This path does not have an handler"
|
|
return Response(
|
|
404,
|
|
"Not Found",
|
|
Headers([("Content-Type", "text/text"), ("Content-Length", f"{len(data)}")]),
|
|
data
|
|
)
|
|
else: return handle
|
|
if not "upgrade" in request.headers.get("Connection", "").lower():
|
|
return Response(
|
|
426,
|
|
"Upgrade Required",
|
|
Headers([("Connection", "Upgrade"), ("Upgrade", "websocket")]),
|
|
b"WebSocket upgrade required\n"
|
|
)
|
|
|
|
async def main():
|
|
wtask = asyncio.create_task(worker())
|
|
wtask2 = asyncio.create_task(rds_worker())
|
|
async with websockets.serve(handler, "localhost", 8765, process_request=process_requests, server_header="FM-DX-Cliserver"):
|
|
print("Server running on ws://localhost:8765")
|
|
await wtask
|
|
await wtask2
|
|
|
|
def start_process():
|
|
proc = Process(target=serialpart, args=(inq,outq,rds))
|
|
proc.start()
|
|
return proc
|
|
|
|
if __name__ == "__main__":
|
|
# I fucking hate windows, i wish they'd hang gates in a city center like they did mussolini
|
|
freeze_support()
|
|
proc = start_process()
|
|
try: asyncio.run(main())
|
|
except SystemExit:
|
|
inq.put("!x")
|
|
outq.put("!x")
|
|
rds.put("!x")
|
|
proc.terminate()
|
|
proc.join()
|
|
raise |