# SmartSpace Systems: occupancy-aware lighting controller (MicroPython, Raspberry Pi Pico W) # First-year design project, UVic. Wi-Fi credentials removed; set your own below. # Pins: PIR GP14, break beams GP15 (entry side) and GP16 (exit side), light sensor ADC GP27, LED GP17. import network import machine import time import uasyncio as asyncio # ========================================== # CONFIGURATION # ========================================== WIFI_SSID = 'YOUR_NETWORK_NAME' WIFI_PASSWORD = 'YOUR_PASSWORD' # Light threshold: 0-65535 range for Pico ADC. # With the flipped voltage divider, the value goes UP when it gets dark. # It rests around 7k in light, and jumps to 24k when covered. LIGHT_THRESHOLD = 15000 # ========================================== # HARDWARE PINOUT & SETUP # ========================================== # PIR Sensor on GP14 pir_sensor = machine.Pin(14, machine.Pin.IN) # Break Beams on GP15 and GP16 (internal pull-ups enabled) # Depending on the sensor, "broken" might mean the pin reads HIGH or LOW. # Assuming standard IR break beams: unbroken = pulled LOW by sensor normally, # or if using pull-ups: unbroken = HIGH, broken (beam blocked) = LOW. beam1_entry = machine.Pin(15, machine.Pin.IN, machine.Pin.PULL_UP) beam2_exit = machine.Pin(16, machine.Pin.IN, machine.Pin.PULL_UP) # Photoresistor on ADC0 (GP26) ldr_sensor = machine.ADC(27) # LED (Room Lights) on GP17 room_led = machine.Pin(17, machine.Pin.OUT) room_led.value(0) # Ensure off at startup # ========================================== # GLOBAL STATE VARIABLES # ========================================== occupancy_count = 0 motion_detected = False light_is_on = False auto_mode = True # Settings for independent modes light_independent = False motion_independent = False breakbeam_independent = False # Variables for processing the directional break-beam logic # 0 = Idle, 1 = Beam 1 triggered first, 2 = Beam 2 triggered first, 3 = Trigger complete, waiting for clear beam_state = 0 last_beam_time = time.ticks_ms() BEAM_TIMEOUT_MS = 3000 # Reset state if sequence takes longer than 3 seconds # ========================================== # CORE LOGIC: SENSOR FUSION & ACTUATION # ========================================== def process_directional_counting(): """ Non-blocking directional counting logic using a state machine. By reading the pin state continuously without sleeping, we ensure we don't miss fast triggers. Sequence for Entry (Count Increment): Idle -> Beam 1 breaks -> Beam 2 breaks sequence -> Both clear -> Idle Sequence for Exit (Count Decrement): Idle -> Beam 2 breaks -> Beam 1 breaks sequence -> Both clear -> Idle """ global occupancy_count, beam_state, last_beam_time # Read raw pin values # (Assuming pull-up: value is 0 when the output is pulled down/broken by the sensor) # Adjust `not` condition based on your exact break-beam modules. b1_broken = not beam1_entry.value() b2_broken = not beam2_exit.value() current_time = time.ticks_ms() if beam_state == 0: # IDLE: Waiting for a beam to break if b1_broken and b2_broken: # Both broke simultaneously (or too fast to poll individually) beam_state = 4 last_beam_time = current_time elif b1_broken: # Person is walking IN, tripped outer beam first beam_state = 1 last_beam_time = current_time elif b2_broken: # Person is walking OUT, tripped inner beam first beam_state = 2 last_beam_time = current_time elif beam_state == 1: # PENDING ENTRY: Beam 1 was broken first. if b2_broken: occupancy_count += 1 beam_state = 3 # Move to waiting for clear last_beam_time = current_time # Reset if it's been over 5s (they turned around and left before hitting beam 2) elif time.ticks_diff(current_time, last_beam_time) > 5000: beam_state = 0 elif beam_state == 2: # PENDING EXIT: Beam 2 was broken first. if b1_broken: occupancy_count = max(0, occupancy_count - 1) # Prevent negative count beam_state = 3 # Move to waiting for clear last_beam_time = current_time # Reset if it's been over 5s elif time.ticks_diff(current_time, last_beam_time) > 5000: beam_state = 0 elif beam_state == 3: # WAIT FOR CLEAR: Both beams must be unblocked before checking for the next person if not b1_broken and not b2_broken: # Ensure it stays clear for 500ms before accepting next trigger (debounce) if time.ticks_diff(current_time, last_beam_time) > 500: beam_state = 0 else: # If any beam is still broken, reset the clear timer last_beam_time = current_time elif beam_state == 4: # SIMULTANEOUS BREAK: Both broke at the same time. Wait to see which one clears first! if not b1_broken and b2_broken: # Beam 1 cleared first, meaning they are moving IN past the door occupancy_count += 1 beam_state = 3 last_beam_time = current_time elif b1_broken and not b2_broken: # Beam 2 cleared first, meaning they are moving OUT past the door occupancy_count = max(0, occupancy_count - 1) beam_state = 3 last_beam_time = current_time elif not b1_broken and not b2_broken: beam_state = 0 elif time.ticks_diff(current_time, last_beam_time) > 5000: beam_state = 0 async def update_sensors_task(): """ Main sensor polling loop using uasyncio to remain non-blocking. Runs constantly to process beam states rapidly. """ global motion_detected, light_is_on, auto_mode, light_independent, motion_independent, breakbeam_independent while True: # 1. Update break beams & count process_directional_counting() # 2. Update occupancy validation (PIR OR Count) motion_detected = bool(pir_sensor.value()) eff_occupancy = (occupancy_count > 0) and not breakbeam_independent eff_motion = motion_detected and not motion_independent is_occupied = eff_occupancy or eff_motion # 3. Read Analog Light Level light_level = ldr_sensor.read_u16() is_dark_enough = (light_level > LIGHT_THRESHOLD) or light_independent # 4. Actuation Logic: Occupied + Low Ambient Light (or manual override) if not auto_mode: room_led.value(1) light_is_on = True elif is_occupied and is_dark_enough: room_led.value(1) light_is_on = True else: room_led.value(0) light_is_on = False # Yield to event loop to allow socket server to respond promptly. # 20ms delay is fast enough for human movement through a break beam. await asyncio.sleep_ms(20) # ========================================== # WEB SERVER & UI # ========================================== def render_html(): """ Returns a minimal HTML string for the dashboard to prevent MemoryError. Uses embedded CSS and minimal tags. Auto-reloads using JS. """ return f""" SmartSpace Controller

