[show][config] fix show mux status health field; add support for hwmode functionality to toggle mux, check mux direction for Y cable (#1467)

This PR fixes the show mux status "health" field.

There is also support for new commands for muxcable
show muxcable hwmode muxdirection
sudo config muxcable hwmode state active all
which can check the muxdirection and toggle the mux direction directly through eeprom

What I did
Fixed the show mux status health field
Also added these new commands

vdahiya@BN9-0101-0301-01UT0:~$ show muxcable hwmode muxdirection 
Port         Direction
-----------  -----------
Ethernet0    active
Ethernet4    active
Ethernet8    active
Ethernet12   active
Ethernet16   active
vdahiya@BN9-0101-0301-01UT0:~$ sudo config muxcable hwmode state active all
Muxcables at all ports will be changed to active state. Continue? [y/N]: y
Success in toggling port Ethernet0 to active
Success in toggling port Ethernet4 to active
Success in toggling port Ethernet8 to active
Success in toggling port Ethernet12 to active
Success in toggling port Ethernet16 to active
Signed-off-by: vaibhav-dahiya <[email protected]>

Author: vdahiya12

config/muxcable.py

@@ -16,6 +16,8 @@
 CONFIG_SUCCESSFUL = 100
 CONFIG_FAIL = 1
 
+VENDOR_NAME = "Credo"
+VENDOR_MODEL = "CAC125321P2PA0MS"
 
 # Helper functions
 
@@ -146,7 +148,7 @@ def mode(state, port, json_output):
 
         if per_npu_statedb[asic_index] is not None:
             y_cable_asic_table_keys = port_table_keys[asic_index]
-            logical_key = "MUX_CABLE_TABLE"+"|"+port
+            logical_key = "MUX_CABLE_TABLE|{}".format(port)
             if logical_key in y_cable_asic_table_keys:
                 port_status_dict = {}
                 lookup_statedb_and_update_configdb(
@@ -188,11 +190,13 @@ def mode(state, port, json_output):
 
         sys.exit(CONFIG_SUCCESSFUL)
 
+
 @muxcable.group(cls=clicommon.AbbreviationGroup)
 def prbs():
     """Enable/disable PRBS mode on a port"""
     pass
 
+
 @prbs.command()
 @click.argument('port', required=True, default=None, type=click.INT)
 @click.argument('target', required=True, default=None, type=click.INT)
@@ -209,6 +213,7 @@ def enable(port, target, mode_value, lane_map):
     click.echo("PRBS config sucessful")
     sys.exit(CONFIG_SUCCESSFUL)
 
+
 @prbs.command()
 @click.argument('port', required=True, default=None, type=click.INT)
 @click.argument('target', required=True, default=None, type=click.INT)
@@ -223,6 +228,7 @@ def disable(port, target):
     click.echo("PRBS disable sucessful")
     sys.exit(CONFIG_SUCCESSFUL)
 
+
 @muxcable.group(cls=clicommon.AbbreviationGroup)
 def loopback():
     """Enable/disable loopback mode on a port"""
@@ -244,6 +250,7 @@ def enable(port, target, lane_map):
     click.echo("loopback config sucessful")
     sys.exit(CONFIG_SUCCESSFUL)
 
+
 @loopback.command()
 @click.argument('port', required=True, default=None, type=click.INT)
 @click.argument('target', required=True, default=None, type=click.INT)
@@ -257,3 +264,208 @@ def disable(port, target):
         sys.exit(CONFIG_FAIL)
     click.echo("loopback disable sucessful")
     sys.exit(CONFIG_SUCCESSFUL)
+
+
+@muxcable.group(cls=clicommon.AbbreviationGroup)
+def hwmode():
+    """Configure muxcable hardware directly"""
+    pass
+
+
+@hwmode.command()
+@click.argument('state', metavar='<operation_status>', required=True, type=click.Choice(["active", "standby"]))
+@click.argument('port', metavar='<port_name>', required=True, default=None)
+def state(state, port):
+    """Configure the muxcable mux state {active/standby}"""
+
+    per_npu_statedb = {}
+    transceiver_table_keys = {}
+    transceiver_dict = {}
+
+    # Getting all front asic namespace and correspding config and state DB connector
+
+    namespaces = multi_asic.get_front_end_namespaces()
+    for namespace in namespaces:
+        asic_id = multi_asic.get_asic_index_from_namespace(namespace)
+        per_npu_statedb[asic_id] = SonicV2Connector(use_unix_socket_path=False, namespace=namespace)
+        per_npu_statedb[asic_id].connect(per_npu_statedb[asic_id].STATE_DB)
+
+        transceiver_table_keys[asic_id] = per_npu_statedb[asic_id].keys(
+            per_npu_statedb[asic_id].STATE_DB, 'TRANSCEIVER_INFO|*')
+
+    if port is not None and port != "all":
+        click.confirm(('Muxcable at port {} will be changed to {} state. Continue?'.format(port, state)), abort=True)
+        logical_port_list = platform_sfputil_helper.get_logical_list()
+        if port not in logical_port_list:
+            click.echo("ERR: This is not a valid port, valid ports ({})".format(", ".join(logical_port_list)))
+            sys.exit(CONFIG_FAIL)
+
+        asic_index = None
+        if platform_sfputil is not None:
+            asic_index = platform_sfputil_helper.get_asic_id_for_logical_port(port)
+            if asic_index is None:
+                # TODO this import is only for unit test purposes, and should be removed once sonic_platform_base
+                # is fully mocked
+                import sonic_platform_base.sonic_sfp.sfputilhelper
+                asic_index = sonic_platform_base.sonic_sfp.sfputilhelper.SfpUtilHelper().get_asic_id_for_logical_port(port)
+                if asic_index is None:
+                    click.echo("Got invalid asic index for port {}, cant retreive mux status".format(port))
+                    sys.exit(CONFIG_FAIL)
+
+        if platform_sfputil is not None:
+            physical_port_list = platform_sfputil_helper.logical_port_name_to_physical_port_list(port)
+
+        if not isinstance(physical_port_list, list):
+            click.echo(("ERR: Unable to locate physical port information for {}".format(port)))
+            sys.exit(CONFIG_FAIL)
+        if len(physical_port_list) != 1:
+            click.echo("ERR: Found multiple physical ports ({}) associated with {}".format(
+                ", ".join(physical_port_list), port))
+            sys.exit(CONFIG_FAIL)
+
+        transceiver_dict[asic_index] = per_npu_statedb[asic_index].get_all(
+            per_npu_statedb[asic_index].STATE_DB, 'TRANSCEIVER_INFO|{}'.format(port))
+
+        vendor_value = get_value_for_key_in_dict(transceiver_dict[asic_index], port, "manufacturer", "TRANSCEIVER_INFO")
+        model_value = get_value_for_key_in_dict(transceiver_dict[asic_index], port, "model", "TRANSCEIVER_INFO")
+
+        """ This check is required for checking whether or not this port is connected to a Y cable
+        or not. The check gives a way to differentiate between non Y cable ports and Y cable ports.
+        TODO: this should be removed once their is support for multiple vendors on Y cable"""
+
+        if vendor_value != VENDOR_NAME or model_value != VENDOR_MODEL:
+            click.echo("ERR: Got invalid vendor value and model for port {}".format(port))
+            sys.exit(CONFIG_FAIL)
+
+        physical_port = physical_port_list[0]
+
+        logical_port_list_for_physical_port = platform_sfputil_helper.get_physical_to_logical()
+
+        logical_port_list_per_port = logical_port_list_for_physical_port.get(physical_port, None)
+
+        """ This check is required for checking whether or not this logical port is the one which is 
+        actually mapped to physical port and by convention it is always the first port.
+        TODO: this should be removed with more logic to check which logical port maps to actual physical port
+        being used"""
+
+        if port != logical_port_list_per_port[0]:
+            click.echo("ERR: This logical Port {} is not on a muxcable".format(port))
+            sys.exit(CONFIG_FAIL)
+
+        import sonic_y_cable.y_cable
+        read_side = sonic_y_cable.y_cable.check_read_side(physical_port)
+        if read_side == False or read_side == -1:
+            click.echo(("ERR: Unable to get read_side for the cable port {}".format(port)))
+            sys.exit(CONFIG_FAIL)
+
+        mux_direction = sonic_y_cable.y_cable.check_mux_direction(physical_port)
+        if mux_direction == False or mux_direction == -1:
+            click.echo(("ERR: Unable to get mux direction for the cable port {}".format(port)))
+            sys.exit(CONFIG_FAIL)
+
+        if int(read_side) == 1:
+            if state == "active":
+                res = sonic_y_cable.y_cable.toggle_mux_to_torA(physical_port)
+            elif state == "standby":
+                res = sonic_y_cable.y_cable.toggle_mux_to_torB(physical_port)
+            click.echo("Success in toggling port {} to {}".format(port, state))
+        elif int(read_side) == 2:
+            if state == "active":
+                res = sonic_y_cable.y_cable.toggle_mux_to_torB(physical_port)
+            elif state == "standby":
+                res = sonic_y_cable.y_cable.toggle_mux_to_torA(physical_port)
+            click.echo("Success in toggling port {} to {}".format(port, state))
+
+        if res == False:
+            click.echo("ERR: Unable to toggle port {} to {}".format(port, state))
+            sys.exit(CONFIG_FAIL)
+
+    elif port == "all" and port is not None:
+
+        click.confirm(('Muxcables at all ports will be changed to {} state. Continue?'.format(state)), abort=True)
+        logical_port_list = platform_sfputil_helper.get_logical_list()
+
+        rc = True
+        for port in logical_port_list:
+            if platform_sfputil is not None:
+                physical_port_list = platform_sfputil_helper.logical_port_name_to_physical_port_list(port)
+
+            asic_index = None
+            if platform_sfputil is not None:
+                asic_index = platform_sfputil_helper.get_asic_id_for_logical_port(port)
+                if asic_index is None:
+                    # TODO this import is only for unit test purposes, and should be removed once sonic_platform_base
+                    # is fully mocked
+                    import sonic_platform_base.sonic_sfp.sfputilhelper
+                    asic_index = sonic_platform_base.sonic_sfp.sfputilhelper.SfpUtilHelper().get_asic_id_for_logical_port(port)
+                    if asic_index is None:
+                        click.echo("Got invalid asic index for port {}, cant retreive mux status".format(port))
+
+            if not isinstance(physical_port_list, list):
+                click.echo(("ERR: Unable to locate physical port information for {}".format(port)))
+                continue
+
+            if len(physical_port_list) != 1:
+                click.echo("ERR: Found multiple physical ports ({}) associated with {}".format(
+                    ", ".join(physical_port_list), port))
+                continue
+
+            transceiver_dict[asic_index] = per_npu_statedb[asic_index].get_all(
+                per_npu_statedb[asic_index].STATE_DB, 'TRANSCEIVER_INFO|{}'.format(port))
+            vendor_value = transceiver_dict[asic_index].get("manufacturer", None)
+            model_value = transceiver_dict[asic_index].get("model", None)
+
+            """ This check is required for checking whether or not this port is connected to a Y cable
+            or not. The check gives a way to differentiate between non Y cable ports and Y cable ports.
+            TODO: this should be removed once their is support for multiple vendors on Y cable"""
+
+            if vendor_value != VENDOR_NAME or model_value != VENDOR_MODEL:
+                continue
+
+            physical_port = physical_port_list[0]
+
+            logical_port_list_for_physical_port = platform_sfputil_helper.get_physical_to_logical()
+
+            logical_port_list_per_port = logical_port_list_for_physical_port.get(physical_port, None)
+
+            """ This check is required for checking whether or not this logical port is the one which is 
+            actually mapped to physical port and by convention it is always the first port.
+            TODO: this should be removed with more logic to check which logical port maps to actual physical port
+            being used"""
+
+            if port != logical_port_list_per_port[0]:
+                continue
+
+            import sonic_y_cable.y_cable
+            read_side = sonic_y_cable.y_cable.check_read_side(physical_port)
+            if read_side == False or read_side == -1:
+                click.echo(("ERR: Unable to get read side for the cable port {}".format(port)))
+                rc = False
+                continue
+
+            mux_direction = sonic_y_cable.y_cable.check_mux_direction(physical_port)
+            if mux_direction == False or mux_direction == -1:
+                click.echo(("ERR: Unable to get mux direction for the cable port {}".format(port)))
+                rc = False
+                continue
+
+            if int(read_side) == 1:
+                if state == "active":
+                    res = sonic_y_cable.y_cable.toggle_mux_to_torA(physical_port)
+                elif state == "standby":
+                    res = sonic_y_cable.y_cable.toggle_mux_to_torB(physical_port)
+                click.echo("Success in toggling port {} to {}".format(port, state))
+            elif int(read_side) == 2:
+                if state == "active":
+                    res = sonic_y_cable.y_cable.toggle_mux_to_torB(physical_port)
+                elif state == "standby":
+                    res = sonic_y_cable.y_cable.toggle_mux_to_torA(physical_port)
+                click.echo("Success in toggling port {} to {}".format(port, state))
+
+            if res == False:
+                rc = False
+                click.echo("ERR: Unable to toggle port {} to {}".format(port, state))
+
+        if rc == False:
+            click.echo("ERR: Unable to toggle one or more ports to {}".format(state))
+            sys.exit(CONFIG_FAIL)