Permission Denied due to Contention (#387)

* Bug fixed: incorrect loop condition logic. (#383) (#386)

* Added `_available_ports`. (#386)

* Bug fixed: contention. (#386)

* Code reformatted. (#386)

leads_comm_serial/connection.py

@@ -27,7 +27,7 @@ def receive(self, chunk_size: int = 1) -> bytes | None:
         start = _time()
         try:
             msg = chunk = b""
-            while (timeout := self._serial.timeout) and _time() - start < timeout and self._separator not in chunk:
+            while not ((t_o := self._serial.timeout) and _time() - start > t_o) and self._separator not in chunk:
                 msg += (chunk := self._require_open_serial().read(chunk_size))
             return self.with_remainder(msg)
         except IOError:

leads_comm_serial/identity.py

@@ -6,6 +6,9 @@
 
 from leads_comm_serial.connection import SerialConnection
 
+_available_ports: list[str] = [_p for _p, _, __ in _comports()]
+_lock: _Lock = _Lock()
+
 
 class AutoIdentity(object, metaclass=_ABCMeta):
     def __init__(self, retry: bool = False, remainder: bytes = b"", seperator: bytes = b";") -> None:
@@ -21,59 +24,65 @@ def meta(self) -> tuple[bool, bytes, bytes]:
     def suggest_next_port(self, tried_port: str | None = None) -> str | None:
         if tried_port:
             self._tried_ports.append(tried_port)
-        for port, _, __ in _comports():
+        for port in _available_ports:
             if port not in self._tried_ports:
                 return port
 
     @_abstractmethod
     def check_identity(self, connection: SerialConnection) -> bool:
         raise NotImplementedError
 
-    def establish_connection(self, serial: _Serial) -> SerialConnection:
-        _detect_ports()
+    def _establish_connection_no_lock(self, serial: _Serial) -> SerialConnection:
         try:
+            _detect_ports()
             if port := _instances[self]:
                 serial.port = port
+                self._retry = False
             elif not serial.port:
                 raise ConnectionError("No available port")
+            elif serial.port not in _available_ports:
+                raise ValueError("Port taken")
             serial.open()
             connection = SerialConnection(serial).suspect()
             if port or self.check_identity(connection):
                 return connection.trust()
             for instance in _instances.keys():
                 if instance.check_identity(connection):
                     _instances[instance] = serial.port
-            serial.close()
             raise ValueError("Unexpected identity")
         except (_SerialException, ValueError) as e:
+            serial.close()
             if not self._retry:
                 raise ConnectionError(f"Unable to establish connection due to {repr(e)}")
             serial.port = self.suggest_next_port(serial.port)
-            return self.establish_connection(serial)
+            return self._establish_connection_no_lock(serial)
+
+    def establish_connection(self, serial: _Serial) -> SerialConnection:
+        _lock.acquire()
+        try:
+            return self._establish_connection_no_lock(serial)
+        finally:
+            _lock.release()
 
 
 _instances: dict[AutoIdentity, str | None] = {}
 _ports_detected: bool = False
-_ports_lock: _Lock = _Lock()
 
 
 def _detect_ports() -> None:
     global _ports_detected
-    _ports_lock.acquire()
-    try:
-        if _ports_detected:
-            return
-        for port, _, __ in _comports():
-            serial = _Serial()
-            serial.port = port
-            serial.open()
-            connection = SerialConnection(serial).suspect()
-            for instance in _instances.keys():
-                connection._separator = instance.meta()[2]
-                if instance.check_identity(connection):
-                    _instances[instance] = port
-                    break
-            serial.close()
-    finally:
-        _ports_detected = True
-        _ports_lock.release()
+    if _ports_detected:
+        return
+    for port in tuple(_available_ports):
+        serial = _Serial()
+        serial.port = port
+        serial.open()
+        connection = SerialConnection(serial).suspect()
+        for instance in _instances.keys():
+            connection._separator = instance.meta()[2]
+            if instance.check_identity(connection):
+                _instances[instance] = port
+                _available_ports.remove(port)
+                break
+        serial.close()
+    _ports_detected = True