[show][config] add support for setting and displaying switching modes…

… on Y cable (#1501)

Summary
There is also support for new commands for muxcable/Y cable
show muxcable hwmode switchmode
sudo config muxcable hwmode setswitchmode manual/auto all
which can basically set/display the switching modes on the cable to auto or manual
using the cli

What I did
Also added these new commands

admin@STR43-0101-0101-01LT0:~$ show muxcable hwmode switchmode 
Port       Switching
---------  -----------
Ethernet0  manual
Ethernet4  manual
admin@STR43-0101-0101-01LT0:/usr$ sudo config muxcable hwmode setswitchmode manual all
Muxcable at port all will be changed to manual switching mode. Continue? [y/N]: y
Success in switching mode on port Ethernet0 to manual
Success in switching mode on port Ethernet4 to manual
admin@STR43-0101-0101-01LT0:/usr$ sudo config muxcable hwmode setswitchmode auto all
Muxcable at port all will be changed to auto switching mode. Continue? [y/N]: y
Success in switching mode on port Ethernet0 to auto
Success in switching mode on port Ethernet4 to auto

Signed-off-by: vaibhav-dahiya <[email protected]>

config/muxcable.py

@@ -429,7 +429,7 @@ def state(state, port):
 
             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 
+            """ 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"""
@@ -470,3 +470,164 @@ def state(state, port):
         if rc == False:
             click.echo("ERR: Unable to toggle one or more ports to {}".format(state))
             sys.exit(CONFIG_FAIL)
+
+
+@hwmode.command()
+@click.argument('state', metavar='<operation_status>', required=True, type=click.Choice(["auto", "manual"]))
+@click.argument('port', metavar='<port_name>', required=True, default=None)
+def setswitchmode(state, port):
+    """Configure the muxcable mux switching mode {auto/manual}"""
+
+    per_npu_statedb = {}
+    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)
+
+    if port is not None and port != "all":
+        click.confirm(('Muxcable at port {} will be changed to {} switching mode. 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 not re.match(VENDOR_MODEL_REGEX, model_value):
+            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)
+
+        if state == "auto":
+            mode = sonic_y_cable.y_cable.SWITCHING_MODE_AUTO
+        elif state == "manual":
+            mode = sonic_y_cable.y_cable.SWITCHING_MODE_MANUAL
+        import sonic_y_cable.y_cable
+        result = sonic_y_cable.y_cable.set_switching_mode(physical_port, mode)
+        if result == False:
+            click.echo(("ERR: Unable to set switching mode for the cable port {}".format(port)))
+            sys.exit(CONFIG_FAIL)
+
+        click.echo("Success in switching mode on port {} to {}".format(port, state))
+
+    elif port == "all" and port is not None:
+
+        click.confirm(('Muxcable at port {} will be changed to {} switching mode. Continue?'.format(port, 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 not re.match(VENDOR_MODEL_REGEX, model_value):
+                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
+
+            if state == "auto":
+                mode = sonic_y_cable.y_cable.SWITCHING_MODE_AUTO
+            elif state == "manual":
+                mode = sonic_y_cable.y_cable.SWITCHING_MODE_MANUAL
+            import sonic_y_cable.y_cable
+            result = sonic_y_cable.y_cable.set_switching_mode(physical_port, mode)
+            if result == False:
+                rc = False
+                click.echo("ERR: Unable to set switching mode on port {} to {}".format(port, state))
+
+            click.echo("Success in switching mode on port {} to {}".format(port, state))
+
+        if rc == False:
+            click.echo("ERR: Unable to set switching mode one or more ports to {}".format(state))
+            sys.exit(CONFIG_FAIL)

show/muxcable.py

@@ -511,7 +511,7 @@ def muxdirection(port):
 
         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 
+        """ 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"""
@@ -594,7 +594,7 @@ def muxdirection(port):
 
             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 
+            """ 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"""
@@ -644,3 +644,179 @@ def muxdirection(port):
         click.echo(tabulate(body, headers=headers))
         if rc == False:
             sys.exit(EXIT_FAIL)
+
+
+@hwmode.command()
+@click.argument('port', metavar='<port_name>', required=False, default=None)
+def switchmode(port):
+    """Shows the current switching mode of the muxcable {auto/manual}"""
+
+    per_npu_statedb = {}
+    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)
+
+    if port is not None:
+
+        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(EXIT_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)
+
+        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 not re.match(VENDOR_MODEL_REGEX, model_value):
+            click.echo("ERR: Got invalid vendor value and model for port {}".format(port))
+            sys.exit(EXIT_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(EXIT_FAIL)
+        if len(physical_port_list) != 1:
+            click.echo("ERR: Found multiple physical ports ({}) associated with {}".format(
+                ", ".join(physical_port_list), port))
+            sys.exit(EXIT_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(EXIT_FAIL)
+
+        import sonic_y_cable.y_cable
+        switching_mode = sonic_y_cable.y_cable.get_switching_mode(physical_port)
+        if switching_mode == -1:
+            click.echo(("ERR: Unable to get switching mode for the cable port {}".format(port)))
+            sys.exit(EXIT_FAIL)
+
+        if switching_mode == sonic_y_cable.y_cable.SWITCHING_MODE_AUTO:
+            state = "auto"
+        elif switching_mode == sonic_y_cable.y_cable.SWITCHING_MODE_MANUAL:
+            state = "manual"
+        else:
+            click.echo(("ERR: Unable to get switching state mode, port {}".format(port)))
+            state = "unknown"
+        headers = ['Port', 'Switching']
+
+        body = [[port, state]]
+        click.echo(tabulate(body, headers=headers))
+
+    else:
+
+        logical_port_list = platform_sfputil_helper.get_logical_list()
+
+        rc = True
+        body = []
+        for port in logical_port_list:
+
+            temp_list = []
+            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):
+                continue
+            if len(physical_port_list) != 1:
+                continue
+
+            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:
+                    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 not re.match(VENDOR_MODEL_REGEX, model_value):
+                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
+            switching_mode = sonic_y_cable.y_cable.get_switching_mode(physical_port)
+            if switching_mode == -1:
+                rc = False
+                temp_list.append(port)
+                temp_list.append("unknown")
+                body.append(temp_list)
+                continue
+
+            if switching_mode == sonic_y_cable.y_cable.SWITCHING_MODE_AUTO:
+                state = "auto"
+            elif switching_mode == sonic_y_cable.y_cable.SWITCHING_MODE_MANUAL:
+                state = "manual"
+            else:
+                rc = False
+                temp_list.append(port)
+                temp_list.append("unknown")
+                body.append(temp_list)
+                continue
+            temp_list.append(port)
+            temp_list.append(state)
+            body.append(temp_list)
+
+        headers = ['Port', 'Switching']
+
+        click.echo(tabulate(body, headers=headers))
+        if rc == False:
+            sys.exit(EXIT_FAIL)