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/python/index.html
"""

from rspy import log


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( '        --recycle      Recycle all ports' )
        sys.exit(2)
    try:
        opts,args = getopt.getopt( sys.argv[1:], '',
            longopts = [ 'help', 'recycle', 'enable' ])
    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.w( 'No acroname library is available!' )
    raise

hub = None


class NoneFoundError( RuntimeError ):
    """
    """
    def __init__( self, message = None ):
        super().__init__( self, message  or  'no Acroname module found' )


def discover():
    """
    Return all Acroname module specs in a list. Raise NoneFoundError if one is not found!
    """

    log.d( 'discovering Acroname modules ...' )
    # see https://acroname.com/reference/_modules/brainstem/module.html#Module.discoverAndConnect
    try:
        log.debug_indent()
        specs = brainstem.discover.findAllModules( brainstem.link.Spec.USB )
        if not specs:
            raise NoneFoundError()
        for spec in specs:
            log.d( '...', spec )
    finally:
        log.debug_unindent()

    return specs


def connect( spec = None ):
    """
    Connect to the hub. Raises RuntimeError on failure
    """

    global hub
    if not hub:
        hub = brainstem.stem.USBHub3p()

    if spec:
        specs = [spec]
    else:
        specs = discover()
        spec = specs[0]

    result = 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 )


def find_all_hubs():
    """
    Yields all hub port numbers
    """
    from rspy import lsusb
    #
    # 24ff:8013 = 
    #   iManufacturer           Acroname Inc.
    #   iProduct                USBHub3p-3[A]
    hubs = set( lsusb.devices_by_vendor( '24ff' ))
    ports = set()
    # go thru the tree and find only the top-level ones (which we should encounter first)
    for dev,dev_port in lsusb.tree():
        if dev not in hubs:
            continue
        for port in ports:
            # ignore everything inside existing hubs - we only want the top-level
            if dev_port.startswith( port + '.' ):
                break
        else:
            ports.add( dev_port )
            yield dev_port


def is_connected():
    try:
        connect()
        return True
    except Exception:
        return False


def disconnect():
    global hub
    hub.disconnect()
    del hub
    hub = None


def all_ports():
    """
    :return: a list of all possible ports, even if currently unoccupied or disabled
    """
    return range(8)


def ports():
    """
    :return: a list of all ports currently occupied (and enabled)
    """
    occupied_ports = []
    for port in all_ports():
        if port_power( port ) > 0.0:
            occupied_ports.append( port )
    return occupied_ports


def is_port_enabled( port ):
    return port_state( port ) == "OK"


def port_state( port ):
    if port < 0 or port > 7:
        raise ValueError( "port number must be [0-7]" )
    #
    global hub
    status = hub.usb.getPortState( port )
    #
    if status.value == 0:
        return "Disabled"
    if status.value == 11:
        return "Disconnected"
    if status.value > 100:
        return "OK"
    return "Unknown Error ({})".format( status.value )


def enable_ports( 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
    """
    global hub
    result = True
    changed = False
    for port in all_ports():
        #
        if ports is None or port in ports:
            if not is_port_enabled( port ):
                action_result = hub.usb.setPortEnable( port )
                if action_result != brainstem.result.Result.NO_ERROR:
                    result = False
                else:
                    changed = True
        #
        elif disable_other_ports:
            if is_port_enabled( port ):
                action_result = 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( ports ):
    """
    :param ports: List of port numbers
    :return: True if no errors found, False otherwise
    """
    global hub
    result = True
    for port in ports:
        #
        action_result = hub.usb.setPortDisable( port )
        if action_result != brainstem.result.Result.NO_ERROR:
            result = False
    #
    return result


def recycle_ports( portlist = None, timeout = 2 ):
    """
    Disable and enable a port
    :param timeout: how long to wait before re-enabling
    :return: True if everything OK, False otherwise
    """
    if portlist is None:
        portlist = ports()
    #
    result = disable_ports( portlist )
    #
    import time
    time.sleep( timeout )
    #
    result = enable_ports( portlist ) and result
    #
    return result


def set_ports_usb2( portlist = None, timeout = 100e-3 ):
    """
    Set USB ports to USB2
    """
    if portlist is None:
        portlist = ports()
    #
    recycle_ports( portlist, timeout = timeout )
    #
    global hub
    for port in portlist:
        hub.usb.setSuperSpeedDataEnable( port )
        hub.usb.setHiSpeedDataEnable( port )
        hub.usb.setSuperSpeedDataDisable( port )


def set_ports_usb3( portlist = None, timeout = 100e-3 ):
    """
    Set USB ports to support USB3
    """
    if portlist is None:
        portlist = ports()
    #
    recycle_ports( portlist, timeout = timeout )
    #
    global hub
    for port in portlist:
        hub.usb.setSuperSpeedDataEnable( port )
        hub.usb.setHiSpeedDataEnable( port )
        hub.usb.setHiSpeedDataDisable( port )


def port_power( port ):
    """
    """
    if port < 0 or port > 7:
        raise ValueError("port number can be only within 0 and 7 (inclusive)")
    #
    global hub
    micro_volt = hub.usb.getPortVoltage( port )
    micro_curr = hub.usb.getPortCurrent( port )
    volt = float(micro_volt.value) / 10.0 ** 6
    amps = float(micro_curr.value) / 10.0 ** 6
    #
    return volt * amps


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__':
    for opt,arg in opts:
        if opt in ('--enable'):
            connect()
            enable_ports()   # so ports() will return all
        elif opt in ('--recycle'):
            connect()
            enable_ports()   # so ports() will return all
            recycle_ports()