ti_mo
162 lines
4.8 KiB
Python
162 lines
4.8 KiB
Python
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()
|