import asyncio import threading import time import redis import pickle import os from datetime import datetime, timezone from zoneinfo import ZoneInfo # Python 3.9+ from bleak import BleakScanner, BleakClient from typing import Set, Iterable from config import * from sensors import * from ble import * from ble_redis import * # In-memory cache + thread-safe lock _mac_cache: Set[str] = set() _lock = threading.Lock() _sensors_cache = [] # ---------------------------------------------------------------------- # Pickle persistence helpers # ---------------------------------------------------------------------- STATE_FILE = "scanner_state.pkl" def save_state() -> None: """ Persist the MAC cache and sensor objects to disk. Called automatically after each cache refresh and at program exit. """ with _lock: state = { "mac_cache": _mac_cache, "sensors": _sensors_cache, } try: with open(STATE_FILE, "wb") as f: pickle.dump(state, f, protocol=pickle.HIGHEST_PROTOCOL) print("[Scanner] State persisted to", STATE_FILE) except Exception as exc: # Don't crash the app – just log the error print("[Scanner] Warning: could not persist state:", exc) def load_state() -> None: """ Load the MAC cache and sensor objects from disk if the file exists. Called automatically on import. """ if not os.path.exists(STATE_FILE): return try: with open(STATE_FILE, "rb") as f: state = pickle.load(f) with _lock: # NOTE: we replace the lists/sets to avoid mutating old references _mac_cache.clear() _mac_cache.update(state.get("mac_cache", [])) global _sensors_cache _sensors_cache = state.get("sensors", []) print("[Scanner] Loaded state from", STATE_FILE) except Exception as exc: print("[Scanner] Warning: could not load state:", exc) # Load any persisted state on module import load_state() # Redis Functions r = redis.Redis(host=bind_ip, port=redis_port) def update_redis(data): # push data to redis # Publish to the Redis channel that the WS server is listening on r.publish('temp_hum', json.dumps(data)) if debug_output: print("=== Stats Redis Update ===") print(json.dumps(data, indent=2)) return True # this function filters MACs def _is_match(mac: str, name: str) -> bool: """Return True if the MAC starts with the configured prefix and is correct model.""" if name == "LYWSD03MMC": return mac.upper().startswith(MAC_PREFIX.upper()) # this function scans for MACs async def _scan_once() -> Iterable[str]: """Do one scan cycle and return the matching MACs.""" devices = await BleakScanner.discover(timeout=5.0) # 5s scan return [dev.address for dev in devices if _is_match(dev.address, dev.name)] # this very important function updates the MAC cache and the sensor objects def update_cache() -> None: """ Called by the scheduler (in a separate thread). It runs a tiny asyncio event loop to call the Bleak scan, then updates the cache. """ # Run the async scan in a fresh loop; Bleak requires this. # Gets the MACs print("--- Starting update_cache loop ---") new_macs = scan_macs() # now that we have MACs, let's get sensor data print("Updating sensor data...") sensor_data = None try: sensor_data = sensor_data_query(new_macs) # print(sensor_data) except Exception as e: print(f"Aww fuck, no data :( - {e}") # now we update the data in the sensor objects print("Sensor data collected:") for sensor_datum in sensor_data: print(f"MAC - {sensor_datum['name']} | temp - {sensor_datum['temp']} | humidity - {sensor_datum['humidity']}") for sensor in _sensors_cache: if sensor.name == sensor_datum['name']: sensor.add_or_update(key="temp", value=sensor_datum['temp']) sensor.add_or_update(key="humidity", value=sensor_datum['humidity']) # if redis is enabled, push full details to the redis server if enable_redis: update_redis(get_sensor_details()) save_state() # this helper function scans for MACs def scan_macs(): print("Scanning for MACs:") new_macs = asyncio.run(_scan_once()) with _lock: for new_mac in new_macs: print(f"{new_mac}") new_sensor = Sensor(name=f"{new_mac}") if new_mac not in set(_mac_cache): _sensors_cache.append(new_sensor) _mac_cache.update(new_macs) print(f"Found {len(new_macs)} new MACs, total {len(_mac_cache)} MACs in cache.") return new_macs # return just the MACs def get_cache_snapshot() -> Set[str]: """Return a thread-safe snapshot of the current cache.""" with _lock: return set(_mac_cache) # return sensor data and timestamps, can return int or string def get_sensor_details(friendly_date = False): sensor_list = [] for sensor in _sensors_cache: first_seen = sensor.first_seen last_seen = sensor.last_seen if friendly_date: first_seen = ts_to_pst_string(sensor.first_seen) last_seen = ts_to_pst_string(sensor.last_seen) # print(f"Summary - {sensor.name}") new_sensor = { "name": sensor.name, "first_seen": first_seen, "last_seen": last_seen, "comment": sensor.comment, "data": sensor.to_dict() } sensor_list.append(new_sensor) result = { "sensors": sensor_list } return result # return just sensor data without timestamps def get_sensor_summary(): sensor_list = [] for sensor in _sensors_cache: # print(f"Summary - {sensor.name}") new_sensor = { "name": sensor.name, "comment": sensor.comment, "data": sensor.to_dict(simple_return=True) } sensor_list.append(new_sensor) result = { "sensors": sensor_list } return result # helper function for ts int to string def ts_to_pst_string(ts_str): utc_dt = datetime.fromtimestamp(ts_str, tz=timezone.utc) pacific_dt = utc_dt.astimezone(ZoneInfo("America/Los_Angeles")) return pacific_dt.strftime("%Y-%m-%d %H:%M:%S")