acroname.py

Go to the documentation of this file.

# License: Apache 2.0. See LICENSE file in root directory.
# Copyright(c) 2021 Intel Corporation. All Rights Reserved.
 
"""
Brainstem Acroname Hub
 
See documentation for brainstem here:
https://acroname.com/reference/api/python/index.html
"""
 
from rspy import log, device_hub
import time
import platform, re
 
if __name__ == '__main__':
    import os, sys, getopt
    def usage():
        ourname = os.path.basename( sys.argv[0] )
        print( 'Syntax: acroname [options]' )
        print( '        Control the acroname USB hub' )
        print( 'Options:' )
        print( '        --enable       Enable all ports' )
        print( '        --disable      Disable all ports' )
        print( '        --recycle      Recycle all ports' )
        print( '        --reset        Reset the acroname' )
        sys.exit(2)
    try:
        opts,args = getopt.getopt( sys.argv[1:], '',
            longopts = [ 'help', 'recycle', 'enable', 'disable', 'reset' ])
    except getopt.GetoptError as err:
        print( '-F-', err )   # something like "option -a not recognized"
        usage()
    if args or not opts:
        usage()
    # See the end of the file for all the option handling
 
 
try:
    import brainstem
except ModuleNotFoundError:
    log.d( 'no acroname library is available' )
    raise
 
class NoneFoundError( RuntimeError ):
    def __init__( self, message = None ):
        super().__init__( self, message  or  'no Acroname module found' )
 
