Version 1.3

Signed-off-by: Pol Henarejos <pol.henarejos@cttc.es>

picokey/PicoKey.py

@@ -25,6 +25,7 @@ from .RescuePicoKey import RescuePicoKey
 from .RescueMonitor import RescueMonitor, RescueMonitorObserver
 from .PhyData import PhyData
 from .core import NamedIntEnum
+from .core.exceptions import PicoKeyNotFoundError, PicoKeyInvalidStateError
 import usb.core
 from .core.log import get_logger
 
@@ -156,8 +157,10 @@ class PicoKey:
                 def update(self, actions: tuple[Optional[usb.core.Device], Optional[usb.core.Device]]):
                     (connected, disconnected) = actions
                     if connected:
+                        logger.debug("Observer: Rescue device connected")
                         pass
                     if disconnected:
+                        logger.debug("Observer: Rescue device disconnected, closing...")
                         self.__device.close()
             try:
                 self.__card = RescuePicoKey()
@@ -169,7 +172,7 @@ class PicoKey:
                 self.__monitor = RescueMonitor(device=self.__card, cls_callback=self.__observer)
             except Exception:
                 logger.error("No PicoKey device detected")
-                raise Exception('time-out: no card inserted')
+                raise PicoKeyNotFoundError('time-out: no card inserted')
         try:
             logger.debug("Selecting applet in rescue mode...")
             resp, sw1, sw2 = self.select_applet(rescue=True)
@@ -208,27 +211,35 @@ class PicoKey:
             self.__card.close()
         else:
             logger.debug("Removing card monitor observer...")
-            self.__monitor.deleteObserver(self.__observer)
-            self.__observer = None
-            self.__monitor = None
+            if (self.__monitor and self.__observer):
+                self.__monitor.deleteObserver(self.__observer)
+                self.__observer = None
+                self.__monitor = None
             logger.debug("Disconnecting and releasing card...")
-            self.__card.disconnect()
-            logger.debug("Card disconnected")
-            self.__card.release()
+            try:
+                self.__card.disconnect()
+                logger.debug("Card disconnected")
+                self.__card.release()
+            except Exception as e:
+                logger.error("Error during card disconnect/release: " + str(e))
         self.__card = None
 
     def transmit(self, apdu: list[int]):
         if (not self.__card):
             logger.error("No device connected")
-            raise Exception('No device connected')
-        response, sw1, sw2 = self.__card.transmit(apdu)
-        return response, sw1, sw2
+            raise PicoKeyNotFoundError('No device connected')
+        try:
+            response, sw1, sw2 = self.__card.transmit(apdu)
+            return response, sw1, sw2
+        except Exception as e:
+            logger.error("Transmission error: " + str(e))
+            raise PicoKeyInvalidStateError("Transmission error: " + str(e))
 
     def send(self, command: int, cla: int = 0x00, p1: int =0x00, p2: int=0x00, ne : Optional[int] = None, data : Optional[list[int]] = None, codes : list[int] = []):
         logger.debug(f"Sending command {hex(command)} with cla={hex(cla)}, p1={hex(p1)}, p2={hex(p2)}, ne={ne}")
         if (not self.__card):
             logger.error("No device connected")
-            raise Exception('No device connected')
+            raise PicoKeyNotFoundError('No device connected')
         lc = []
         dataf = []
         if (data):
@@ -256,12 +267,18 @@ class PicoKey:
             response, sw1, sw2 = self.__card.transmit(apdu)
         except Exception:
             logger.debug("Reconnecting card after transmit failure")
-            self.__card.reconnect()
+
+            try:
+                self.__card.reconnect()
+            except Exception as e:
+                logger.error("Reconnection failed: " + str(e))
+                self.close()
+                raise PicoKeyNotFoundError('Reconnection failed: ' + str(e))
             try:
                 response, sw1, sw2 = self.__card.transmit(apdu)
             except Exception as e:
                 logger.error("APDU transmission error after reconnect: " + str(e))
-                raise Exception("APDU transmission error after reconnect: " + str(e))
+                raise PicoKeyInvalidStateError("APDU transmission error after reconnect: " + str(e))
 
         code = (sw1<<8|sw2)
         if (sw1 != 0x90):

picokey/RescueMonitor.py

@@ -62,6 +62,9 @@ class RescueMonitor:
 
     def _run(self):
         while self._running:
+            if (self._dev is None) or (self._dev.device is None):
+                time.sleep(self.interval)
+                continue
             dev = usb.core.find(idVendor=self._dev.device.idVendor, idProduct=self._dev.device.idProduct)
 
             if dev and not self._device_present:

picokey/RescuePicoKey.py

