Black formatting

Author: patfuchs

sciopy/EIT_16_32_64_128.py

@@ -29,8 +29,12 @@
 }
 
 from .sciopy_dataclasses import EitMeasurementSetup
-from sciopydev.sciopy.usb_message_parser import (MessageParser, make_eitframes_hex, get_data_as_matrix,
-                                                 make_results_folder)
+from sciopydev.sciopy.usb_message_parser import (
+    MessageParser,
+    make_eitframes_hex,
+    get_data_as_matrix,
+    make_results_folder,
+)
 
 
 class EIT_16_32_64_128:
@@ -152,7 +156,6 @@ def disconnect_device(self):
         """
         self.device.close()
 
-
     def send_message(self, message):
         """
         Wrapper function to send a byte array to the device. Communication method is based on the defined serial
@@ -166,7 +169,6 @@ def send_message(self, message):
         elif self.serial_protocol == "FS":
             self.device.write(message)
 
-
     def read_message(self):
         """
         Wrapper function to read single bytes from the device. Communication method is based on the defined serial
@@ -253,7 +255,9 @@ def write_command_string(self, command):
         """
         self.cMessageParser.bPrintMessages = self.print_msg
         self.send_message(command)
-        self.cMessageParser.read_usb_till_timeout(bSaveData=False, bDeleteDataFrame=True)
+        self.cMessageParser.read_usb_till_timeout(
+            bSaveData=False, bDeleteDataFrame=True
+        )
 
     # --- sciospec device commands
 
@@ -482,8 +486,9 @@ def ResetMeasurementSetup(self):
         self.write_command_string(bytearray([0xB0, 0x01, 0x01, 0xB0]))
         self.print_msg = False
 
-
-    def update_measurement_mode(self, meamode: str = "skip4", boundary: str = "external"):
+    def update_measurement_mode(
+        self, meamode: str = "skip4", boundary: str = "external"
+    ):
         """
         Updates the measurement modes out of the options "singleended", "skip0", "skip2" or "skip4"
 
@@ -492,20 +497,27 @@ def update_measurement_mode(self, meamode: str = "skip4", boundary: str = "exter
             boundary (str): Return if the boundary for skip patterns is internal of a channel group or external for all
                 optional channels
         """
-        meamodeoptions = {"singleended": 0x01, "skip0": 0x02, "skip2": 0x03, "skip4": 0x04}
+        meamodeoptions = {
+            "singleended": 0x01,
+            "skip0": 0x02,
+            "skip2": 0x03,
+            "skip4": 0x04,
+        }
         self.print_msg = True
         try:
             cmd = meamodeoptions[meamode]
         except:
-            print("Option for measurement mode is unknown. Measurement mode ist set to single-ended.")
+            print(
+                "Option for measurement mode is unknown. Measurement mode ist set to single-ended."
+            )
             cmd = 0x01
         if boundary == "external":
             bnd = 0x02
         else:
             bnd = 0x01
         self.write_command_string(bytearray([0xB0, 0x03, 0x08, cmd, bnd, 0xB0]))
         self.print_msg = False
-        #todo read out ACK messages
+        # todo read out ACK messages
 
     def GetMeasurementSetup(self, setup_of: str):
         """
@@ -542,7 +554,7 @@ def GetMeasurementSetup(self, setup_of: str):
         print("TBD: Translation")
         self.print_msg = False
 
-    #todo
+    # todo
     def StartStopMeasurementFast(self, return_as="pot_mat"):
         """
         Starts and stops a measurement process using the configured serial protocol (HS or FS).
@@ -582,9 +594,15 @@ def StartStopMeasurementFast(self, return_as="pot_mat"):
         elif return_as == "pot_mat":
             return self.get_data_as_matrix()
 