SmartSpace

Mode

{'Auto (Sensors)' if auto_mode else 'Always ON'}

Occupancy Count

{occupancy_count}

Motion Detected?

{'YES' if motion_detected else 'NO'}

Lights Status

{'ON' if light_is_on else 'OFF'}
Settings
Light Independent
Motion Independent
Breakbeam Independent
""" async def serve_client(reader, writer): """ Handle incoming web client requests """ global auto_mode, light_independent, motion_independent, breakbeam_independent try: # Read the HTTP request (non-blocking) request = await reader.read(1024) if not request: return request_str = request.decode('utf-8') # Check for mode toggle route if 'GET /set_mode?auto=1' in request_str: auto_mode = True elif 'GET /set_mode?auto=0' in request_str: auto_mode = False elif 'GET /set_config?light=1' in request_str: light_independent = True elif 'GET /set_config?light=0' in request_str: light_independent = False elif 'GET /set_config?motion=1' in request_str: motion_independent = True elif 'GET /set_config?motion=0' in request_str: motion_independent = False elif 'GET /set_config?breakbeam=1' in request_str: breakbeam_independent = True elif 'GET /set_config?breakbeam=0' in request_str: breakbeam_independent = False # Give a lightweight HTTP response headers + concise HTML body response = render_html() header = "HTTP/1.1 200 OK\r\nContent-Type: text/html\r\nConnection: close\r\n\r\n" await writer.awrite(header) await writer.awrite(response) except Exception as e: print("Client error:", e) finally: # Always close connection to free socket resources await writer.aclose() async def start_server(): """ Start the uasyncio web socket server """ print("Starting web server...") # Port 80 server = await asyncio.start_server(serve_client, "0.0.0.0", 80) print("Web server running!") # Keep server task alive indefinitely while True: await asyncio.sleep(3600) # ========================================== # MAIN BOOTSTRAP & WIFI # ========================================== def connect_wifi(): """ Setup wireless station and connect to network """ wlan = network.WLAN(network.STA_IF) wlan.active(True) wlan.connect(WIFI_SSID, WIFI_PASSWORD) print("Connecting to network...") # Time ticks check to stay non-blocking conceptually, though we block boot sequence start_time = time.ticks_ms() while not wlan.isconnected(): if time.ticks_diff(time.ticks_ms(), start_time) > 15000: print("WiFi connection timeout!") return False time.sleep_ms(500) # Blocking sleep is acceptable ONLY during initial setup ip = wlan.ifconfig()[0] # Wait half a second before printing to ensure the terminal has caught up time.sleep_ms(500) print("\n" + "="*40) print(f"SUCCESS! Connected to WiFi: {WIFI_SSID}") print(f"Dashboard IP Addr: {ip}") print("="*40 + "\n") return True async def main(): print("Initializing SmartSpace Controller...") # Wait for WiFi to connect before spinning up webserver connect_wifi() # Create the concurrent task for polling sensors rapidly asyncio.create_task(update_sensors_task()) # Start the async web server, which also blocks logic progression (we want it to run indefinitely) await start_server() # Run the event loop if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: print("Script stopped by user.") # Fail-safe cleanup room_led.value(0)