Add py.typed & type check package

koradctl/pretty.py

@@ -1,26 +1,34 @@
+from __future__ import annotations
+
 from typing import Union
-from collections import namedtuple
+from typing import NamedTuple
 
-status_fields = {
-    'output_enabled':  lambda x: bool( x & 0x40),
-    'ovp_ocp_enabled': lambda x: bool( x & 0x80),
-    'cv_active':       lambda x: bool( x & 0x01),
-    'cc_active':       lambda x: bool(~x & 0x01),
-}
-Status = namedtuple('Status', status_fields.keys())
+class Status(NamedTuple):
+    output_enabled: bool
+    ovp_ocp_enabled: bool
+    cv_active: bool
+    cc_active: bool
 
-def pretty_status(response: bytes):
-    status_byte = response[0]
-    d = { k: fn(status_byte) for k,fn in status_fields.items() }
-    return Status(**d)
+    @staticmethod
+    def from_status_byte(status_byte: int) -> Status:
+        assert 0 <= status_byte <= 255
+        return Status(
+            output_enabled=bool(status_byte & 0x40),
+            ovp_ocp_enabled=bool(status_byte & 0x80),
+            cv_active=bool(status_byte & 0x01),
+            cc_active=bool(~status_byte & 0x01),
+        )
 
+def pretty_status(response: bytes) -> Status:
+    status_byte = response[0]
+    return Status.from_status_byte(status_byte)
 
-reading_fields = {
-    'value': lambda x,y: round(float(x), 3),
-    'units': lambda x,y: y,
-}
-Reading = namedtuple('Reading', reading_fields.keys())
+class Reading(NamedTuple):
+    value: float
+    units: str
 
 def pretty_reading(response: Union[bytes, float], units: str = '?'):
-    d = { k: fn(response, units) for k,fn in reading_fields.items() }
-    return Reading(**d)
+    return Reading(
+        value = round(float(response), 3),
+        units = units
+    )

koradctl/psu.py

@@ -41,7 +41,7 @@ def insert_command_delay(self):
 
         sleep(time_until_next_command)
 
-    def issue_command(self, command: Union[bytes, str], max_response_size: int = 1000, wait_for_response: bool = True, allow_retry: bool = False) -> bytes:
+    def issue_command(self, command: Union[bytes, str], max_response_size: int = 1000, wait_for_response: bool = True, allow_retry: bool = False) -> Union[bytes, None]:
         """
         this function is really for internal use... you shouldn't need to call it
         it will encode the command (if necessary), and then wait for a response
@@ -68,7 +68,7 @@ def issue_command(self, command: Union[bytes, str], max_response_size: int = 100
 
         return response
 
-    def issue_command_trim(self, command: Union[bytes, str], max_response_size: int = 1000, wait_for_response: bool = True, allow_retry: bool = False, trim_chars: bytes = b'\x00') -> bytes:
+    def issue_command_trim(self, command: Union[bytes, str], max_response_size: int = 1000, wait_for_response: bool = True, allow_retry: bool = False, trim_chars: bytes = b'\x00') -> Union[bytes, None]:
         """
         this function is really for internal use... you shouldn't need to call it
         it will call issue_command(), and then trim the response by removing any
@@ -86,7 +86,11 @@ def get_identity(self) -> str:
         """
         get the power supply's identity string, e.g: "TENMA 72-2540 V2.1"
         """
-        return self.issue_command_trim('*IDN?', allow_retry=True).decode('ascii')
+        response = self.issue_command_trim('*IDN?', allow_retry=True)
+        if response is None:
+            raise RuntimeError("Powersupply failed to answer command.")
+
+        return response.decode('ascii')
 
     def is_tested(self) -> bool:
         """
@@ -101,8 +105,10 @@ def is_tested(self) -> bool:
 
     def get_status(self) -> Status:
         response = self.issue_command('STATUS?', allow_retry=True)
-        return pretty_status(response)
+        if response is None:
+            raise RuntimeError("Powersupply failed to answer command.")
 
+        return pretty_status(response)
 
     def get_output_state(self) -> bool:
         return self.get_status().output_enabled
@@ -123,6 +129,9 @@ def set_ovp_state(self, enabled: bool):
 
     def get_voltage_setpoint(self) -> Reading:
         response = self.issue_command_trim('VSET1?', allow_retry=True)
+        if response is None:
+            raise RuntimeError("Powersupply failed to answer command.")
+
         return pretty_reading(response, 'V')
 
     def set_voltage_setpoint(self, voltage: float):
@@ -131,6 +140,9 @@ def set_voltage_setpoint(self, voltage: float):
 
     def get_current_setpoint(self) -> Reading:
         response = self.issue_command_trim('ISET1?', allow_retry=True)
+        if response is None:
+            raise RuntimeError("Powersupply failed to answer command.")
+
         return pretty_reading(response, 'I')
 
     def set_current_setpoint(self, current: float):
@@ -139,10 +151,16 @@ def set_current_setpoint(self, current: float):
 
     def get_output_voltage(self) -> Reading:
         response = self.issue_command_trim('VOUT1?', allow_retry=True)
+        if response is None:
+            raise RuntimeError("Powersupply failed to answer command.")
+
         return pretty_reading(response, 'V')
 
     def get_output_current(self) -> Reading:
         response = self.issue_command_trim('IOUT1?', allow_retry=True)
+        if response is None:
+            raise RuntimeError("Powersupply failed to answer command.")
+
         return pretty_reading(response, 'I')
 
     def get_output_power(self) -> Reading:

koradctl/types.py

@@ -1,10 +1,10 @@
-from typing import Union, List, Tuple
+from typing import Any, Union, List, Tuple
 
 true_values = [ '1', 'on', 'yes' ]
 false_values = [ '0', 'off', 'no' ]
 toggle_values = [ 't', 'toggle' ]
 
-def from_options(arg: str, options: List[Tuple[List[str], bool]]):
+def from_options(arg: str, options: List[Tuple[List[str], Any]]):
     for values, ret in options:
         if arg in values:
             return ret

requirements.txt

@@ -1 +1 @@
-pyserial==3.4
+pyserial~=3.0

setup.py

@@ -46,6 +46,7 @@ def load_package_info(self):
     url='https://github.com/attie/koradctl',
     license='BSD-3-Clause',
     packages=[ 'koradctl' ],
+    package_data={m.package_info['proj_name']: ["py.typed"]},
     install_requires=[
         'pyserial',
     ],