-
-    def StartStopMeasurementNew(self, timeout:int=0, return_as="pot_mat", bSaveData:bool=False, bDeleteData:bool=False,
-                                sSavepath:str="C/", bResultsFolder=True):
+    def StartStopMeasurementNew(
+        self,
+        timeout: int = 0,
+        return_as="pot_mat",
+        bSaveData: bool = False,
+        bDeleteData: bool = False,
+        sSavepath: str = "C/",
+        bResultsFolder=True,
+    ):
         """
         Starts and stops a measurement process using the configured serial protocol (HS or FS).
         Sends appropriate commands to the device to initiate and terminate measurement.
@@ -611,24 +629,40 @@ def StartStopMeasurementNew(self, timeout:int=0, return_as="pot_mat", bSaveData:
 
         # Start measurement
         self.cMessageParser.clear_out_data()
-        sCurrentPath= make_results_folder(bResultsFolder, bSaveData, sSavepath) # No new path is created  if bResultsFolder=False
-                                    
+        sCurrentPath = make_results_folder(
+            bResultsFolder, bSaveData, sSavepath
+        )  # No new path is created  if bResultsFolder=False
+
         self.send_message(bytearray([0xB4, 0x01, 0x01, 0xB4]))
         self.cMessageParser.bPrintMessages = False
         if timeout != 0:
-            self.cMessageParser.read_usb_for_seconds(timeout, bSaveData=bSaveData, bDeleteDataFrame=bDeleteData,
-                                                            sSavePath=sCurrentPath,bResultsFolder=False )
+            self.cMessageParser.read_usb_for_seconds(
+                timeout,
+                bSaveData=bSaveData,
+                bDeleteDataFrame=bDeleteData,
+                sSavePath=sCurrentPath,
+                bResultsFolder=False,
+            )
         else:
-            if self.setup.burst_count ==0:
+            if self.setup.burst_count == 0:
                 print("Burst count for this setup needs to be >=1")
                 return
-            self.cMessageParser.read_usb_till_timeout(bSaveData=bSaveData, bDeleteDataFrame=bDeleteData,sSavePath=sSavepath, bResultsFolder=False)
+            self.cMessageParser.read_usb_till_timeout(
+                bSaveData=bSaveData,
+                bDeleteDataFrame=bDeleteData,
+                sSavePath=sSavepath,
+                bResultsFolder=False,
+            )
 
         # Stop measurement
         self.send_message(bytearray([0xB4, 0x01, 0x00, 0xB4]))
         # All data is returned if wanted
-        data = self.cMessageParser.read_usb_till_timeout(bSaveData=bSaveData, bDeleteDataFrame=bDeleteData,
-                                                          sSavePath=sCurrentPath, bResultsFolder=False)
+        data = self.cMessageParser.read_usb_till_timeout(
+            bSaveData=bSaveData,
+            bDeleteDataFrame=bDeleteData,
+            sSavePath=sCurrentPath,
+            bResultsFolder=False,
+        )
 
         if bDeleteData:
             return
@@ -639,8 +673,6 @@ def StartStopMeasurementNew(self, timeout:int=0, return_as="pot_mat", bSaveData:
         elif return_as == "eitframe":
             return data
 
-
-
     def get_data_as_matrix(self):
         """
         Converts the raw EIT data into a 3D matrix of potentials.

sciopy/com_util.py

@@ -15,6 +15,7 @@
 from glob import glob
 from .datatype_conversion import *
 
+
 def available_serial_ports() -> list:
     """
     Lists serial port names.
@@ -66,7 +67,6 @@ def clTbt_dp(val: float) -> list:
     return [int(ele) for ele in struct.pack(">d", val)]
 
 
-
 def reshape_full_message_in_bursts(lst: list, ssms: EitMeasurementSetup) -> np.ndarray:
     """
     Takes the full message buffer and splits this message depeding on the measurement configuration into the
@@ -106,11 +106,10 @@ def length_correction(array: list) -> list:
     # split in burst count messages
     split_length = lst.shape[0] // ssms.burst_count
     for split in range(ssms.burst_count):
-        split_list.append(lst[split * split_length: (split + 1) * split_length])
+        split_list.append(lst[split * split_length : (split + 1) * split_length])
     return np.array(split_list)
 
 
