31 lines
876 B
Python
31 lines
876 B
Python
import threading
|
|
from contextlib import contextmanager
|
|
|
|
# Dedicated lock for I2C bus access (used by GrovePi sensors)
|
|
grove_lock = threading.Lock()
|
|
# Dedicated lock for UART/Serial port access
|
|
serial_lock = threading.Lock()
|
|
|
|
@contextmanager
|
|
def safe_grove_access(timeout=1.0):
|
|
"""
|
|
Safely acquire the Grove I2C lock with a timeout.
|
|
Yields True if lock acquired, False otherwise.
|
|
Guarantees lock release only when successfully acquired.
|
|
"""
|
|
acquired = grove_lock.acquire(timeout=timeout)
|
|
try:
|
|
yield acquired
|
|
finally:
|
|
if acquired:
|
|
grove_lock.release()
|
|
|
|
@contextmanager
|
|
def safe_serial_access(timeout=1.0):
|
|
"""Safely acquire the UART/Serial lock with a timeout."""
|
|
acquired = serial_lock.acquire(timeout=timeout)
|
|
try:
|
|
yield acquired
|
|
finally:
|
|
if acquired:
|
|
serial_lock.release() |