forked from sonic-net/sonic-buildimage
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[ CLI ]Refactor show interface transceiver cli command (sonic-net#471)
- Loading branch information
Showing
6 changed files
with
380 additions
and
3 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,219 @@ | ||
| #!/usr/bin/python | ||
| """ | ||
| Script to show sfp eeprom and presence status. | ||
| Not like sfputil this scripts get the sfp data from DB directly. | ||
| """ | ||
| import argparse | ||
| import json | ||
| import sys | ||
| import click | ||
| import re | ||
| import operator | ||
| import os | ||
|
|
||
| from natsort import natsorted | ||
| from swsssdk import SonicV2Connector, port_util | ||
| from tabulate import tabulate | ||
|
|
||
| # Mock the redis for unit test purposes # | ||
| try: | ||
| if os.environ["UTILITIES_UNIT_TESTING"] == "1": | ||
| modules_path = os.path.join(os.path.dirname(__file__), "..") | ||
| test_path = os.path.join(modules_path, "sonic-utilities-tests") | ||
| sys.path.insert(0, modules_path) | ||
| sys.path.insert(0, test_path) | ||
| import mock_tables.dbconnector | ||
| except KeyError: | ||
| pass | ||
|
|
||
| qsfp_data_map = {'modelname': 'Vendor PN', 'vendor_oui': 'Vendor OUI', | ||
| 'vendor_date': 'Vendor Date Code(YYYY-MM-DD Lot)', | ||
| 'manufacturename': 'Vendor Name', | ||
| 'hardwarerev': 'Vendor Rev', 'serialnum': 'Vendor SN', | ||
| 'type': 'Identifier', 'ext_identifier': 'Extended Identifier', | ||
| 'ext_rateselect_compliance': 'Extended RateSelect Compliance', | ||
| 'cable_length': 'cable_length', 'cable_type': 'Length', | ||
| 'nominal_bit_rate': 'Nominal Bit Rate(100Mbs)', | ||
| 'specification_compliance':'Specification compliance', | ||
| 'encoding': 'Encoding', 'Connector': 'Connector' | ||
| } | ||
|
|
||
| sfp_dom_channel_monitor_map = {'rx1power': 'RXPower', 'tx1bias': 'TXBias', 'tx1power': 'TXPower'} | ||
|
|
||
| qsfp_dom_channel_monitor_map = {'rx1power': 'RX1Power', 'rx2power': 'RX2Power', | ||
| 'rx3power': 'RX3Power', 'rx4power': 'RX4Power', | ||
| 'tx1bias': 'TX1Bias', 'tx2bias': 'TX2Bias', | ||
| 'tx3bias': 'TX3Bias', 'tx4bias': 'TX4Bias', | ||
| 'tx1power': 'TX1Power', 'tx2power': 'TX2Power', | ||
| 'tx3power': 'TX3Power', 'tx4power': 'TX4Power'} | ||
|
|
||
| dom_module_monitor_map = {'temperature': 'Temperature', 'voltage': 'Vcc'} | ||
|
|
||
| dom_value_unit_map = {'rx1power': 'dBm', 'rx2power': 'dBm', | ||
| 'rx3power': 'dBm', 'rx4power': 'dBm', | ||
| 'tx1bias': 'mA', 'tx2bias': 'mA', | ||
| 'tx3bias': 'mA', 'tx4bias': 'mA', | ||
| 'tx1power': 'dBm', 'tx2power': 'dBm', | ||
| 'tx3power': 'dBm', 'tx4power': 'dBm', | ||
| 'temperature': 'C', 'voltage': 'Volts'} | ||
|
|
||
|
|
||
|
|
||
| class SFPShow(object): | ||
|
|
||
| def __init__(self): | ||
| super(SFPShow,self).__init__() | ||
| self.adb = SonicV2Connector(host="127.0.0.1") | ||
| self.adb.connect(self.adb.APPL_DB) | ||
|
|
||
| self.sdb = SonicV2Connector(host="127.0.0.1") | ||
| self.sdb.connect(self.sdb.STATE_DB) | ||
| return | ||
|
|
||
| # Convert channel monitor values to cli output string | ||
| def convert_channel_monitor_value_to_output_string(self, sorted_key_table, dom_info_dict, channel_monitor_map): | ||
| out_put='' | ||
| ident = ' ' | ||
| for key in sorted_key_table: | ||
| if dom_info_dict[key] != 'N/A': | ||
| if dom_info_dict[key] == 'Unknown': | ||
| out_put = out_put + ident + ident + channel_monitor_map[key] + ': ' + dom_info_dict[key] + '\n' | ||
| else: | ||
| out_put = out_put + ident + ident + channel_monitor_map[key] + ': ' + dom_info_dict[key] + dom_value_unit_map[key] + '\n' | ||
| return out_put | ||
|
|
||
| # Convert dom sensor info in DB to cli output string | ||
| def convert_dom_to_output_string(self, sfp_type, dom_info_dict): | ||
| ident = ' ' | ||
| out_put_dom = '' | ||
| if sfp_type.startswith('QSFP'): | ||
| out_put_dom = out_put_dom + ident + 'ChannelMonitorValues' + ': ' + '\n' | ||
| sorted_dom_channel_monitor_info_key_table = natsorted(qsfp_dom_channel_monitor_map) | ||
| out_put_channel = self.convert_channel_monitor_value_to_output_string(sorted_dom_channel_monitor_info_key_table, dom_info_dict, qsfp_dom_channel_monitor_map) | ||
| out_put_dom = out_put_dom + out_put_channel | ||
|
|
||
| out_put_dom = out_put_dom + ident + 'ModuleMonitorValues' + ': ' + '\n' | ||
|
|
||
| else: | ||
| out_put_dom = out_put_dom + ident + 'MonitorData' + ': ' + '\n' | ||
| sorted_dom_channel_monitor_info_key_table = natsorted(sfp_dom_channel_monitor_map) | ||
| out_put_channel = self.convert_channel_monitor_value_to_output_string(sorted_dom_channel_monitor_info_key_table, dom_info_dict, sfp_dom_channel_monitor_map) | ||
| out_put_dom = out_put_dom + out_put_channel | ||
|
|
||
| sorted_dom_module_monitor_info_key_table = natsorted(dom_module_monitor_map) | ||
| out_put_module = self.convert_channel_monitor_value_to_output_string(sorted_dom_module_monitor_info_key_table, dom_info_dict, dom_module_monitor_map) | ||
| out_put_dom = out_put_dom + out_put_module | ||
|
|
||
| return out_put_dom | ||
|
|
||
| # Convert sfp info in DB to cli output string | ||
| def convert_sfp_info_to_output_string(self, sfp_info_dict): | ||
| ident = ' ' | ||
| out_put = '' | ||
|
|
||
| sorted_qsfp_data_map = sorted(qsfp_data_map.items(), key=operator.itemgetter(1)) | ||
| for key in sorted_qsfp_data_map: | ||
| key1 = key[0] | ||
| if key1 == 'cable_type': | ||
| out_put = out_put + ident + sfp_info_dict['cable_type'] + ': ' + sfp_info_dict['cable_length'] + '\n' | ||
| elif key1 == 'cable_length': | ||
| pass | ||
| elif key1 == 'specification_compliance': | ||
| spefic_compliance_dict = eval(sfp_info_dict['specification_compliance']) | ||
| sorted_compliance_key_table = natsorted(spefic_compliance_dict) | ||
| out_put = out_put + ident + qsfp_data_map['specification_compliance'] + ': ' + '\n' | ||
| for compliance_key in sorted_compliance_key_table: | ||
| out_put = out_put + ident + ident + compliance_key + ': ' + spefic_compliance_dict[compliance_key] + '\n' | ||
| else: | ||
| out_put = out_put + ident + qsfp_data_map[key1] + ': ' + sfp_info_dict[key1] + '\n' | ||
|
|
||
| return out_put | ||
|
|
||
| # Convert sfp info and dom sensor info in DB to cli output string | ||
| def convert_interface_sfp_info_to_cli_output_string(self, state_db, interface_name, dump_dom): | ||
| out_put = '' | ||
| sfp_info_dict = state_db.get_all(state_db.STATE_DB, 'TRANSCEIVER_INFO|{}'.format(interface_name)) | ||
| out_put = interface_name + ': ' + 'SFP EEPROM detected' + '\n' | ||
| sfp_info_output = self.convert_sfp_info_to_output_string(sfp_info_dict) | ||
| out_put = out_put + sfp_info_output | ||
|
|
||
| if dump_dom: | ||
| sfp_type = sfp_info_dict['type'] | ||
| dom_info_dict = state_db.get_all(state_db.STATE_DB, 'TRANSCEIVER_DOM_SENSOR|{}'.format(interface_name)) | ||
| dom_output = self.convert_dom_to_output_string(sfp_type, dom_info_dict) | ||
| out_put = out_put + dom_output | ||
|
|
||
| return out_put | ||
|
|
||
| def display_eeprom(self, interfacename, dump_dom): | ||
| out_put = '' | ||
|
|
||
| if interfacename is not None: | ||
| presence = self.sdb.exists(self.sdb.STATE_DB, 'TRANSCEIVER_INFO|{}'.format(interfacename)) | ||
| if presence: | ||
| out_put = self.convert_interface_sfp_info_to_cli_output_string(self.sdb, interfacename, dump_dom) | ||
| else: | ||
| out_put = out_put + interfacename + ': ' + 'SFP EEPROM Not detected' + '\n' | ||
| else: | ||
| port_table_keys = self.adb.keys(self.adb.APPL_DB, "PORT_TABLE:*") | ||
| sorted_table_keys = natsorted(port_table_keys) | ||
| for i in sorted_table_keys: | ||
| interface = re.split(':', i, maxsplit=1)[-1].strip() | ||
| if interface and interface.startswith('Ethernet'): | ||
| presence = self.sdb.exists(self.sdb.STATE_DB, 'TRANSCEIVER_INFO|{}'.format(interface)) | ||
| if presence: | ||
| out_put = out_put + self.convert_interface_sfp_info_to_cli_output_string(self.sdb, interface, dump_dom) | ||
| else: | ||
| out_put = out_put + interface + ': ' + 'SFP EEPROM Not detected' + '\n' | ||
|
|
||
| out_put = out_put + '\n' | ||
|
|
||
| print out_put | ||
|
|
||
| def display_presence(self, interfacename): | ||
| port_table = [] | ||
| header = ['Port', 'Presence'] | ||
|
|
||
| if interfacename is not None: | ||
| presence = self.sdb.exists(self.sdb.STATE_DB, 'TRANSCEIVER_INFO|{}'.format(interfacename)) | ||
| if presence: | ||
| port_table.append((interfacename, 'Present')) | ||
| else: | ||
| port_table.append((interfacename, 'Not present')) | ||
| else: | ||
| port_table_keys = self.adb.keys(self.adb.APPL_DB, "PORT_TABLE:*") | ||
| for i in port_table_keys: | ||
| key = re.split(':', i, maxsplit=1)[-1].strip() | ||
| if key and key.startswith('Ethernet'): | ||
| presence = self.sdb.exists(self.sdb.STATE_DB, 'TRANSCEIVER_INFO|{}'.format(key)) | ||
| if presence: | ||
| port_table.append((key,'Present')) | ||
| else: | ||
| port_table.append((key,'Not present')) | ||
|
|
||
| sorted_port_table = natsorted(port_table) | ||
| click.echo(tabulate(sorted_port_table, header)) | ||
|
|
||
| # This is our main entrypoint - the main 'sfpshow' command | ||
| @click.group() | ||
| def cli(): | ||
| """sfpshow - Command line utility for display SFP transceivers information""" | ||
| pass | ||
|
|
||
| # 'eeprom' subcommand | ||
| @cli.command() | ||
| @click.option('-p', '--port', metavar='<port_name>', help="Display SFP EEPROM data for port <port_name> only") | ||
| @click.option('-d', '--dom', 'dump_dom', is_flag=True, help="Also display Digital Optical Monitoring (DOM) data") | ||
| def eeprom(port, dump_dom): | ||
| sfp = SFPShow() | ||
| sfp.display_eeprom(port, dump_dom) | ||
|
|
||
| # 'presence' subcommand | ||
| @cli.command() | ||
| @click.option('-p', '--port', metavar='<port_name>', help="Display SFP presence for port <port_name> only") | ||
| def presence(port): | ||
| sfp = SFPShow() | ||
| sfp.display_presence(port) | ||
|
|
||
| if __name__ == "__main__": | ||
| cli() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1 +1,22 @@ | ||
| {} | ||
| { | ||
| "PORT_TABLE:Ethernet0": { | ||
| "lanes": "0", | ||
| "alias": "Ethernet0", | ||
| "description": "ARISTA01T2:Ethernet1", | ||
| "speed": "25000", | ||
| "oper_status": "down", | ||
| "pfc_asym": "off", | ||
| "mtu": "9100", | ||
| "admin_status": "up" | ||
| }, | ||
| "PORT_TABLE:Ethernet200": { | ||
| "lanes": "200,201,202,203", | ||
| "alias": "Ethernet200", | ||
| "description": "Ethernet200", | ||
| "speed": "100000", | ||
| "oper_status": "down", | ||
| "fec": "rs", | ||
| "mtu": "9100", | ||
| "pfc_asym": "off" | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.