hack: refactor software from cli-pretty to show

Signed-off-by: Richard Alpe <richard@bit42.se>

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)

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()

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__":

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 <xpath>` 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)

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":