import re from typing import Union, Optional from enum import IntEnum from UniTAP.dev import * from UniTAP.libs.lib_tsi import BaseIO, DeviceMaskInternal from UniTAP.utils import tsi_logging as logging class DeviceMask(IntEnum): Sink = 0 Source = 1 All = 2 Nothing = 3 class DeviceAlreadyInUse(Exception): """ Redefinition of base exception. Define error of 'device already in use'. """ pass class FailedToOpenDevice(Exception): """ Redefinition of base exception. Define error of 'device already in use'. """ pass class DeviceNotFound(Exception): """ Redefinition of base exception. Define error of 'device not found'. """ pass class DeviceNotSupported(Exception): """ Redefinition of base exception. Define error of 'device not supported'. """ pass class TsiLib: """ Class `TsiLib` allows working with TSI Devices. - Open selected device `open`. """ __device_mask_convert = {DeviceMask.Sink: DeviceMaskInternal.Sink, DeviceMask.Source: DeviceMaskInternal.Source, DeviceMask.All: DeviceMaskInternal.All, DeviceMask.Nothing: DeviceMaskInternal.Nothing} def __init__(self): logging.info(f"[UniTAP] TsiLib.init {self}") self.__io = BaseIO() self.__device_list = [] def cleanup(self): """ Clear list of devices and call function `TSI_Clean` for cleaning TSI library. """ self.__device_list.clear() self.__io.cleanup() logging.info(f"[UniTAP] TsiLib.cleanup {self}") def __del__(self): """ Automatically call the destructor after the script completes. Call function `cleanup`. """ self.cleanup() logging.info(f"[UniTAP] TsiLib.del {self}") def open(self, info: Union[str, int]) -> TSIDevice: """ Open selected TSI device. Args: info (str|int) - serial number of device or device index. Returns: object of `TSIDevice` type """ dev_handle = None dev_full_name = "" if isinstance(info, str) and len(info) == 8: for dev_idx in range(self.__io.get_device_count()): dev_full_name = self.__io.get_device_name(dev_idx) if info.lower() in dev_full_name.lower(): dev_io = self.__io.create_device_io(dev_idx) break else: available_device_list = [] for dev_idx in range(self.__io.get_device_count()): available_device_list.append(f"{dev_idx}: {self.__io.get_device_name(dev_idx)}") available_device_list_str = '\n'.join(available_device_list) raise DeviceNotFound(f"Device [{info}] was not found, available device list:\n" f"{available_device_list_str}") elif isinstance(info, int): if info < self.__io.get_device_count(): dev_full_name = self.__io.get_device_name(info) dev_io = self.__io.create_device_io(info) else: available_device_list = [] for dev_idx in range(self.__io.get_device_count()): available_device_list.append(f"{dev_idx}: {self.__io.get_device_name(dev_idx)}") available_device_list_str = '\n'.join(available_device_list) raise DeviceNotFound(f"Device with index {info} was not found, available device list:\n" f"{available_device_list_str}") else: raise ValueError("'info' must be serial number (8 symbol str) or device index.") if "in use" in dev_full_name: raise DeviceAlreadyInUse(f"Device [{info}] is already in use by other application.") if "not supported" in dev_full_name: raise DeviceNotSupported(f"Device [{info}] is not supported.") if dev_io is None: raise FailedToOpenDevice(f"Failed to open device [{info}].") dev_series, dev_model, dev_serial = self.__extract_dev_info(dev_full_name) dev_series_model = f'{dev_series}-{dev_model}' role_list = [] for role_idx in range(dev_io.get_device_role_count()): role_full_name = dev_io.get_device_role_name(role_idx) role_list.append(MODEL_TO_CLASS[dev_series_model].ROLE_DICT[role_full_name]) self.__device_list.append(TSIDevice(dev_io, dev_series_model, dev_serial, role_list)) return weakref.proxy(self.__device_list[-1]) def close(self, device: TSIDevice): """ Close selected device `TSIDevice` (removing from list of devices) Args: device (`TSIDevice`) - object of `TSIDevice` type """ self.__device_list.remove(device) def get_list_of_available_devices(self, require_caps: Optional[DeviceMask] = None, unallowed_caps: Optional[DeviceMask] = None) -> list: """ Returns list of available devices for using by selected masks. Args: require_caps (`DeviceMask`) - object of `DeviceMask` type unallowed_caps (`DeviceMask`) - object of `DeviceMask` type Returns: object of `list` type """ available_device_list = [] if require_caps is not None and unallowed_caps is not None: self.__set_search_mask(require_caps, unallowed_caps) for dev_idx in range(self.__io.get_device_count()): available_device_list.append(f"{dev_idx}: {self.__io.get_device_name(dev_idx)}") return available_device_list def __set_search_mask(self, require_caps: DeviceMask, unallowed_caps: DeviceMask): return self.__io.get_devices_by_mask(self.__device_mask_convert.get(require_caps), self.__device_mask_convert.get(unallowed_caps)) def get_list_of_available_roles(self, dev_series_model: Union[str, TSIDevice]) -> list: """ Returns list of available roles for the selected device. Args: dev_series_model (`str`|`TSIDevice`) - serial number of device Returns: object of `list` type with `MODEL_TO_CLASS` roles """ if isinstance(dev_series_model, str) and dev_series_model not in MODEL_TO_CLASS.keys(): _list_model_to_class = '\n'.join(f'{i}: {e}' for i, e in enumerate(MODEL_TO_CLASS.keys())) raise ValueError(f"Incorrect Device series model {dev_series_model}. Must be from list:\n" f"{_list_model_to_class}") available_role_list = [] if isinstance(dev_series_model, str): devices = self.get_list_of_available_devices() for dev in devices: series_model = re.findall(r'UCD-\d+', dev)[0] if dev_series_model == series_model: available_role_list = list(MODEL_TO_CLASS[dev_series_model].ROLE_DICT.values()) elif isinstance(dev_series_model, TSIDevice): available_role_list = dev_series_model.available_roles return available_role_list def get_str_list_of_available_roles(self, dev_series_model: Union[str, TSIDevice]) -> list: """ Returns list of available roles for the selected device. Args: dev_series_model (`str`|`TSIDevice`) - serial number of device Returns: object of `list` type with `str` """ if isinstance(dev_series_model, str) and dev_series_model not in MODEL_TO_CLASS.keys(): _list_model_to_class = '\n'.join(f'{i}: {e}' for i, e in enumerate(MODEL_TO_CLASS.keys())) raise ValueError(f"Incorrect Device series model {dev_series_model}. Must be from list:\n" f"{_list_model_to_class}") available_role_list = [] if isinstance(dev_series_model, str): devices = self.get_list_of_available_devices() for dev in devices: series_model = re.findall(r'UCD-\d+', dev)[0] if dev_series_model == series_model: available_role_list = [e for i, e in enumerate(MODEL_TO_CLASS[dev_series_model].ROLE_DICT.keys())] elif isinstance(dev_series_model, TSIDevice): dev_series_model = re.findall(r'UCD\d+', str(dev_series_model.available_roles[0]))[0]. \ replace("UCD", "UCD-") available_role_list = [f"{i}: {e}" for i, e in enumerate(MODEL_TO_CLASS[dev_series_model].ROLE_DICT.keys())] return available_role_list @staticmethod def __extract_dev_info(dev_full_name: str): if dev_full_name.count(" ") == 1: dev_name, dev_serial = dev_full_name.split(' ') else: dev_name, dev_serial = dev_full_name.split(":")[0].split(" ") dev_series, dev_model = dev_name.split('-') dev_serial = dev_serial[1:9] return dev_series, int(dev_model), dev_serial