bluetooth.py

import socket
import json
import threading


class BluetoothManager:
    def __init__(self, mac="20:25:05:00:CB:AD", port=1):
        self.HC05_MAC = mac
        self.PORT = port
        self.sock = self.connect()

        self._lock = threading.Lock()
        self._latest = None
        self._running = True
        self._thread = threading.Thread(target=self._reader_loop, daemon=True)
        self._thread.start()

    def connect(self):
        sock = socket.socket(socket.AF_BLUETOOTH, socket.SOCK_STREAM, socket.BTPROTO_RFCOMM)
        sock.connect((self.HC05_MAC, self.PORT))
        sock.settimeout(5)
        return sock

    def read_line(self):
        data = b""
        while True:
            char = self.sock.recv(1)
            if not char:
                break
            data += char
            if char == b'\n':
                break
        return data.decode('utf-8', errors='ignore').strip()

    def _reader_loop(self):
        while self._running:
            try:
                line = self.read_line()
                if not line:
                    continue
                data = json.loads(line)
                distance = data.get("d")
                with self._lock:
                    self._latest = distance
            except socket.timeout:
                continue
            except json.JSONDecodeError:
                continue
            except Exception as e:
                print("[err] >>> ", e)
                break

    @property
    def latest_distance(self):
        with self._lock:
            return self._latest

    def close(self):
        self._running = False
        try:
            self.sock.close()
        except Exception:
            pass


if __name__ == "__main__":
    import time
    bt = BluetoothManager()
    print("[msg] >>> connected, polling latest distance...")
    try:
        while True:
            d = bt.latest_distance
            print(f"Distance: {d} cm" if d is not None else "Out of range")
            time.sleep(0.2)
    except KeyboardInterrupt:
        bt.close()
        print("\nDisconnected.")