timo/signal-light-api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit

Browse Source
This commit is contained in:
timo
2026-07-25 20:47:48 +02:00
commit 58be310d63
4 changed files with 910 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main.py
+500
View File
@@ -0,0 +1,500 @@
from contextlib import asynccontextmanager
from threading import Event, Lock, Thread
from typing import Dict, Optional
from time import sleep
from fastapi import FastAPI
from gpiozero import OutputDevice
# BCM GPIO pin numbers
RELAY_PINS = {
1: 26, # Green
2: 19, # Yellow
3: 13, # Red
4: 6, # Alert
}
# Most relay HATs are active-low.
# Change this to True if your relay board is active-high.
RELAY_ACTIVE_HIGH = False
# Alert: one second on, one second off.
ALERT_INTERVAL_SECONDS = 1.0
# Demo: duration of each color in the sequence.
DEMO_INTERVAL_SECONDS = 0.25
# Green -> Yellow -> Red -> Yellow -> repeat
DEMO_SEQUENCE = (
(1, "green"),
(2, "yellow"),
(3, "red"),
(2, "yellow"),
)
relays: Dict[int, OutputDevice] = {}
# Protects access to the three color relays.
color_lock = Lock()
# Protects alert thread state.
alert_lock = Lock()
# Protects demo thread state.
demo_lock = Lock()
active_color: Optional[str] = None
alert_stop_event = Event()
alert_thread: Optional[Thread] = None
demo_stop_event = Event()
demo_thread: Optional[Thread] = None
def reset_relays() -> None:
"""
Reset the entire relay system.
- Stops demo mode
- Stops alert mode
- Turns every relay off
"""
stop_demo(turn_colors_off=False)
stop_alert()
with color_lock:
relays[1].off()
relays[2].off()
relays[3].off()
global active_color
active_color = None
if 4 in relays:
relays[4].off()
def turn_colors_off_locked() -> None:
"""
Turn off green, yellow, and red.
color_lock must already be held.
Alert is not affected.
"""
global active_color
relays[1].off()
relays[2].off()
relays[3].off()
active_color = None
def set_color_locked(relay_number: int, color_name: str) -> None:
"""
Activate exactly one color relay.
color_lock must already be held.
Alert is not affected.
"""
global active_color
turn_colors_off_locked()
relays[relay_number].on()
active_color = color_name
# ---------------------------------------------------------------------------
# Alert mode
# ---------------------------------------------------------------------------
def alert_worker() -> None:
"""
Toggle the alert relay until alert_stop_event is set.
"""
alert_relay = relays[4]
try:
while not alert_stop_event.is_set():
alert_relay.on()
if alert_stop_event.wait(ALERT_INTERVAL_SECONDS):
break
alert_relay.off()
if alert_stop_event.wait(ALERT_INTERVAL_SECONDS):
break
finally:
alert_relay.off()
def start_alert() -> None:
"""
Start alert blinking.
Color and demo modes are not affected.
Repeated calls do not create additional threads.
"""
global alert_thread
with alert_lock:
if alert_thread is not None and alert_thread.is_alive():
return
alert_stop_event.clear()
alert_thread = Thread(
target=alert_worker,
name="relay-alert-thread",
daemon=True,
)
alert_thread.start()
def stop_alert() -> None:
"""
Stop alert blinking.
Color and demo modes are not affected.
"""
global alert_thread
with alert_lock:
alert_stop_event.set()
thread = alert_thread
alert_thread = None
# Do not hold alert_lock while waiting for the worker.
if thread is not None and thread.is_alive():
thread.join()
if 4 in relays:
relays[4].off()
def is_alert_running() -> bool:
with alert_lock:
return alert_thread is not None and alert_thread.is_alive()
# ---------------------------------------------------------------------------
# Demo mode
# ---------------------------------------------------------------------------
def demo_worker() -> None:
"""
Continuously cycle through:
green -> yellow -> red -> yellow
"""
try:
while not demo_stop_event.is_set():
for relay_number, color_name in DEMO_SEQUENCE:
if demo_stop_event.is_set():
break
with color_lock:
set_color_locked(relay_number, color_name)
if demo_stop_event.wait(DEMO_INTERVAL_SECONDS):
break
finally:
# Leave all color relays off when demo mode ends.
with color_lock:
turn_colors_off_locked()
def start_demo() -> None:
"""
Start demo mode.
Alert mode is not affected.
Repeated calls do not create additional demo threads.
"""
global demo_thread
with demo_lock:
if demo_thread is not None and demo_thread.is_alive():
return
demo_stop_event.clear()
demo_thread = Thread(
target=demo_worker,
name="relay-demo-thread",
daemon=True,
)
demo_thread.start()
def stop_demo(turn_colors_off: bool = True) -> None:
"""
Stop demo mode.
Alert mode is not affected.
When turn_colors_off is True, all three color relays are turned off.
"""
global demo_thread
with demo_lock:
demo_stop_event.set()
thread = demo_thread
demo_thread = None
# Do not hold demo_lock while waiting for the worker.
if thread is not None and thread.is_alive():
thread.join()
if turn_colors_off and relays:
with color_lock:
turn_colors_off_locked()
def is_demo_running() -> bool:
with demo_lock:
return demo_thread is not None and demo_thread.is_alive()
# ---------------------------------------------------------------------------
# Manual color mode
# ---------------------------------------------------------------------------
def activate_manual_color(relay_number: int, color_name: str) -> None:
"""
Stop demo mode and activate a manually selected color.
Alert mode is not affected.
"""
stop_demo(turn_colors_off=True)
with color_lock:
set_color_locked(relay_number, color_name)
def turn_everything_off() -> None:
"""
Stop demo and alert modes and turn off every relay.
"""
stop_demo(turn_colors_off=True)
stop_alert()
if relays:
with color_lock:
turn_colors_off_locked()
# ---------------------------------------------------------------------------
# FastAPI lifecycle
# ---------------------------------------------------------------------------
@asynccontextmanager
async def lifespan(app: FastAPI):
try:
# Initialize GPIOs
for relay_number, gpio_pin in RELAY_PINS.items():
relays[relay_number] = OutputDevice(
pin=gpio_pin,
active_high=RELAY_ACTIVE_HIGH,
initial_value=False,
)
# Give the GPIO driver a moment to initialize
sleep(0.5)
# Force every relay into the OFF state
reset_relays()
yield
finally:
reset_relays()
for relay in relays.values():
relay.close()
relays.clear()
app = FastAPI(
title="Raspberry Pi Signal Light API",
description=(
"Controls mutually exclusive green, yellow, and red relays. "
"Alert blinking operates independently. "
"Demo mode cycles through green, yellow, red, and yellow."
),
version="3.0.0",
lifespan=lifespan,
)
# ---------------------------------------------------------------------------
# API endpoints
# ---------------------------------------------------------------------------
@app.get("/")
def root():
return {
"service": "Raspberry Pi Signal Light API",
"active_color": active_color,
"alert_active": is_alert_running(),
"demo_active": is_demo_running(),
"endpoints": {
"green": "POST /green",
"yellow": "POST /yellow",
"red": "POST /red",
"start_alert": "POST /alert",
"stop_alert": "POST /alert/off",
"start_demo": "POST /demo",
"stop_demo": "POST /demo/off",
"all_off": "POST /off",
"status": "GET /status",
},
}
@app.post("/green")
def green():
activate_manual_color(
relay_number=1,
color_name="green",
)
return {
"active_color": "green",
"demo_active": False,
"alert_active": is_alert_running(),
}
@app.post("/yellow")
def yellow():
activate_manual_color(
relay_number=2,
color_name="yellow",
)
return {
"active_color": "yellow",
"demo_active": False,
"alert_active": is_alert_running(),
}
@app.post("/red")
def red():
activate_manual_color(
relay_number=3,
color_name="red",
)
return {
"active_color": "red",
"demo_active": False,
"alert_active": is_alert_running(),
}
@app.post("/alert")
def alert():
start_alert()
return {
"active_color": active_color,
"alert_active": True,
"demo_active": is_demo_running(),
"mode": "blinking",
"on_seconds": ALERT_INTERVAL_SECONDS,
"off_seconds": ALERT_INTERVAL_SECONDS,
}
@app.post("/alert/off")
def alert_off():
stop_alert()
return {
"active_color": active_color,
"alert_active": False,
"demo_active": is_demo_running(),
}
@app.post("/demo")
def demo():
start_demo()
return {
"demo_active": True,
"alert_active": is_alert_running(),
"sequence": [
color_name
for _, color_name in DEMO_SEQUENCE
],
"interval_seconds": DEMO_INTERVAL_SECONDS,
}
@app.post("/demo/off")
def demo_off():
stop_demo(turn_colors_off=True)
return {
"active_color": None,
"demo_active": False,
"alert_active": is_alert_running(),
}
@app.post("/off")
def off():
turn_everything_off()
return {
"active_color": None,
"alert_active": False,
"demo_active": False,
}
@app.get("/status")
def status():
with color_lock:
color_status = {
"green": {
"gpio": RELAY_PINS[1],
"on": relays[1].is_active,
},
"yellow": {
"gpio": RELAY_PINS[2],
"on": relays[2].is_active,
},
"red": {
"gpio": RELAY_PINS[3],
"on": relays[3].is_active,
},
}
current_color = active_color
alert_running = is_alert_running()
demo_running = is_demo_running()
return {
"active_color": current_color,
"alert_active": alert_running,
"demo_active": demo_running,
"relays": {
**color_status,
"alert": {
"gpio": RELAY_PINS[4],
"on": relays[4].is_active,
"blinking": alert_running,
},
},
}