Initial push

serial_to_camstream.py

@@ -0,0 +1,161 @@
+import tkinter as tk
+import serial
+import serial.tools.list_ports
+import threading
+from PIL import Image, ImageTk
+from io import BytesIO
+import queue
+
+def connect() -> serial.Serial:
+    while True:
+        ports: list = serial.tools.list_ports.comports()
+        for i, port in enumerate(ports):
+            print(f"{i}: {port.device} - {port.description}")
+        
+        selected_port: str = input("Which port to use?\n")
+        if selected_port.__len__() < 1: continue
+        if(not "0123456789".__contains__(selected_port[0])): continue
+        
+        selected_port_idx: int = int(selected_port[0])
+        if selected_port_idx >= ports.__len__() or selected_port_idx < 0: continue
+        
+        s: serial.Serial = serial.Serial(port=ports[selected_port_idx].device, baudrate=921600, timeout=1)
+        try_counter: int = 0
+        while not s.is_open:
+            if try_counter == 10:
+                print("Failed to open port for 10 times -> stopping thread")
+                return
+            try:
+                s.open()
+            except:
+                print("Opening port failed!")
+                print(f"Try {try_counter}")
+                try_counter += 1
+                
+        print(f"Connected successfully! ({ports[selected_port_idx].device})")
+        try:
+            s.reset_input_buffer()
+        except Exception: pass
+        return s
+        
+
+def receiver_thread(s: serial.Serial, kill_thread_event: threading.Event, image_queue: queue.Queue) -> None:
+        buf: bytearray = bytearray()
+        delimiter: str = b"\x04\r\n"
+        
+        while s.is_open and not kill_thread_event.is_set():
+            try:
+                n = s.in_waiting
+            except Exception: 
+                n = 1
+            
+            chunk = s.read(n)
+            
+            if kill_thread_event.is_set(): return
+            if chunk: buf.extend(chunk)
+
+            while True:                
+                idx = buf.find(delimiter)
+                if idx == -1: break
+            
+                jpg: bytes = bytes(buf[:idx])
+                del buf[: idx + delimiter.__len__()]
+                if not jpg: continue
+                
+                try:
+                    image_queue.put_nowait(jpg)
+                except queue.Full:
+                    try:
+                        _ = image_queue.get_nowait()
+                    except queue.Empty: pass
+                    
+                    try:
+                        image_queue.put_nowait(jpg)
+                    except queue.Full: pass
+                
+            if buf.__len__() > 5000000: buf.clear()
+        
+    
+def detect_motion(last_image: bytes, new_image: bytes) -> bool:
+    length1: int = last_image.__len__()
+    length2: int = new_image.__len__()
+    length = length1
+    if length1 < length2:
+        new_image = new_image[:length1]
+    elif length2 < length1:
+        last_image = last_image[:length2]
+        length = length2
+    
+    def diff(pixel1, pixel2) -> int:
+        return pixel1 - pixel2
+    
+    sum: int = 0
+    for i in range(length):
+        sum += diff(last_image[i], new_image[i])
+    
+    norm = sum / length
+    ret = norm <= -1 or norm >= 1
+    # print(f"Norm: {norm}")
+    return ret
+
+def main():
+    kill_thread_event: threading.Event = threading.Event()
+    image_queue: queue.Queue = queue.Queue(maxsize=2)
+    
+    s: serial.Serial = connect()
+    
+    receiver = threading.Thread(target=receiver_thread, daemon=True, args=(s, kill_thread_event, image_queue))
+    receiver.start()
+        
+    root = tk.Tk()
+    root.title("ESP32-CAM Stream")
+    root.geometry("1280x720")
+    lbl = tk.Label(root)
+    lbl.pack(fill="both", expand=True)
+    
+    state = {"photo": None}
+    
+    last_image: bytes | None = None
+    
+    def poll_image():
+        nonlocal last_image
+        jpg = None
+        while True:
+            try:
+                jpg = image_queue.get_nowait()
+            except queue.Empty:
+                break
+
+        if jpg is not None:
+            if last_image is not None and detect_motion(last_image, jpg):
+                print("Motion detected!")
+            
+            last_image = jpg
+            
+            try:
+                img = Image.open(BytesIO(jpg)).convert("RGB")
+                
+                w = lbl.winfo_width()
+                h = lbl.winfo_height()
+                if w > 1 and h > 1:
+                    img = img.resize((w, h), Image.BILINEAR)
+                    
+                photo = ImageTk.PhotoImage(img)
+                state["photo"] = photo
+                lbl.configure(image=photo)
+            except Exception: pass
+            
+        root.after(10, poll_image)
+    
+    def on_close():
+        kill_thread_event.set()
+        root.destroy()
+        print("Closing!")
+
+    root.protocol("WM_DELETE_WINDOW", on_close)
+    poll_image()
+    root.mainloop()
+    s.close()
+
+if __name__ == '__main__':
+    main()