Simplified threading lock uses. (#500) (#501)

Author: ATATC

leads/comm/prototype.py

@@ -58,14 +58,11 @@ def _register_process(self, *args, **kwargs) -> None:
         :param kwargs: kwargs passed to `run()`
         :exception RuntimeError: duplicated registration
         """
-        self._lock.acquire()
         if self._main_thread:
             raise RuntimeError("A service can only run once")
-        try:
+        with self._lock:
             self._main_thread = _Thread(name=f"service{hash(self)}", target=self._run, daemon=True, args=args,
                                         kwargs=kwargs)
-        finally:
-            self._lock.release()
 
     def _parallel_run(self, *args, **kwargs) -> None:
         """

leads/dt/odometer.py

@@ -25,8 +25,5 @@ def __init__(self) -> None:
 
     @_override
     def write(self, payload: float) -> None:
-        self._lock.acquire()
-        try:
+        with self._lock:
             super().write(payload)
-        finally:
-            self._lock.release()

leads/logger.py

@@ -58,12 +58,9 @@ def format(self, msg: str, font: int, color: int | None, background: int | None)
         return f"\033[{font}{f";{color}" if color else ""}{f";{background + 10}" if background else ""}m{msg}\033[0m"
 
     def print(self, msg: str, level: int) -> None:
-        self._lock.acquire()
-        try:
+        with self._lock:
             if self._debug_level <= level:
                 print(msg)
-        finally:
-            self._lock.release()
 
     def info(self, *msg: str, sep: str = " ", end: str = "\n",
              f: tuple[int, int | None, int | None] = (REGULAR, None, None)) -> None:

leads/os.py

@@ -21,11 +21,8 @@ def active(self) -> bool:
 
     @active.setter
     def active(self, active: bool) -> None:
-        self._lock.acquire()
-        try:
+        with self._lock:
             self._active = active
-        finally:
-            self._lock.release()
 
 
 _thread_flags: _ThreadFlags = _ThreadFlags()

leads_comm_serial/identity.py

@@ -58,11 +58,8 @@ def _establish_connection_no_lock(self, serial: _Serial) -> SerialConnection:
             return self._establish_connection_no_lock(serial)
 
     def establish_connection(self, serial: _Serial) -> SerialConnection:
-        _lock.acquire()
-        try:
+        with _lock:
             return self._establish_connection_no_lock(serial)
-        finally:
-            _lock.release()
 
 
 _instances: dict[AutoIdentity, str | None] = {}