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