class Acroname(device_hub.device_hub):
 
 
 
    def __init__(self):
        super().__init__()
        if discover() is None:  # raise an error if there is no hub connected
            raise NoneFoundError()
        self.hub = None
        self.all_hubs = None
 
    def connect(self,  reset = False, req_spec = None ):
        """
        Connect to the hub. Raises RuntimeError on failure
        :param reset: When true, the acroname will be reset as part of the connection process
        :param req_spec: Required spec to connect to.
        """
        if not self.hub:
            self.hub = brainstem.stem.USBHub3p()
 
        if req_spec:
            specs = [req_spec]
        else:
            specs = discover()
 
        spec = specs[0]
        result = self.hub.connectFromSpec( spec )
        if result != brainstem.result.Result.NO_ERROR:
            raise RuntimeError( "failed to connect to Acroname (result={})".format( result ))
        elif len(specs) > 1:
            log.d( 'connected to', spec )
 
        if reset:
            log.d("resetting Acroname...")
            result = self.hub.system.reset()
            # According to brainstem support:
            # Result error is expected, so we do not check it
            # * there is also a brainstem internal console print on Linux "error release -4"
            # Disconnection is needed after a reset command
            self.hub.disconnect()
 
            result = None
            for i in range(10):
                result = self.hub.connectFromSpec(spec)
                if result != brainstem.result.Result.NO_ERROR:
                    time.sleep(1)
                else:
                    log.d('reconnected')
                    return
            raise RuntimeError("failed to reconnect to Acroname (result={})".format(result))
 
 
    def is_connected(self):
        return self.hub is not None and self.hub.isConnected()
 
 
    def disconnect(self):
        if self.hub:
            self.hub.disconnect()
            del self.hub
            self.hub = None
 
 
    def all_ports(self):
        """
        :return: a list of all possible ports, even if currently unoccupied or disabled
        """
        return range(8)
 
 
    def ports(self):
        """
        :return: a list of all ports currently occupied (and enabled)
        """
        occupied_ports = []
        for port in self.all_ports():
            if self._port_power(port) > 0.0:
                occupied_ports.append( port )
        return occupied_ports
 
 
    def is_port_enabled(self,  port ):
        """
        query if the input port number is enabled through Acroname
        It doesn't mean the device is enumerated as sometimes if the FW crashed we can have a situation where the port is "Disconnected" and not "OK"
        :param port: port number;
        :return: True if Acroname enabled this port, False otherwise
        """
 
        return self.port_state( port ) != "Disabled"
 
 
    def port_state(self,  port ):
        if port < 0 or port > 7:
            raise ValueError( "port number must be [0-7]" )
        #
        status = self.hub.usb.getPortState( port )
        #log.d("getPortState for port", port ,"return", port_state_bitmask_to_string( status.value ))
        return self.port_state_bitmask_to_string( status.value )
 
 
    def port_state_bitmask_to_string(self,  bitmask ):
        """
        https://acroname.com/reference/devices/usbhub3p/entities/usb.html#usb-port-state
        Bit   |  Port State: Result Bitwise Description
        -----------------------------------------------
        0     |  USB Vbus Enabled - Port is Disabled/Enabled by the Acroname
        1     |  USB2 Data Enabled
        2     |  Reserved
        3     |  USB3 Data Enabled
        4:10  |  Reserved
        11    |  USB2 Device Attached
        12    |  USB3 Device Attached
        13:18 |  Reserved
        19    |  USB Error Flag
        20    |  USB2 Boost Enabled
        21:22 |  Reserved
        23    |  Device Attached
        24:31 |  Reserved
        """
 
        if bitmask == 0: # Port is disabled by Acroname
            return "Disabled"
        if bitmask == 11: # Port is enabled but no device was detected
            return "Disconnected"
        if bitmask > 100: # Normally we hope it will cover "Device Attached" use cases (Note, can also be turn on when 'USB Error Flag' is on...but we havn't seen that )
            return "OK"
        return "Unknown Error ({})".format( bitmask )
 
 
    def enable_ports(self,  ports = None, disable_other_ports = False, sleep_on_change = 0 ):
        """
        Set enable state to provided ports
        :param ports: List of port numbers; if not provided, enable all ports
        :param disable_other_ports: if True, the ports not in the list will be disabled
        :param sleep_on_change: Number of seconds to sleep if any change is made
        :return: True if no errors found, False otherwise
        """
        result = True
        changed = False
        for port in self.all_ports():
            #
            if ports is None or port in ports:
                if not self.is_port_enabled( port ):
                    #log.d( "enabling port", port)
                    action_result = self.hub.usb.setPortEnable( port )
                    if action_result != brainstem.result.Result.NO_ERROR:
                        result = False
                        log.e("Failed to enable port", port)
                    else:
                        changed = True
            #
            elif disable_other_ports:
                if self.is_port_enabled( port ):
                    #log.d("disabling port", port)
                    action_result = self.hub.usb.setPortDisable( port )
                    if action_result != brainstem.result.Result.NO_ERROR:
                        result = False
                    else:
                        changed = True
        #
        if changed and sleep_on_change:
            import time
            time.sleep( sleep_on_change )
        #
        return result
 
 
    def disable_ports(self,  ports = None, sleep_on_change = 0 ):
        """
        :param ports: List of port numbers; if not provided, disable all ports
        :param sleep_on_change: Number of seconds to sleep if any change is made
        :return: True if no errors found, False otherwise
        """
        result = True
        changed = False
        for port in self.all_ports():
            if ports is None or port in ports:
                if self.is_port_enabled( port ):
                        #log.d("disabling port", port)
                        action_result = self.hub.usb.setPortDisable( port )
                        if action_result != brainstem.result.Result.NO_ERROR:
                            result = False
                            log.e("Failed to disable port", port)
                        else:
                            changed = True
        if changed and sleep_on_change:
            import time
            time.sleep( sleep_on_change )
        #
        return result
 
 
    def set_ports_usb2(self,  portlist = None, timeout = 100e-3 ):
        """
        Set USB ports to USB2
        """
        if portlist is None:
            portlist = self.ports()
        #
        self.recycle_ports( portlist, timeout = timeout )
        #
        for port in portlist:
            self.hub.usb.setSuperSpeedDataEnable( port )
            self.hub.usb.setHiSpeedDataEnable( port )
            self.hub.usb.setSuperSpeedDataDisable( port )
 
 
    def set_ports_usb3(self,  portlist = None, timeout = 100e-3 ):
        """
        Set USB ports to support USB3
        """
        if portlist is None:
            portlist = self.ports()
        #
        self.recycle_ports( portlist, timeout = timeout )
        #
        for port in portlist:
            self.hub.usb.setSuperSpeedDataEnable( port )
            self.hub.usb.setHiSpeedDataEnable( port )
            self.hub.usb.setHiSpeedDataDisable( port )
 
 
    def _port_power(self,  port ):
        """
        """
        if port < 0 or port > 7:
            raise ValueError("port number can be only within 0 and 7 (inclusive)")
        #
        micro_volt = self.hub.usb.getPortVoltage( port )
        micro_curr = self.hub.usb.getPortCurrent( port )
        volt = float(micro_volt.value) / 10.0 ** 6
        amps = float(micro_curr.value) / 10.0 ** 6
        #
        return volt * amps
 
    if 'windows' in platform.system().lower():
        def get_port_by_location(self, usb_location):
            """
            """
            if usb_location:
                #
                # T265 locations look differently...
                match = re.fullmatch(r'Port_#(\d+)\.Hub_#(\d+)', usb_location, re.IGNORECASE)
                if match:
                    # We don't know how to get the port from these yet!
                    return None  # int(match.group(2))
                else:
                    split_location = [int(x) for x in usb_location.split('.')]
                    # lambda helper to return the last 2 non-zero numbers, used when connecting using an additional hub
                    # ex: laptop -> hub -> acroname
                    get_last_two_digits = lambda array: tuple(
                        reversed(list(reversed([i for i in array if i != 0]))[:2]))
                    # only the last two digits are necessary
                    first_index, second_index = get_last_two_digits(split_location)
 
                    return get_port_from_usb(first_index, second_index)
    else:
 
        def get_port_by_location(self, usb_location):
            """
            """
            if not self.all_hubs:
                self.all_hubs = set(device_hub.find_all_hubs('24ff'))  # 24ff is Acroname VendorID
            if usb_location:
                #
                # Devices connected through an acroname will be in one of two sub-hubs under the acroname main
                # hub. Each is a 4-port hub with a different port (4 for ports 0-3, 3 for ports 4-7):
                #     /:  Bus 02.Port 1: Dev 1, Class=root_hub, Driver=xhci_hcd/6p, 10000M
                #         |__ Port 2: Dev 2, If 0, Class=Hub, Driver=hub/4p, 5000M                       <--- ACRONAME
                #             |__ Port 3: Dev 3, If 0, Class=Hub, Driver=hub/4p, 5000M
                #                 |__ Port X: Dev, If...
                #                 |__ Port Y: ...
                #             |__ Port 4: Dev 4, If 0, Class=Hub, Driver=hub/4p, 5000M
                #                 |__ Port Z: ...
                # (above is output from 'lsusb -t')
                # For the above acroname at '2-2' (bus 2, port 2), there are at least 3 devices:
                #     2-2.3.X
                #     2-2.3.Y
                #     2-2.4.Z
                # Given the two sub-ports (3.X, 3.Y, 4.Z), we can get the port number.
                # NOTE: some of our devices are hubs themselves! For example, the SR300 will show as '2-2.3.2.1' --
                # we must start a known hub or else the ports we look at are meaningless...
                #
                for port in self.all_hubs:
                    if usb_location.startswith(port + '.'):
                        match = re.search(r'^(\d+)\.(\d+)', usb_location[len(port) + 1:])
                        if match:
                            return get_port_from_usb(int(match.group(1)), int(match.group(2)))
 
