tza/tunerlistd
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

more final fixes

Browse Source
This commit is contained in:
marios8543 2025-05-19 17:37:46 +03:00
parent 0422d3f57a
commit 1b26a88451
3 changed files with 13 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
tunerlistd/buttondecoder.py
View File

@ -35,12 +35,12 @@ class ButtonEvent:
return return
thread_sleep(0.001) thread_sleep(0.001)
self.long_press = True self.long_press = True
def __repr__(self): def __repr__(self):
return str(self) return str(self)
def __str__(self): def __str__(self):
return f"{self.button} {self.state}" return self.button + " LONG PRESS" if self.long_press else ""
class ButtonDecoder: class ButtonDecoder:
def __init__(self, pi): def __init__(self, pi):
@ -75,10 +75,11 @@ class ButtonDecoder:
btnev = ButtonEvent(btn) btnev = ButtonEvent(btn)
self.pending_buttons[line[0]][line[1]] = btnev self.pending_buttons[line[0]][line[1]] = btnev
future = self.executor.submit(btnev.wait_for_longpress) future = self.executor.submit(btnev.wait_for_longpress)
future.add_done_callback(lambda: self.queue.put_nowait(btn)) future.add_done_callback(lambda _: self.queue.put_nowait(btnev))
else: else:
btnev = self.pending_buttons[line[0]][line[1]] btnev = self.pending_buttons[line[0]][line[1]]
btnev.event.set() if btnev:
btnev.event.set()
except Exception as e: except Exception as e:
print(e) print(e)

10
tunerlistd/main.py
View File

@ -113,7 +113,7 @@ async def button_listener():
btn = button.button if not button.long_press else f"{button.button}_LP" btn = button.button if not button.long_press else f"{button.button}_LP"
if btn in BUTTON_MAPPINGS: if btn in BUTTON_MAPPINGS:
try: try:
func, arg = BUTTON_MAPPINGS[button.button] func, arg = BUTTON_MAPPINGS[btn]
func(arg) func(arg)
except Exception as e: except Exception as e:
print(e) print(e)
@ -165,7 +165,7 @@ async def websocket_endpoint(websocket: WebSocket):
try: try:
while True: while True:
payload = await queue.get() payload = await queue.get()
print(payload) # print(payload)
await websocket.send_text(dumps(payload)) await websocket.send_text(dumps(payload))
except CancelledError: except CancelledError:
return return
@ -186,11 +186,5 @@ async def connect_aa(request: Request):
queue.put_nowait({"wireless_aa_ip": ip}) queue.put_nowait({"wireless_aa_ip": ip})
return ip return ip
@app.get("/aabtn", response_class=PlainTextResponse)
async def aabtn(request: Request):
btn = request.query_params.get("btn")
queue.put_nowait({"button": {"btn": btn, "state": 2}})
return btn
if __name__ == "__main__": if __name__ == "__main__":
run(app=app, host="0.0.0.0", port=5959, log_level="info", ) run(app=app, host="0.0.0.0", port=5959, log_level="info", )

12
tunerlistd/tunerlist.py
View File

@ -4,7 +4,7 @@ from time import sleep as thread_sleep
import re import re
BAUD_RATE = 9600 BAUD_RATE = 9600
PORT = "/dev/ttyAMA2" PORT = "/dev/ttyUSB0"
DEBUG = True DEBUG = True
fadePattern = re.compile(r"\d F \d") fadePattern = re.compile(r"\d F \d")
@ -91,11 +91,9 @@ class TunerList:
except Exception as e: except Exception as e:
print(e) print(e)
txt = "" txt = ""
for i in range(7, 11): start = (6 if line[0] == 0x90 else 10)
try: for i in line[start: start+4]:
txt += line[i].decode("ascii") txt += hex(i)[2:]
except:
txt += hex(line[i])[2:]
txt += " " txt += " "
self.text = txt self.text = txt
finally: finally:
@ -146,4 +144,4 @@ if __name__ == "__main__":
except KeyError: except KeyError:
pass pass
get_event_loop().run_until_complete(main()) get_event_loop().run_until_complete(main())