108 lines
3.3 KiB
Python
108 lines
3.3 KiB
Python
import threading
|
||
import time
|
||
from typing import Optional
|
||
|
||
|
||
class GPIOController:
|
||
"""Thread-safe singleton controller for GPIO output pin."""
|
||
|
||
_instance = None
|
||
_instance_lock = threading.Lock()
|
||
|
||
def __init__(self, pin: int, simulation_mode: bool = False):
|
||
self.pin = pin
|
||
self.simulation_mode = simulation_mode
|
||
self._lock = threading.Lock()
|
||
self._blink_stop_event = threading.Event()
|
||
self._blink_thread: Optional[threading.Thread] = None
|
||
self._state = False
|
||
|
||
self._gpio = None
|
||
self._gpio_available = False
|
||
if not self.simulation_mode:
|
||
try:
|
||
import RPi.GPIO as GPIO # type: ignore[import-not-found]
|
||
|
||
self._gpio = GPIO
|
||
self._gpio.setmode(GPIO.BCM)
|
||
self._gpio.setup(self.pin, GPIO.OUT)
|
||
self._gpio.output(self.pin, GPIO.LOW)
|
||
self._gpio_available = True
|
||
except Exception:
|
||
self._gpio = None
|
||
self._gpio_available = False
|
||
|
||
@classmethod
|
||
def get_instance(cls, pin: int = 18, simulation_mode: bool = False):
|
||
with cls._instance_lock:
|
||
if cls._instance is None:
|
||
cls._instance = cls(pin=pin, simulation_mode=simulation_mode)
|
||
return cls._instance
|
||
|
||
@classmethod
|
||
def _reset_for_tests(cls):
|
||
with cls._instance_lock:
|
||
if cls._instance is not None:
|
||
cls._instance.cleanup()
|
||
cls._instance = None
|
||
|
||
def _write_locked(self, state: bool) -> None:
|
||
self._state = state
|
||
if self._gpio_available and self._gpio is not None:
|
||
self._gpio.output(self.pin, self._gpio.HIGH if state else self._gpio.LOW)
|
||
|
||
def on(self) -> None:
|
||
with self._lock:
|
||
self._write_locked(True)
|
||
|
||
def off(self) -> None:
|
||
with self._lock:
|
||
self._write_locked(False)
|
||
|
||
def is_on(self) -> bool:
|
||
with self._lock:
|
||
return bool(self._state)
|
||
|
||
@property
|
||
def rpi_gpio_available(self) -> bool:
|
||
"""True, если поднят настоящий RPi.GPIO (Raspberry Pi). В симуляции всегда False."""
|
||
return bool(self._gpio_available)
|
||
|
||
def stop_blink(self) -> None:
|
||
self._blink_stop_event.set()
|
||
if self._blink_thread and self._blink_thread.is_alive():
|
||
self._blink_thread.join(timeout=2)
|
||
self._blink_thread = None
|
||
self._blink_stop_event.clear()
|
||
|
||
def blink(self, times: int = 3, delay: float = 0.2) -> None:
|
||
if times <= 0:
|
||
return
|
||
|
||
self.stop_blink()
|
||
|
||
def _worker():
|
||
for _ in range(times):
|
||
if self._blink_stop_event.is_set():
|
||
break
|
||
self.on()
|
||
time.sleep(delay)
|
||
if self._blink_stop_event.is_set():
|
||
break
|
||
self.off()
|
||
time.sleep(delay)
|
||
self.off()
|
||
|
||
self._blink_thread = threading.Thread(target=_worker, daemon=True, name="gpio-blink")
|
||
self._blink_thread.start()
|
||
|
||
def cleanup(self) -> None:
|
||
self.stop_blink()
|
||
self.off()
|
||
if self._gpio_available and self._gpio is not None:
|
||
try:
|
||
self._gpio.cleanup(self.pin)
|
||
except Exception:
|
||
pass
|
||
|