specs = None
def discover(retries = 0):
    """
    Return all Acroname module specs in a list. Raise NoneFoundError if one is not found!
    """
    global specs
    # see https://acroname.com/reference/_modules/brainstem/module.html#Module.discoverAndConnect
    if specs is None:
        log.d('discovering Acroname modules ...')
        try:
            log.debug_indent()
            for i in range(retries + 1):
                specs = brainstem.discover.findAllModules(brainstem.link.Spec.USB)
                if not specs:
                    time.sleep(1)
                else:
                    for spec in specs:
                        log.d( '...', spec )
                    break
            if not specs:
                raise NoneFoundError()
        finally:
            log.debug_unindent()
 
    return specs
 
 
def get_port_from_usb(first_usb_index, second_usb_index ):
    """
    Based on last two USB location index, provide the port number
    """
    acroname_port_usb_map = {(4, 4): 0,
                             (4, 3): 1,
                             (4, 2): 2,
                             (4, 1): 3,
                             (3, 4): 4,
                             (3, 3): 5,
                             (3, 2): 6,
                             (3, 1): 7,
                             }
    return acroname_port_usb_map[(first_usb_index, second_usb_index)]
 
 
 
if __name__ == '__main__':
    acroname = Acroname()
    for opt,arg in opts:
        if opt in ('--enable'):
            acroname.connect()
            acroname.enable_ports()   # so ports() will return all
        elif opt in ('--disable'):
            acroname.connect()
            acroname.disable_ports()
        elif opt in ('--recycle'):
            acroname.connect()
            acroname.enable_ports()   # so ports() will return all
            acroname.recycle_ports()
        elif opt in ('--reset'):
            acroname.connect( reset = True )