-
 def parse_single_frame(lst_ele: np.ndarray) -> SingleFrame:
     """
     Parse single data to the class SingleFrame.
@@ -130,8 +129,8 @@ def parse_single_frame(lst_ele: np.ndarray) -> SingleFrame:
     for i in range(11, 135, 8):
         enum += 1
         channels[f"ch_{enum}"] = complex(
-            bytesarray_to_float(lst_ele[i: i + 4]),
-            bytesarray_to_float(lst_ele[i + 4: i + 8]),
+            bytesarray_to_float(lst_ele[i : i + 4]),
+            bytesarray_to_float(lst_ele[i + 4 : i + 8]),
         )
 
     excitation_stgs = np.array([single_hex_to_int(ele) for ele in lst_ele[3:5]])
@@ -149,7 +148,7 @@ def parse_single_frame(lst_ele: np.ndarray) -> SingleFrame:
 
 
 def split_bursts_in_frames(
-        split_list: np.ndarray, burst_count: int, channel_group: list
+    split_list: np.ndarray, burst_count: int, channel_group: list
 ) -> np.ndarray:
     """
     Takes the splitted list from `reshape_full_message_in_bursts()` and parses the single frames.

sciopy/datatype_conversion.py

@@ -13,6 +13,7 @@
 TWOPOWER16 = 65536
 TWOPOWER8 = 256
 
+
 # -------------------------------------------------------------------------------------------------------------------- #
 # -------------------------------------------------------------------------------------------------------------------- #
 def del_hex_in_list(lst: list) -> np.ndarray:
@@ -169,7 +170,12 @@ def four_byte_to_int(bytelist):
     int
         integer number
     """
-    return TWOPOWER24 * bytelist[0] + TWOPOWER16 * bytelist[1] + TWOPOWER8 * bytelist[2] + bytelist[3]
+    return (
+        TWOPOWER24 * bytelist[0]
+        + TWOPOWER16 * bytelist[1]
+        + TWOPOWER8 * bytelist[2]
+        + bytelist[3]
+    )
 
 
 # -------------------------------------------------------------------------------------------------------------------- #

sciopy/device_interface.py

@@ -0,0 +1,55 @@
+"""
+Project ：sciopy
+Directory: sciopy/sciopy
+File : device_interface.py
+Author ：Patricia Fuchs
+Date ：26.11.2025 14:04
+"""
+
+try:
+    import serial
+except ImportError:
+    print("Could not import module: serial")
+
+# -------------------------------------------------------------------------------------------------------------------- #
+# -------------------------------------------------------------------------------------------------------------------- #
+
+
+class DeviceInterface:
+    def __init__(self):
+        self.sProtocol = "None"
+
+    def send_data(self, data):
+        pass
+
+    def read_data(self):
+        return None
+
+
+import serial
+
+
+class USB_FS_Device(DeviceInterface):
+    def __init__(self, port: str, baudrate: int = 9600, timeout: int = 9000):
+        super().__init__()
+        self.sProtocol = "FS"
+        self.device = serial.Serial(
+            port=port,
+            baudrate=baudrate,
+            timeout=timeout,
+            parity=serial.PARITY_NONE,
+            stopbits=serial.STOPBITS_ONE,
+            bytesize=serial.EIGHTBITS,
+        )
+        self.name = self.device.name
+
+    def send_data(self, data):
+        self.device.write(data)
+
+    def read_data(self):
+        return self.device.read()
+
+
+class USB_HS_Device(DeviceInterface):
+    def __init__(self, port: str, baudrate: int = 9600, timeout: int = 9000):
+        super().__init__()

sciopy/sciopy_dataclasses.py

@@ -4,6 +4,7 @@
 from typing import List, Tuple, Union
 import numpy.typing as npt
 
+
 @dataclass
 class EitMeasurementSetup:
     """
@@ -213,7 +214,6 @@ class PreperationConfig:
     n_samples: int
 
 
-
 # -------------------------------------------------------------------------------------------------------------------- #
 @dataclass
 class EITFrame:
@@ -233,11 +233,10 @@ class EITFrame:
                  for f in used frequency stages:
                     for c in all used channels: -> insert complex value
     """
+
     n_el: int  # Number of used electrodes
     excitation_stgs: npt.NDArray[int]  # Num Excitation Settings X 2
     frequency_stgs: npt.NDArray[int]  # List of Frequency-Sweep Settings,
     timestamp1: int  # [ms]
     timestamp2: int
     ppcData: npt.NDArray[complex]  # Channels 1-(64) all channel groups combined
-
-