| #!/usr/bin/env python3 | |
| # SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause | |
| # | |
| # pylint: disable=too-many-locals, too-many-branches, too-many-statements | |
| # pylint: disable=too-many-return-statements | |
| """ YNL ethtool utility """ | |
| import argparse | |
| import pathlib | |
| import pprint | |
| import sys | |
| import re | |
| import os | |
| # pylint: disable=no-name-in-module,wrong-import-position | |
| sys.path.append(pathlib.Path(__file__).resolve().parent.parent.joinpath('pyynl').as_posix()) | |
| # pylint: disable=import-error | |
| from cli import schema_dir, spec_dir | |
| from lib import YnlFamily | |
| def args_to_req(ynl, op_name, args, req): | |
| """ | |
| Verify and convert command-line arguments to the ynl-compatible request. | |
| """ | |
| valid_attrs = ynl.operation_do_attributes(op_name) | |
| valid_attrs.remove('header') # not user-provided | |
| if len(args) == 0: | |
| print(f'no attributes, expected: {valid_attrs}') | |
| sys.exit(1) | |
| i = 0 | |
| while i < len(args): | |
| attr = args[i] | |
| if i + 1 >= len(args): | |
| print(f'expected value for \'{attr}\'') | |
| sys.exit(1) | |
| if attr not in valid_attrs: | |
| print(f'invalid attribute \'{attr}\', expected: {valid_attrs}') | |
| sys.exit(1) | |
| val = args[i+1] | |
| i += 2 | |
| req[attr] = val | |
| def print_field(reply, *desc): | |
| """ | |
| Pretty-print a set of fields from the reply. desc specifies the | |
| fields and the optional type (bool/yn). | |
| """ | |
| if not reply: | |
| return | |
| if len(desc) == 0: | |
| print_field(reply, *zip(reply.keys(), reply.keys())) | |
| return | |
| for spec in desc: | |
| try: | |
| field, name, tp = spec | |
| except ValueError: | |
| field, name = spec | |
| tp = 'int' | |
| value = reply.get(field, None) | |
| if tp == 'yn': | |
| value = 'yes' if value else 'no' | |
| elif tp == 'bool' or isinstance(value, bool): | |
| value = 'on' if value else 'off' | |
| else: | |
| value = 'n/a' if value is None else value | |
| print(f'{name}: {value}') | |
| def print_speed(name, value): | |
| """ | |
| Print out the speed-like strings from the value dict. | |
| """ | |
| speed_re = re.compile(r'[0-9]+base[^/]+/.+') | |
| speed = [ k for k, v in value.items() if v and speed_re.match(k) ] | |
| print(f'{name}: {" ".join(speed)}') | |
| def do_set(ynl, args, op_name): | |
| """ | |
| Prepare request header, parse arguments and do a set operation. | |
| """ | |
| req = { | |
| 'header': { | |
| 'dev-name': args.device, | |
| }, | |
| } | |
| args_to_req(ynl, op_name, args.args, req) | |
| ynl.do(op_name, req) | |
| def do_get(ynl, args, op_name, extra=None): | |
| """ | |
| Prepare request header and get info for a specific device using doit. | |
| """ | |
| extra = extra or {} | |
| req = {'header': {'dev-name': args.device}} | |
| req['header'].update(extra.pop('header', {})) | |
| req.update(extra) | |
| reply = ynl.do(op_name, req) | |
| if not reply: | |
| return {} | |
| if args.json: | |
| pprint.PrettyPrinter().pprint(reply) | |
| sys.exit(0) | |
| reply.pop('header', None) | |
| return reply | |
| def bits_to_dict(attr): | |
| """ | |
| Convert ynl-formatted bitmask to a dict of bit=value. | |
| """ | |
| ret = {} | |
| if 'bits' not in attr: | |
| return {} | |
| if 'bit' not in attr['bits']: | |
| return {} | |
| for bit in attr['bits']['bit']: | |
| if bit['name'] == '': | |
| continue | |
| name = bit['name'] | |
| value = bit.get('value', False) | |
| ret[name] = value | |
| return ret | |
| def main(): | |
| """ YNL ethtool utility """ | |
| parser = argparse.ArgumentParser(description='ethtool wannabe') | |
| parser.add_argument('--json', action=argparse.BooleanOptionalAction) | |
| parser.add_argument('--show-priv-flags', action=argparse.BooleanOptionalAction) | |
| parser.add_argument('--set-priv-flags', action=argparse.BooleanOptionalAction) | |
| parser.add_argument('--show-eee', action=argparse.BooleanOptionalAction) | |
| parser.add_argument('--set-eee', action=argparse.BooleanOptionalAction) | |
| parser.add_argument('-a', '--show-pause', action=argparse.BooleanOptionalAction) | |
| parser.add_argument('-A', '--set-pause', action=argparse.BooleanOptionalAction) | |
| parser.add_argument('-c', '--show-coalesce', action=argparse.BooleanOptionalAction) | |
| parser.add_argument('-C', '--set-coalesce', action=argparse.BooleanOptionalAction) | |
| parser.add_argument('-g', '--show-ring', action=argparse.BooleanOptionalAction) | |
| parser.add_argument('-G', '--set-ring', action=argparse.BooleanOptionalAction) | |
| parser.add_argument('-k', '--show-features', action=argparse.BooleanOptionalAction) | |
| parser.add_argument('-K', '--set-features', action=argparse.BooleanOptionalAction) | |
| parser.add_argument('-l', '--show-channels', action=argparse.BooleanOptionalAction) | |
| parser.add_argument('-L', '--set-channels', action=argparse.BooleanOptionalAction) | |
| parser.add_argument('-T', '--show-time-stamping', action=argparse.BooleanOptionalAction) | |
| parser.add_argument('-S', '--statistics', action=argparse.BooleanOptionalAction) | |
| # TODO: --show-tunnels tunnel-info-get | |
| # TODO: --show-module module-get | |
| # TODO: --get-plca-cfg plca-get | |
| # TODO: --get-plca-status plca-get-status | |
| # TODO: --show-mm mm-get | |
| # TODO: --show-fec fec-get | |
| # TODO: --dump-module-eerpom module-eeprom-get | |
| # TODO: pse-get | |
| # TODO: rss-get | |
| parser.add_argument('device', metavar='device', type=str) | |
| parser.add_argument('args', metavar='args', type=str, nargs='*') | |
| dbg_group = parser.add_argument_group('Debug options') | |
| dbg_group.add_argument('--dbg-small-recv', default=0, const=4000, | |
| action='store', nargs='?', type=int, metavar='INT', | |
| help="Length of buffers used for recv()") | |
| args = parser.parse_args() | |
| spec = os.path.join(spec_dir(), 'ethtool.yaml') | |
| schema = os.path.join(schema_dir(), 'genetlink-legacy.yaml') | |
| ynl = YnlFamily(spec, schema, recv_size=args.dbg_small_recv) | |
| if args.dbg_small_recv: | |
| ynl.set_recv_dbg(True) | |
| if args.set_priv_flags: | |
| # TODO: parse the bitmask | |
| print("not implemented") | |
| return | |
| if args.set_eee: | |
| do_set(ynl, args, 'eee-set') | |
| return | |
| if args.set_pause: | |
| do_set(ynl, args, 'pause-set') | |
| return | |
| if args.set_coalesce: | |
| do_set(ynl, args, 'coalesce-set') | |
| return | |
| if args.set_features: | |
| # TODO: parse the bitmask | |
| print("not implemented") | |
| return | |
| if args.set_channels: | |
| do_set(ynl, args, 'channels-set') | |
| return | |
| if args.set_ring: | |
| do_set(ynl, args, 'rings-set') | |
| return | |
| if args.show_priv_flags: | |
| flags = bits_to_dict(do_get(ynl, args, 'privflags-get')['flags']) | |
| print_field(flags) | |
| return | |
| if args.show_eee: | |
| eee = do_get(ynl, args, 'eee-get') | |
| ours = bits_to_dict(eee['modes-ours']) | |
| peer = bits_to_dict(eee['modes-peer']) | |
| if 'enabled' in eee: | |
| status = 'enabled' if eee['enabled'] else 'disabled' | |
| if 'active' in eee and eee['active']: | |
| status = status + ' - active' | |
| else: | |
| status = status + ' - inactive' | |
| else: | |
| status = 'not supported' | |
| print(f'EEE status: {status}') | |
| print_field(eee, ('tx-lpi-timer', 'Tx LPI')) | |
| print_speed('Advertised EEE link modes', ours) | |
| print_speed('Link partner advertised EEE link modes', peer) | |
| return | |
| if args.show_pause: | |
| print_field(do_get(ynl, args, 'pause-get'), | |
| ('autoneg', 'Autonegotiate', 'bool'), | |
| ('rx', 'RX', 'bool'), | |
| ('tx', 'TX', 'bool')) | |
| return | |
| if args.show_coalesce: | |
| print_field(do_get(ynl, args, 'coalesce-get')) | |
| return | |
| if args.show_features: | |
| reply = do_get(ynl, args, 'features-get') | |
| available = bits_to_dict(reply['hw']) | |
| requested = bits_to_dict(reply['wanted']).keys() | |
| active = bits_to_dict(reply['active']).keys() | |
| never_changed = bits_to_dict(reply['nochange']).keys() | |
| for f in sorted(available): | |
| value = "off" | |
| if f in active: | |
| value = "on" | |
| fixed = "" | |
| if f not in available or f in never_changed: | |
| fixed = " [fixed]" | |
| req = "" | |
| if f in requested: | |
| if f in active: | |
| req = " [requested on]" | |
| else: | |
| req = " [requested off]" | |
| print(f'{f}: {value}{fixed}{req}') | |
| return | |
| if args.show_channels: | |
| reply = do_get(ynl, args, 'channels-get') | |
| print(f'Channel parameters for {args.device}:') | |
| print('Pre-set maximums:') | |
| print_field(reply, | |
| ('rx-max', 'RX'), | |
| ('tx-max', 'TX'), | |
| ('other-max', 'Other'), | |
| ('combined-max', 'Combined')) | |
| print('Current hardware settings:') | |
| print_field(reply, | |
| ('rx-count', 'RX'), | |
| ('tx-count', 'TX'), | |
| ('other-count', 'Other'), | |
| ('combined-count', 'Combined')) | |
| return | |
| if args.show_ring: | |
| reply = do_get(ynl, args, 'channels-get') | |
| print(f'Ring parameters for {args.device}:') | |
| print('Pre-set maximums:') | |
| print_field(reply, | |
| ('rx-max', 'RX'), | |
| ('rx-mini-max', 'RX Mini'), | |
| ('rx-jumbo-max', 'RX Jumbo'), | |
| ('tx-max', 'TX')) | |
| print('Current hardware settings:') | |
| print_field(reply, | |
| ('rx', 'RX'), | |
| ('rx-mini', 'RX Mini'), | |
| ('rx-jumbo', 'RX Jumbo'), | |
| ('tx', 'TX')) | |
| print_field(reply, | |
| ('rx-buf-len', 'RX Buf Len'), | |
| ('cqe-size', 'CQE Size'), | |
| ('tx-push', 'TX Push', 'bool')) | |
| return | |
| if args.statistics: | |
| print('NIC statistics:') | |
| # TODO: pass id? | |
| strset = do_get(ynl, args, 'strset-get') | |
| pprint.PrettyPrinter().pprint(strset) | |
| req = { | |
| 'groups': { | |
| 'size': 1, | |
| 'bits': { | |
| 'bit': | |
| # TODO: support passing the bitmask | |
| #[ | |
| #{ 'name': 'eth-phy', 'value': True }, | |
| { 'name': 'eth-mac', 'value': True }, | |
| #{ 'name': 'eth-ctrl', 'value': True }, | |
| #{ 'name': 'rmon', 'value': True }, | |
| #], | |
| }, | |
| }, | |
| } | |
| rsp = do_get(ynl, args, 'stats-get', req) | |
| pprint.PrettyPrinter().pprint(rsp) | |
| return | |
| if args.show_time_stamping: | |
| req = { | |
| 'header': { | |
| 'flags': 'stats', | |
| }, | |
| } | |
| tsinfo = do_get(ynl, args, 'tsinfo-get', req) | |
| print(f'Time stamping parameters for {args.device}:') | |
| print('Capabilities:') | |
| _ = [print(f'\t{v}') for v in bits_to_dict(tsinfo['timestamping'])] | |
| print(f'PTP Hardware Clock: {tsinfo.get("phc-index", "none")}') | |
| if 'tx-types' in tsinfo: | |
| print('Hardware Transmit Timestamp Modes:') | |
| _ = [print(f'\t{v}') for v in bits_to_dict(tsinfo['tx-types'])] | |
| else: | |
| print('Hardware Transmit Timestamp Modes: none') | |
| if 'rx-filters' in tsinfo: | |
| print('Hardware Receive Filter Modes:') | |
| _ = [print(f'\t{v}') for v in bits_to_dict(tsinfo['rx-filters'])] | |
| else: | |
| print('Hardware Receive Filter Modes: none') | |
| if 'stats' in tsinfo and tsinfo['stats']: | |
| print('Statistics:') | |
| _ = [print(f'\t{k}: {v}') for k, v in tsinfo['stats'].items()] | |
| return | |
| print(f'Settings for {args.device}:') | |
| linkmodes = do_get(ynl, args, 'linkmodes-get') | |
| ours = bits_to_dict(linkmodes['ours']) | |
| supported_ports = ('TP', 'AUI', 'BNC', 'MII', 'FIBRE', 'Backplane') | |
| ports = [ p for p in supported_ports if ours.get(p, False)] | |
| print(f'Supported ports: [ {" ".join(ports)} ]') | |
| print_speed('Supported link modes', ours) | |
| print_field(ours, ('Pause', 'Supported pause frame use', 'yn')) | |
| print_field(ours, ('Autoneg', 'Supports auto-negotiation', 'yn')) | |
| supported_fec = ('None', 'PS', 'BASER', 'LLRS') | |
| fec = [ p for p in supported_fec if ours.get(p, False)] | |
| fec_str = " ".join(fec) | |
| if len(fec) == 0: | |
| fec_str = "Not reported" | |
| print(f'Supported FEC modes: {fec_str}') | |
| speed = 'Unknown!' | |
| if linkmodes['speed'] > 0 and linkmodes['speed'] < 0xffffffff: | |
| speed = f'{linkmodes["speed"]}Mb/s' | |
| print(f'Speed: {speed}') | |
| duplex_modes = { | |
| 0: 'Half', | |
| 1: 'Full', | |
| } | |
| duplex = duplex_modes.get(linkmodes["duplex"], None) | |
| if not duplex: | |
| duplex = f'Unknown! ({linkmodes["duplex"]})' | |
| print(f'Duplex: {duplex}') | |
| autoneg = "off" | |
| if linkmodes.get("autoneg", 0) != 0: | |
| autoneg = "on" | |
| print(f'Auto-negotiation: {autoneg}') | |
| ports = { | |
| 0: 'Twisted Pair', | |
| 1: 'AUI', | |
| 2: 'MII', | |
| 3: 'FIBRE', | |
| 4: 'BNC', | |
| 5: 'Directly Attached Copper', | |
| 0xef: 'None', | |
| } | |
| linkinfo = do_get(ynl, args, 'linkinfo-get') | |
| print(f'Port: {ports.get(linkinfo["port"], "Other")}') | |
| print_field(linkinfo, ('phyaddr', 'PHYAD')) | |
| transceiver = { | |
| 0: 'Internal', | |
| 1: 'External', | |
| } | |
| print(f'Transceiver: {transceiver.get(linkinfo["transceiver"], "Unknown")}') | |
| mdix_ctrl = { | |
| 1: 'off', | |
| 2: 'on', | |
| } | |
| mdix = mdix_ctrl.get(linkinfo['tp-mdix-ctrl'], None) | |
| if mdix: | |
| mdix = mdix + ' (forced)' | |
| else: | |
| mdix = mdix_ctrl.get(linkinfo['tp-mdix'], 'Unknown (auto)') | |
| print(f'MDI-X: {mdix}') | |
| debug = do_get(ynl, args, 'debug-get') | |
| msgmask = bits_to_dict(debug.get("msgmask", [])).keys() | |
| print(f'Current message level: {" ".join(msgmask)}') | |
| linkstate = do_get(ynl, args, 'linkstate-get') | |
| detected_states = { | |
| 0: 'no', | |
| 1: 'yes', | |
| } | |
| # TODO: wol-get | |
| detected = detected_states.get(linkstate['link'], 'unknown') | |
| print(f'Link detected: {detected}') | |
| if __name__ == '__main__': | |
| main() |