implement pigpio software serial

arduino/tunerlist_i2c_attiny85/tunerlist_i2c_attiny85.ino

@@ -88,7 +88,7 @@ void onReceive(int bytes)
 
 void setup()
 {
-  serial.begin(9600);
+  serial.begin(4800);
 
   Wire.setClock(10000);
   Wire.begin(0x23);

tunerlistd/tunerlist.py

@@ -1,7 +1,10 @@
-import serial
+from pigpip import pi
-from asyncio import Queue, get_event_loop
+from asyncio import Queue, get_event_loop, sleep
 from concurrent.futures import ThreadPoolExecutor
 
+GPIO_PIN = 27
+BAUD_RATE = 4800
+
 BUTTONS = {
     "ok": [0x00, 0x00],
     "ok_hold": [0x00, 0x40],
@@ -18,6 +21,17 @@ BUTTONS = {
     "wheel_down": [0x01, 0x41],
 }
 
+async def serial_wait_for_line(pi):
+    buffer = b''
+    while True:
+        (count, data) = pi.bb_serial_read(GPIO_PIN)
+        if count > 0:
+            buffer += data
+            while b'\n' in buffer:
+                line, buffer = buffer.split(b'\n', 1)
+                return line
+        await sleep(0)
+
 class TunerListState:
     def __init__(self, text, preset):
         self.text = text
@@ -45,16 +59,17 @@ class TunerListState:
         return self.__str__()
 
 class TunerList:
-    def __init__(self, port):
+    def __init__(self):
-        self.serial = serial.Serial()# serial.Serial(port)
+        self.pi = pi()
+        self.pi.bb_serial_read_open(GPIO_PIN, BAUD_RATE, 8)
         self.queue = Queue()
 
         self.text = ""
         self.preset = None
 
-    def _run(self):
+    async def run(self):
         while True:
-            line = self.serial.readline()[:-2]
+            line = await serial_wait_for_line(self.pi)
             if line[0] == 0x0F or line[0] == 0x0C:
                 self.text = line[(8 if line[0] == 0x0F else 5):].decode("ascii")
                 if (line[3] != 0x20):
@@ -71,11 +86,6 @@ class TunerList:
             #    print(hex(c), " ", end="")
             #print("[",self.text,"]", f" [P {self.preset}]")
     
-    async def run(self):
-        with ThreadPoolExecutor() as executor:
-            loop = get_event_loop()
-            await loop.run_in_executor(executor, self._run)
-    
     async def yield_new_state(self):
         while True:
             yield await self.queue.get()