From 0ae395cff892c75ce4ef35d484f8808d03474027 Mon Sep 17 00:00:00 2001 From: Richard Alpe Date: Tue, 6 May 2025 16:31:43 +0200 Subject: [PATCH] hack: refactor software from cli-pretty to show Signed-off-by: Richard Alpe --- buildroot | 2 +- src/show/common.py | 45 +++++++++ src/show/modules/system.py | 111 ++++++++++++++++++++++ src/show/show.py | 77 +++++---------- src/show/sysrepocfg.py | 59 ++++++++++++ src/statd/python/cli_pretty/cli_pretty.py | 86 ----------------- 6 files changed, 239 insertions(+), 141 deletions(-) create mode 100644 src/show/common.py create mode 100644 src/show/modules/system.py create mode 100644 src/show/sysrepocfg.py diff --git a/buildroot b/buildroot index 33d2ca71..74fc86b0 160000 --- a/buildroot +++ b/buildroot @@ -1 +1 @@ -Subproject commit 33d2ca716fa9fdc4d4b1a187d68190f1d36ab60d +Subproject commit 74fc86b09970288bf5610831ffb11a1dd2cf446d diff --git a/src/show/common.py b/src/show/common.py new file mode 100644 index 00000000..fcd74163 --- /dev/null +++ b/src/show/common.py @@ -0,0 +1,45 @@ +class JsonData: + @staticmethod + def get(default, indata, *args): + data = indata + for arg in args: + if arg in data: + data = data.get(arg) + else: + return default + + return data + + +class Decore(): + @staticmethod + def decorate(sgr, txt, restore="0"): + return f"\033[{sgr}m{txt}\033[{restore}m" + + @staticmethod + def invert(txt): + return Decore.decorate("7", txt) + + @staticmethod + def bold(txt): + return Decore.decorate("1", txt) + + @staticmethod + def red(txt): + return Decore.decorate("31", txt, "39") + + @staticmethod + def green(txt): + return Decore.decorate("32", txt, "39") + + @staticmethod + def yellow(txt): + return Decore.decorate("33", txt, "39") + + @staticmethod + def underline(txt): + return Decore.decorate("4", txt, "24") + + @staticmethod + def gray_bg(txt): + return Decore.decorate("100", txt) diff --git a/src/show/modules/system.py b/src/show/modules/system.py new file mode 100644 index 00000000..ae94743a --- /dev/null +++ b/src/show/modules/system.py @@ -0,0 +1,111 @@ +import json + +from common import Decore +from common import JsonData +from dataclasses import dataclass + +class System: + def __init__(self, cfg): + self.services = self.Services(cfg) + self.software = self.Software(cfg) + + class Services: + def __init__(self, cfg): + self.cfg = cfg + + def get(self): + return self.cfg.run("/ietf-system:system-state/infix-system:services") + + def pretty(self): + data = self.get() + return data if data else {} + + class Software: + class Slot: + @dataclass + class Pad: + name: int = 10 + state: int = 10 + version: int = 23 + date: int = 25 + hash: int = 64 + + def __init__(self, data, version, date): # TODO: why is slot version and date on top level? + self.pad = self.Pad() + self.name = data.get('bootname', '') + self.size = data.get('size', '') + self.type = data.get('class', '') + self.hash = data.get('sha256', '') + self.state = data.get('state', '') + self.version = version # TODO + self.date = version # TODO + + def print_detailed(self): + """Detailed information about one bundle""" + print(f"Name : {self.name}") + print(f"State : {self.state}") + print(f"Version : {self.version}") + print(f"Size : {self.size}") + print(f"SHA-256 : {self.hash}") + print(f"Installed : {self.date}") + + @classmethod + def print_hdr(cls): + pad = cls.Pad() + hdr = (f"{'NAME':<{pad.name}}" + f"{'STATE':<{pad.state}}" + f"{'VERSION':<{pad.version}}" + f"{'DATE':<{pad.date}}") + print(Decore.invert(hdr)) + + def print_slot_row(self): + if self.type != "rootfs": + return + + """Brief information about one bundle""" + row = f"{self.name:<{self.pad.name}}" + row += f"{self.state:<{self.pad.state}}" + row += f"{self.version:<{self.pad.version}}" + row += f"{self.date:<{self.pad.date}}" + print(row) + + + def __init__(self, cfg): + self.cfg = cfg + self.data = self.get() + + self.slots = self.data.get("slot", []) + self.version = JsonData.get('', self.data, 'bundle', 'version') + self.date = JsonData.get('', self.data, 'installed', 'datetime') + self.boot_order = self.data.get("boot-order", ["Unknown"]) + + def get(self): + return self.cfg.run("/ietf-system:system-state/infix-system:software") + + def pprint(self): + software = self.data + + print(Decore.invert("BOOT ORDER")) + order = " ".join(boot.strip() for boot in self.boot_order) + print(f"{order}\n") + + self.Slot.print_hdr() + for _slot in reversed(self.slots): + slot = self.Slot(_slot, self.version, self.date) + slot.print_slot_row() + + def pprint_slot(self, name): + match = next((slot for slot in self.slots if slot.get("bootname") == name), None) + if match: + slot = self.Slot(match, self.version, self.date) + slot.print_detailed() + + def show(self, args): + if self.cfg.raw_output: + print(json.dumps(self.data, indent=2)) + return + + if args: + self.pprint_slot(args[0]) + else: + self.pprint() diff --git a/src/show/show.py b/src/show/show.py index 6416a7ef..0061082b 100755 --- a/src/show/show.py +++ b/src/show/show.py @@ -1,35 +1,17 @@ #!/usr/bin/env python3 import re -import subprocess import json from typing import List import os -import shlex import argparse +import subprocess + +from sysrepocfg import Sysrepocfg +from modules.system import System RAW_OUTPUT = False -def run_sysrepocfg(xpath: str) -> dict: - if not isinstance(xpath, str) or not xpath.startswith("/"): - print("Invalid XPATH. It must be a valid string starting with '/'.") - return {} - - safe_xpath = shlex.quote(xpath) - - try: - result = subprocess.run([ - "sysrepocfg", "-f", "json", "-X", "-d", "operational", "-x", safe_xpath - ], capture_output=True, text=True, check=True) - json_data = json.loads(result.stdout) - return json_data - except subprocess.CalledProcessError as e: - print(f"Error running sysrepocfg: {e}") - return {} - except json.JSONDecodeError as e: - print(f"Error parsing JSON output: {e}") - return {} - def cli_pretty(json_data: dict, command: str, *args: str): if not command or not all(isinstance(arg, str) for arg in args): print("Invalid command or arguments. All arguments must be strings.") @@ -47,7 +29,7 @@ def cli_pretty(json_data: dict, command: str, *args: str): print(f"Error running cli-pretty: {e}") def dhcp(args: List[str]) -> None: - data = run_sysrepocfg("/infix-dhcp-server:dhcp-server") + data = Sysrepocfg.run("/infix-dhcp-server:dhcp-server") if not data: print("No interface data retrieved.") return @@ -58,7 +40,7 @@ def dhcp(args: List[str]) -> None: cli_pretty(data, "show-dhcp-server") def hardware(args: List[str]) -> None: - data = run_sysrepocfg("/ietf-hardware:hardware") + data = Sysrepocfg.run("/ietf-hardware:hardware") if not data: print("No hardware data retrieved.") return @@ -69,7 +51,7 @@ def hardware(args: List[str]) -> None: cli_pretty(data, "show-hardware") def ntp(args: List[str]) -> None: - data = run_sysrepocfg("/system-state/ntp") + data = Sysrepocfg.run("/system-state/ntp") if not data: print("No ntp data retrieved.") return @@ -90,7 +72,7 @@ def is_valid_interface_name(interface_name: str) -> bool: return bool(re.match(pattern, interface_name)) def interface(args: List[str]) -> None: - data = run_sysrepocfg("/ietf-interfaces:interfaces") + data = Sysrepocfg.run("/ietf-interfaces:interfaces") if not data: print("No interface data retrieved.") return @@ -111,7 +93,7 @@ def interface(args: List[str]) -> None: print("Too many arguments provided. Only one interface name is expected.") def stp(args: List[str]) -> None: - data = run_sysrepocfg("/ietf-interfaces:interfaces") + data = Sysrepocfg.run("/ietf-interfaces:interfaces") if not data: print("No interface data retrieved.") return @@ -121,31 +103,14 @@ def stp(args: List[str]) -> None: return cli_pretty(data, "show-bridge-stp") -def software(args: List[str]) -> None: - data = run_sysrepocfg("/ietf-system:system-state/infix-system:software") - if not data: - print("No software data retrieved.") - return - - if RAW_OUTPUT: - print(json.dumps(data, indent=2)) - return - if len(args) == 0 or not args[0]: # Treat "" as no arg. - cli_pretty(data, "show-software") - elif len(args) == 1: - name = args[0] - if name not in ("primary", "secondary"): - print(f"Invalid software name: {name}") - return - - cli_pretty(data, f"show-software", "-n", name) - else: - print("Too many arguments provided. Only one name is expected.") +def software(cfg : Sysrepocfg, args: List[str]) -> None: + system = System(cfg) + system.software.show(args) def routes(args: List[str]): ip_version = args[0] if args and args[0] in ["ipv4", "ipv6"] else "ipv4" - data = run_sysrepocfg("/ietf-routing:routing/ribs") + data = Sysrepocfg.run("/ietf-routing:routing/ribs") if not data: print("No route data retrieved.") return @@ -155,14 +120,14 @@ def routes(args: List[str]): cli_pretty(data, "show-routing-table", "-i", ip_version) -def execute_command(command: str, args: List[str]): +def execute_legacy_non_class_command(command: str, args: List[str]): + command_mapping = { 'dhcp': dhcp, 'hardware': hardware, 'interface': interface, 'ntp': ntp, 'routes': routes, - 'software' : software, 'stp': stp, } @@ -171,7 +136,6 @@ def execute_command(command: str, args: List[str]): else: print(f"Unknown command: {command}") - def main(): global RAW_OUTPUT @@ -179,11 +143,16 @@ def main(): parser.add_argument('command', help="Command to execute") parser.add_argument('args', nargs=argparse.REMAINDER, help="Additional arguments for the command") parser.add_argument('-r', '--raw', action='store_true', help="Print raw JSON output from Sysrepo") - + parser.add_argument( "-f", "--datafile", default=None, type=str, help="Path to the input JSON data file") args = parser.parse_args() - RAW_OUTPUT = args.raw - execute_command(args.command, args.args) + RAW_OUTPUT = args.raw + cfg = Sysrepocfg(args.datafile, args.raw) + + if args.command == "software": + software(cfg, args.args) + else: + execute_legacy_non_class_command(args.command, args.args) if __name__ == "__main__": diff --git a/src/show/sysrepocfg.py b/src/show/sysrepocfg.py new file mode 100644 index 00000000..12c20216 --- /dev/null +++ b/src/show/sysrepocfg.py @@ -0,0 +1,59 @@ +import shlex +import sys +import subprocess +import json + +class Sysrepocfg: + def __init__(self, data_file=None, raw=False): + self.data_file = data_file + self.raw_output = raw + + def _run(self, safe_xpath): + try: + result = subprocess.run([ + "sysrepocfg", "-f", "json", "-X", "-d", "operational", "-x", safe_xpath + ], capture_output=True, text=True, check=True) + data = json.loads(result.stdout) + if not data: + sys.exit(f"Error, no data from sysrepo") + return data + except subprocess.CalledProcessError as e: + sys.exit(f"Error, running sysrepocfg: {e}") + except json.JSONDecodeError as e: + sys.exit(f"Error, parsing JSON output: {e}") + + + def _run_emulator(self, xpath): + """ + Emulates `sysrepocfg -X -x ` by extracting a subtree from the JSON file. + + :param json_file: Path to the local JSON file containing sysrepo-style data. + :param xpath: YANG-style XPath (e.g., '/ietf-interfaces:interfaces'). + :return: Extracted subtree or None if not found. + """ + with open(self.data_file, 'r') as f: + data = json.load(f) + + path_parts = xpath.strip('/').split('/') + + node = data + for part in path_parts: + if isinstance(node, dict) and part in node: + node = node[part] + else: + return None # Path doesn't exist + + return node + + + def run(self, xpath: str) -> dict: + if not isinstance(xpath, str) or not xpath.startswith("/"): + print("Invalid XPATH. It must be a valid string starting with '/'.") + return {} + + safe_xpath = shlex.quote(xpath) + + if self.data_file is not None: + return self._run_emulator(safe_xpath) + + return self._run(safe_xpath) diff --git a/src/statd/python/cli_pretty/cli_pretty.py b/src/statd/python/cli_pretty/cli_pretty.py index 85f04961..e2c9639e 100755 --- a/src/statd/python/cli_pretty/cli_pretty.py +++ b/src/statd/python/cli_pretty/cli_pretty.py @@ -49,14 +49,6 @@ class PadRoute: else: raise ValueError(f"unknown IP version: {ipv}") - -class PadSoftware: - name = 10 - date = 25 - hash = 64 - state = 10 - version = 23 - class PadDhcpServer: ip = 17 mac = 19 @@ -287,40 +279,6 @@ class Route: row += f"{hop:<{PadRoute.next_hop}} " print(row) - -class Software: - """Software bundle class """ - def __init__(self, data): - self.data = data - self.name = data.get('bootname', '') - self.size = data.get('size', '') - self.type = data.get('class', '') - self.hash = data.get('sha256', '') - self.state = data.get('state', '') - self.version = get_json_data('', self.data, 'bundle', 'version') - self.date = get_json_data('', self.data, 'installed', 'datetime') - - def is_rootfs(self): - """True if bundle type is 'rootfs'""" - return self.type == "rootfs" - - def print(self): - """Brief information about one bundle""" - row = f"{self.name:<{PadSoftware.name}}" - row += f"{self.state:<{PadSoftware.state}}" - row += f"{self.version:<{PadSoftware.version}}" - row += f"{self.date:<{PadSoftware.date}}" - print(row) - - def detail(self): - """Detailed information about one bundle""" - print(f"Name : {self.name}") - print(f"State : {self.state}") - print(f"Version : {self.version}") - print(f"Size : {self.size}") - print(f"SHA-256 : {self.hash}") - print(f"Installed : {self.date}") - class USBport: def __init__(self, data): self.data = data @@ -1159,45 +1117,6 @@ def show_routing_table(json, ip): route = Route(r, ip) route.print() - -def find_slot(_slots, name): - for _slot in [Software(data) for data in _slots]: - if _slot.name == name: - return _slot - - return False - - -def show_software(json, name): - if not json.get("ietf-system:system-state", "infix-system:software"): - print("Error, cannot find infix-system:software") - sys.exit(1) - - software = get_json_data({}, json, 'ietf-system:system-state', 'infix-system:software') - slots = software.get("slot") - boot_order = software.get("boot-order", ["Unknown"]) - if name: - slot = find_slot(slots, name) - if slot: - slot.detail() - else: - print(Decore.invert("BOOT ORDER")) - order="" - for boot in boot_order: - order+=f"{boot.strip()} " - print(order) - print("") - - hdr = (f"{'NAME':<{PadSoftware.name}}" - f"{'STATE':<{PadSoftware.state}}" - f"{'VERSION':<{PadSoftware.version}}" - f"{'DATE':<{PadSoftware.date}}") - print(Decore.invert(hdr)) - for _s in reversed(slots): - slot = Software(_s) - if slot.is_rootfs(): - slot.print() - def show_hardware(json): if not json.get("ietf-hardware:hardware"): print(f"Error, top level \"ietf-hardware:component\" missing") @@ -1282,9 +1201,6 @@ def main(): parser_show_bridge_stp = subparsers.add_parser('show-bridge-stp', help='Show spanning tree state') - parser_show_software = subparsers.add_parser('show-software', help='Show software versions') - parser_show_software.add_argument('-n', '--name', help='Slotname') - parser_show_hardware = subparsers.add_parser('show-hardware', help='Show USB ports') parser_show_ntp_sources = subparsers.add_parser('show-ntp', help='Show NTP sources') @@ -1301,8 +1217,6 @@ def main(): show_interfaces(json_data, args.name) elif args.command == "show-routing-table": show_routing_table(json_data, args.ip) - elif args.command == "show-software": - show_software(json_data, args.name) elif args.command == "show-bridge-mdb": show_bridge_mdb(json_data) elif args.command == "show-bridge-stp":