Merge remote-tracking branch 'dandroid/main'

2026-01-11 17:21:15 +00:00
parent fd9b558d4c 9457c717d4
commit 8e74d22219
1 changed files with 150 additions and 0 deletions
@@ -0,0 +1,150 @@
+#!/data/data/com.termux/files/usr/bin/python
+
+import termuxgui as tg
+import time
+import sys
+import socket
+import threading
+import select
+import datetime
+from pathlib import Path
+import os
+
+peer_hostname = None
+peer_thread = None
+bytes_recorded = 0
+
+started = False
+
+hosts = {
+    "eculogical.local" : None,
+    "eculogical2.local" : None,
+    "eculogical3.local" : None
+}
+
+def output_pathname():
+    root = Path(os.getenv("HOME"))
+    dir = root / "storage" / "capital-shore"
+    dir.mkdir(0o777, False, True)
+    name = datetime.datetime.utcfromtimestamp(int(time.time())).isoformat() + ".raw"
+    return dir / name
+
+def recording_thread_main():
+    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    sock.connect((peer_hostname, 5000))
+    sock.setblocking(0)
+    interval = 500
+    global bytes_recorded
+    bytes_recorded = 0
+
+    with open(output_pathname(), "wb") as outf:
+        while started:
+            print("sending subscribe")
+            b = bytes([interval >> 8, interval & 0xff, 0x54, 0x11, 0, 25] )
+            sock.send(b)
+            timeout = time.time() + 60
+            while started and time.time() < timeout:
+                ready = select.select([sock], [], [], timeout - time.time())
+                if ready[0]:
+                    data = sock.recv(4000)
+                    bytes_recorded = bytes_recorded + len(data)
+                    outf.write(data)
+                    outf.flush()
+
+    sock.close()
+    print("recording thread terminated")
+
+
+
+
+    # Intercept the back button. Instead of finishing the Activity, it now sends an event instead
+    # a.interceptbackbutton(True)
+    # Also possible: a = tg.Activity(c, intercept=True)
+
+
+with tg.Connection() as c:
+    # create a new Activity. By default, a new Task as created.
+    a = tg.Activity(c)
+    # you can find the Task under a.t
+
+    def add_host(activity, name, parent):
+        selected = (peer_hostname == name)
+        if selected:
+            label = f"* {name}"
+        else:
+            label = name
+        button = tg.Button(activity, label, parent)
+        button.setlinearlayoutparams(0)
+        button.setheight(tg.View.WRAP_CONTENT)
+        hosts[name] = button
+
+    def choose_hostname(hostname):
+        global peer_hostname
+        if hostname == None:
+            started = False
+        peer_hostname = hostname
+        refresh_buttons()
+
+    def refresh_buttons():
+        for name, widget in hosts.items():
+            if name == peer_hostname:
+                label = f"* {name}"
+            else:
+                label = name
+            widget.settext(label)
+        if peer_hostname == None:
+            start_button.settext("----")
+        else:
+            start_button.settext("STOP" if started else "START")
+
+
+    def start_recording(start_or_stop):
+        global started
+        if start_or_stop and peer_hostname != None:
+            started = True
+            print("starting record")
+            peer_thread = threading.Thread(target=recording_thread_main, daemon = True)
+            peer_thread.start()
+        else:
+            print("stopping record")
+            started = False
+
+    instances = tg.LinearLayout(a)
+
+    add_host(a, "eculogical.local", instances)
+    add_host(a, "eculogical2.local", instances)
+    add_host(a, "eculogical3.local", instances)
+
+    tg.Space(a, instances)
+
+    status = tg.TextView(a, "initializing", instances)
+
+    tg.Space(a, instances)
+
+    buttons = tg.LinearLayout(a, instances, False)
+
+    start_button = tg.Button(a, "(start)", buttons)
+    start_button.setlinearlayoutparams(0)
+    start_button.setheight(tg.View.WRAP_CONTENT)
+    # start_button.setwidth(tg.View.WRAP_CONTENT)
+    start_button.setgravity(1, 1)
+
+    def update_byte_counter():
+        global started
+        while True:
+            label = f"recorded {bytes_recorded} bytes" if started else "not recording"
+            status.settext(label)
+            time.sleep(1)
+
+    threading.Thread(target = update_byte_counter, daemon = True).start()
+
+    for ev in c.events():  # waits for events from the gui
+        if ev.type == tg.Event.destroy:
+            sys.exit()
+        if ev.type == tg.Event.click:
+            for k, v in hosts.items():
+                if v == ev.value["id"]:
+                    choose_hostname(k)
+            if ev.value["id"] == start_button:
+                start_recording(not started)
+                refresh_buttons()