@@ -17,6 +17,7 @@
  */
 """
 
+import time
 import usb.core
 import usb.util
 import libusb_package
@@ -26,6 +27,15 @@ from .core.log import get_logger
 
 logger = get_logger("RescuePicoKey")
 
+class RescuePicoKeyError(Exception):
+    pass
+
+class RescuePicoKeyNotFoundError(RescuePicoKeyError):
+    pass
+
+class RescuePicoKeyInvalidStateError(RescuePicoKeyError):
+    pass
+
 class RescuePicoKey:
 
     def __init__(self):
@@ -89,7 +99,7 @@ class RescuePicoKey:
                         break
         if (not found):
             logger.error("No suitable device found")
-            raise Exception('Not found any Pico Key device')
+            raise RescuePicoKeyNotFoundError('Not found any Pico Key device')
 
     @property
     def device(self):
@@ -117,16 +127,24 @@ class RescuePicoKey:
 
     def read(self, timeout=2000):
         logger.debug("Reading data from device")
-        ret = self.__dev.read(self.__in, 4096, timeout)
-        logger.trace(f"Data read from device: {' '.join([f'{x:02X}' for x in ret])}")
-        logger.debug("Read data from device")
-        return ret
+        try:
+            ret = self.__dev.read(self.__in, 4096, timeout)
+            logger.trace(f"Data read from device: {' '.join([f'{x:02X}' for x in ret])}")
+            logger.debug("Read data from device")
+            return ret
+        except Exception as e:
+            logger.error("USB read error: " + str(e))
+            raise RescuePicoKeyInvalidStateError("USB read error: " + str(e))
 
     def write(self, data, timeout=2000):
         logger.debug("Writing data to device")
         logger.trace(f"Data to write to device: {' '.join([f'{x:02X}' for x in data])}")
-        assert(self.__dev.write(self.__out, data, timeout) == len(data))
-        logger.debug("Wrote data to device")
+        try:
+            assert(self.__dev.write(self.__out, data, timeout) == len(data))
+            logger.debug("Wrote data to device")
+        except Exception as e:
+            logger.error("USB write error: " + str(e))
+            raise RescuePicoKeyInvalidStateError("USB write error: " + str(e))
 
     def exchange(self, data, timeout=2000):
         logger.debug("Exchanging data with device")
@@ -134,12 +152,12 @@ class RescuePicoKey:
             self.write(data=data, timeout=timeout)
         except Exception as e:
             logger.error("USB write error: " + str(e))
-            raise Exception("USB write error: " + str(e))
+            raise RescuePicoKeyInvalidStateError("USB write error: " + str(e))
         try:
             ret = self.read(timeout=timeout)
         except Exception as e:
             logger.error("USB read error: " + str(e))
-            raise Exception("USB read error: " + str(e))
+            raise RescuePicoKeyInvalidStateError("USB read error: " + str(e))
         return ret
 
     def powerOn(self):
@@ -162,3 +180,16 @@ class RescuePicoKey:
             self.powerOn()
         rapdu = self.__iccd.SendApdu(apdu=apdu)
         return rapdu[:-2], rapdu[-2], rapdu[-1]
+
+    def reconnect(self):
+        logger.debug("Reconnecting to device")
+        self.close()
+        time.sleep(1)
+        try:
+            self.__init__()
+        except Exception as e:
+            logger.error("Reconnection failed: %s", e)
+            self.close()
+            raise e
+        logger.debug("Reconnected to device")
+        return self

picokey/__init__.py

@@ -6,3 +6,4 @@ from .SWCodes import SWCodes
 from .RescuePicoKey import RescuePicoKey
 from .PhyData import PhyData, PhyCurve, PhyUsbItf, PhyLedDriver, PhyOpt
 from .core import enums
+from .core.exceptions import PicoKeyError, PicoKeyNotFoundError, PicoKeyInvalidStateError

picokey/_version.py

@@ -18,4 +18,4 @@
  */
 """
 
-__version__ = "1.2.1"
+__version__ = "1.3"

picokey/core/__init__.py

@@ -1 +1,2 @@
 from .enums import *
+from .exceptions import *

picokey/core/enums.py

@@ -1,3 +1,22 @@
+"""
+/*
+ * This file is part of the pypicokey distribution (https://github.com/polhenarejos/pypicokey).
+ * Copyright (c) 2025 Pol Henarejos.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, version 3.
+ *
+ * This program is distributed in the hope that it will be useful, but
+ * WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+"""
+
 import enum
 from typing import Union
 

picokey/core/exceptions.py

@@ -0,0 +1,27 @@
+"""
+/*
+ * This file is part of the pypicokey distribution (https://github.com/polhenarejos/pypicokey).
+ * Copyright (c) 2025 Pol Henarejos.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, version 3.
+ *
+ * This program is distributed in the hope that it will be useful, but
+ * WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+"""
+
+class PicoKeyError(Exception):
+    pass
+
+class PicoKeyNotFoundError(PicoKeyError):
+    pass
+
+class PicoKeyInvalidStateError(PicoKeyError):
+    pass

picokey/core/log.py

@@ -56,4 +56,3 @@ def get_logger(name: str):
         logger.addHandler(handler)
 
     return logger
-