From caf9f2ada5f91ac152eccafe0b48e5ae222145b8 Mon Sep 17 00:00:00 2001 From: PawelPlesniak Date: Wed, 18 Mar 2026 12:13:34 +0100 Subject: [PATCH 1/4] Ready --- pyproject.toml | 1 + src/drunc/apps/check_np0x_hw_status.py | 269 +++++++++++++++++++++++++ 2 files changed, 270 insertions(+) create mode 100644 src/drunc/apps/check_np0x_hw_status.py diff --git a/pyproject.toml b/pyproject.toml index cafd7f66a..30fde922d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -50,6 +50,7 @@ application-registry-service = "drunc.apps.app_connectivity_server:main" drunc-ssh-doctor = "drunc.apps.ssh_doctor:main" drunc-setup-ssh-config = "drunc.apps.ssh_configurator:main" drunc-check-np0x-connections = "drunc.apps.check_np0x_connection_and_status:main" +drunc-check-np0x-hw = "drunc.apps.check_np0x_hw_status:main" [tool.setuptools.packages.find] diff --git a/src/drunc/apps/check_np0x_hw_status.py b/src/drunc/apps/check_np0x_hw_status.py new file mode 100644 index 000000000..faea30fb1 --- /dev/null +++ b/src/drunc/apps/check_np0x_hw_status.py @@ -0,0 +1,269 @@ +""" +Pings all the WIBs in the NP0x cryostats and displays their status (online vs offline). + +Currently only does WIBs, will be updated to FEMBs, ongoing discussions are in place +with Roger. + +Will do AMC crates and AMCs too, but later. +""" + +import subprocess +from concurrent.futures import ThreadPoolExecutor, as_completed +from datetime import datetime + +import pytz +from daqpytools.logging.formatter import timezone_name as tz +from rich import box +from rich.columns import Columns +from rich.console import Console +from rich.live import Live +from rich.table import Table + +# Data Mapping +WIB_DATA = { + "NP04 CB": { + "001": "10.73.137.20", + "002": "10.73.137.21", + "003": "10.73.137.22", + "004": "10.73.137.23", + "005": "10.73.137.24", + }, + "APA1": { + "101": "10.73.137.26", + "102": "10.73.137.27", + "103": "10.73.137.28", + "104": "10.73.137.29", + "105": "10.73.137.30", + }, + "APA2": { + "201": "10.73.137.31", + "202": "10.73.137.32", + "203": "10.73.137.33", + "204": "10.73.137.34", + "205": "10.73.137.35", + }, + "APA3": { + "301": "10.73.137.36", + "302": "10.73.137.37", + "303": "10.73.137.38", + "304": "10.73.137.39", + "305": "10.73.137.40", + }, + "APA4": { + "401": "10.73.137.41", + "402": "10.73.137.42", + "403": "10.73.137.43", + "404": "10.73.137.44", + "405": "10.73.137.45", + }, + "NP02 CB": { + "601": "10.73.137.50", + "602": "10.73.137.51", + "603": "10.73.137.52", + "604": "10.73.137.53", + "605": "10.73.137.54", + "606": "10.73.137.122", + }, + "CRP4": { + "1001": "10.73.137.126", + "1002": "10.73.137.127", + "1003": "10.73.137.128", + "1004": "10.73.137.137", + "1005": "10.73.137.129", + "1006": "10.73.137.130", + }, + "CRP5": { + "1101": "10.73.137.131", + "1102": "10.73.137.132", + "1103": "10.73.137.133", + "1104": "10.73.137.134", + "1105": "10.73.137.135", + "1106": "10.73.137.136", + }, +} + + +def ping_host(ip: str) -> bool: + """ + Pings a given IP address to check if it's online. + + Any exceptions that occur during the ping process will be caught and result in a + False return value. + + Args: + ip: The IP address to ping + + Returns: + True if the host is online (ping successful), False otherwise + + Raises: + None + """ + + # Construct the ping command. Ping once with a timeout of 1 second. + cmd = ["ping", "-c", "1", "-W", "1", ip] + + # Execute the ping command, suppressing output and errors. Use a timeout to avoid + # hanging on unresponsive hosts. + try: + result = subprocess.run( + cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=1.5 + ) + return result.returncode == 0 + except Exception: + return False + + +def make_wib_table( + category: str, wibs: dict[str, str], status_map: dict[str, bool | None] +) -> Table: + """ + Creates a Rich Table for a given category of WIBs, showing their online status and + address. + + Args: + category: The name of the category (e.g., "NP04 CB", "APA1", etc.). This is an + entry from the WIB_DATA mapping and will be used as the table title. + wibs: A mapping of WIB numbers to their IP addresses for this category + status_map: A mapping of IP addresses to their online status (True/False/None) + + Returns: + A Rich Table object representing the WIBs in this category and their status + + Raises: + None + """ + + # Create a Rich Table with a title based on the category. + table = Table(title=f"[magenta]{category}[/]", box=box.ROUNDED, border_style="dim") + + # Add columns for WIB number, address, and online status. + table.add_column("WIB #", justify="center") + table.add_column("Address", style="dim white") + table.add_column("Online?", justify="center") + + # For each relevant WIB, show the status. + for wib_num, ip in wibs.items(): + # Look up the status of this IP in the status map. + res = status_map.get(ip) + + # WIB is online + if res is True: + status = "[bold green]✔[/]" + + # WIB is offline + elif res is False: + status = "[bold red]✘[/]" + + # Status is unknown (ping not completed yet) + else: + status = "[yellow]...[/]" + + # Add a row to the table for this WIB with its number, IP address, and status. + table.add_row(wib_num, ip, status) + + return table + + +def generate_table(status_map: dict[str, bool | None]) -> Columns: + """ + Generates a Rich Columns object containing tables for each category of WIBs. + + Args: + status_map: A mapping of IP addresses to their online status + + Returns: + A Rich Columns object containing the WIB status tables for display + + Raises: + None + """ + + tables: list[Table] = [ + make_wib_table(cat, wibs, status_map) for cat, wibs in WIB_DATA.items() + ] + return Columns(tables, equal=True, expand=False) + + +def main() -> None: + """ + Main function to execute the NP0x HW status check and display results in a + live-updating table. + + This function initializes the console, parses out all the IP addresses, and pings + them concurrently while updating the display in real-time. It handles graceful + shutdown on keyboard interrupt and ensures that all threads are properly terminated. + + Args: + None + + Returns: + None + + Raises: + Any exceptions that occur during the execution of the host checks will be + handled within the get_host_info function. + """ + + # Initialize the console for Rich output + console = Console() + + # Get the timezone-aware current time for display in the header. + now = datetime.now(pytz.timezone(tz)) + + # Print a blank line for spacing and a header to indicate the start of the scan + console.print("\n") + console.print( + f"[bold cyan]Checking NP0x hardware status at {now.strftime('%Y-%m-%d %H:%M:%S %Z')}[/]\n" + ) + + # Extract all IPs from the WIB_DATA for processing + all_ips: list[str] = [ip for cat in WIB_DATA.values() for ip in cat.values()] + + # Initialize a results map to store the status of each IP, starting with None (unknown) + results_map: dict[str, bool | None] = {ip: None for ip in all_ips} + + # 1. Create executor outside a context manager to allow manual shutdown control + executor = ThreadPoolExecutor(max_workers=40) + futures = {executor.submit(ping_host, ip): ip for ip in all_ips} + + # Use Rich's Live to create a live-updating table. The table will be refreshed as + # results come in from the concurrent checks. + try: + with Live( + generate_table(results_map), console=console, auto_refresh=True + ) as live: + # While we have commands executing, we wait for them to complete. Use a + # timeout to periodically refresh the table and check for completed futures. + while futures: + try: + # Update the table with results as they come in. Remove completed + # futures from the tracking dict to avoid re-processing. + for future in as_completed(futures, timeout=0.1): + ip = futures.pop(future) + results_map[ip] = future.result() + live.update(generate_table(results_map)) + + # No futures have completed, but refresh the table to keep it "live" + except TimeoutError: + live.update(generate_table(results_map)) + + # If all futures are done, break out of the loop to finish up + if not futures: + break + + # Handle graceful shutdown on keyboard interrupt. This allows the user to stop the + # scan early without leaving hanging threads. + except KeyboardInterrupt: + pass + + # When exiting, all threads are properly shut down. + finally: + executor.shutdown(wait=False, cancel_futures=True) + + # Final display of results after all checks are complete. + console.print("\n[bold green]Scan complete.[/]") + + +if __name__ == "__main__": + main() From 22f7406473e4ec87a7563f58914c1cc9993e6d94 Mon Sep 17 00:00:00 2001 From: PawelPlesniak Date: Wed, 1 Apr 2026 19:42:21 +0200 Subject: [PATCH 2/4] Requiring documentation but functional --- src/drunc/apps/check_np0x_hw_status.py | 452 ++++++++++++++----------- 1 file changed, 257 insertions(+), 195 deletions(-) diff --git a/src/drunc/apps/check_np0x_hw_status.py b/src/drunc/apps/check_np0x_hw_status.py index faea30fb1..2bb553dfa 100644 --- a/src/drunc/apps/check_np0x_hw_status.py +++ b/src/drunc/apps/check_np0x_hw_status.py @@ -1,269 +1,331 @@ """ -Pings all the WIBs in the NP0x cryostats and displays their status (online vs offline). +Script to check the power status (on/off) of the NP0x readout hardware. -Currently only does WIBs, will be updated to FEMBs, ongoing discussions are in place -with Roger. +For WIBs, each device gets pinged. If reachable, attempt to query its FEMB power status. -Will do AMC crates and AMCs too, but later. +Including the AMCs is planned for the future. """ +import os import subprocess +import sys from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime import pytz -from daqpytools.logging.formatter import timezone_name as tz from rich import box from rich.columns import Columns from rich.console import Console from rich.live import Live from rich.table import Table -# Data Mapping -WIB_DATA = { - "NP04 CB": { - "001": "10.73.137.20", - "002": "10.73.137.21", - "003": "10.73.137.22", - "004": "10.73.137.23", - "005": "10.73.137.24", - }, - "APA1": { - "101": "10.73.137.26", - "102": "10.73.137.27", - "103": "10.73.137.28", - "104": "10.73.137.29", - "105": "10.73.137.30", - }, - "APA2": { - "201": "10.73.137.31", - "202": "10.73.137.32", - "203": "10.73.137.33", - "204": "10.73.137.34", - "205": "10.73.137.35", - }, - "APA3": { - "301": "10.73.137.36", - "302": "10.73.137.37", - "303": "10.73.137.38", - "304": "10.73.137.39", - "305": "10.73.137.40", - }, - "APA4": { - "401": "10.73.137.41", - "402": "10.73.137.42", - "403": "10.73.137.43", - "404": "10.73.137.44", - "405": "10.73.137.45", - }, - "NP02 CB": { - "601": "10.73.137.50", - "602": "10.73.137.51", - "603": "10.73.137.52", - "604": "10.73.137.53", - "605": "10.73.137.54", - "606": "10.73.137.122", - }, - "CRP4": { - "1001": "10.73.137.126", - "1002": "10.73.137.127", - "1003": "10.73.137.128", - "1004": "10.73.137.137", - "1005": "10.73.137.129", - "1006": "10.73.137.130", - }, - "CRP5": { - "1101": "10.73.137.131", - "1102": "10.73.137.132", - "1103": "10.73.137.133", - "1104": "10.73.137.134", - "1105": "10.73.137.135", - "1106": "10.73.137.136", - }, -} - -def ping_host(ip: str) -> bool: +# --- DYNAMIC PATH SETUP --- +def setup_wib_path(): """ - Pings a given IP address to check if it's online. + Sets up the path to the local copy of the dune-wib-firmware repository if it exists. - Any exceptions that occur during the ping process will be caught and result in a - False return value. + Searches for the dune-wib-firmware source directory in the work area root and in + sourcecode. If the directory is found, it is added to sys.path for dynamic imports. + Returns the path if found, else None. Args: - ip: The IP address to ping + None Returns: - True if the host is online (ping successful), False otherwise + str or None: The path to the dune-wib-firmware/sw directory if found, else None. Raises: - None + EnvironmentError: If DBT_AREA_ROOT is not set in the environment. """ + # Define the root of this work area to start the search from + work_area_root = os.getenv("DBT_AREA_ROOT", None) + if not work_area_root: + raise EnvironmentError( + "DBT_AREA_ROOT environment variable not set, ensure you are running from a " + "DUNE DAQ release." + ) - # Construct the ping command. Ping once with a timeout of 1 second. - cmd = ["ping", "-c", "1", "-W", "1", ip] + # Define potential search paths + search_paths = [ + work_area_root, + os.path.join(work_area_root, "sourcecode"), + ] - # Execute the ping command, suppressing output and errors. Use a timeout to avoid - # hanging on unresponsive hosts. + # Search for the dune-wib-firmware/sw directory in the defined paths + for base_source in search_paths: + potential_path = os.path.join(base_source, "dune-wib-firmware/sw") + if os.path.isdir(potential_path): + if potential_path not in sys.path: + sys.path.insert(0, potential_path) # TODO: Is this needed? + return potential_path + return None + + +# Get the path to the WIB firmware/software interface if it exists, and attempt to +# import the WIB library if found. If the library is not available, the script will +# still run but will only show online/offline status without FEMB details. +WIB_FW_SW_IFACE_PATH = setup_wib_path() +WIB_LIB_AVAILABLE = False +WIB = None +wibpb = None + +if WIB_FW_SW_IFACE_PATH: try: - result = subprocess.run( + import wib_pb2 as wibpb + from wib import WIB + + WIB_LIB_AVAILABLE = True + except (ImportError, ModuleNotFoundError): + pass + +# Define the addresses of the hardware to check. This is structured as: +# { +# "Apparatus Name": { +# "Apparatus Resource/Asset/Name TBC": { +# "WIB Number": "IP Address", +# ... +# }, +# ... +# }, +# ... +# } +WIB_DATA = { + "NP02 CB": { + "NP02 CB": { + "601": "10.73.137.50", + "602": "10.73.137.51", + "603": "10.73.137.52", + "604": "10.73.137.53", + "605": "10.73.137.54", + "606": "10.73.137.122", + } + }, + "NP02": { + "CRP4": { + "1001": "10.73.137.126", + "1002": "10.73.137.127", + "1003": "10.73.137.128", + "1004": "10.73.137.137", + "1005": "10.73.137.129", + "1006": "10.73.137.130", + }, + "CRP5": { + "1101": "10.73.137.131", + "1102": "10.73.137.132", + "1103": "10.73.137.133", + "1104": "10.73.137.134", + "1105": "10.73.137.135", + "1106": "10.73.137.136", + }, + }, + "NP04 CB": { + "NP04 CB": { + "001": "10.73.137.20", + "002": "10.73.137.21", + "003": "10.73.137.22", + "004": "10.73.137.23", + "005": "10.73.137.24", + } + }, + "NP04": { + "APA1": { + "101": "10.73.137.26", + "102": "10.73.137.27", + "103": "10.73.137.28", + "104": "10.73.137.29", + "105": "10.73.137.30", + }, + "APA2": { + "201": "10.73.137.31", + "202": "10.73.137.32", + "203": "10.73.137.33", + "204": "10.73.137.34", + "205": "10.73.137.35", + }, + "APA3": { + "301": "10.73.137.36", + "302": "10.73.137.37", + "303": "10.73.137.38", + "304": "10.73.137.39", + "305": "10.73.137.40", + }, + "APA4": { + "401": "10.73.137.41", + "402": "10.73.137.42", + "403": "10.73.137.43", + "404": "10.73.137.44", + "405": "10.73.137.45", + }, + }, +} + + +def check_hardware(ip: str) -> dict: + # Default state is 'In Progress' (None) + final_status = {"online": False, "fembs": [False] * 4} + + try: + # 1. Ping + cmd = ["ping", "-c", "1", "-W", "1", ip] + ping_res = subprocess.run( cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=1.5 ) - return result.returncode == 0 - except Exception: - return False + if ping_res.returncode == 0: + final_status["online"] = True -def make_wib_table( - category: str, wibs: dict[str, str], status_map: dict[str, bool | None] -) -> Table: - """ - Creates a Rich Table for a given category of WIBs, showing their online status and - address. + # 2. Protocol Check + if WIB_LIB_AVAILABLE: + try: + wib_inst = WIB(ip) + req = wibpb.GetFEMBStatus() + rep = wibpb.GetFEMBStatus.FEMBStatus() + # We wrap this in a sub-try because gRPC can hang + wib_inst.send_command(req, rep) + if hasattr(rep, "femb_power") and len(rep.femb_power) == 4: + final_status["fembs"] = list(rep.femb_power) + else: + final_status["fembs"] = [False] * 4 + except Exception: + final_status["fembs"] = [False] * 4 + else: + final_status["fembs"] = [False] * 4 + else: + final_status["online"] = False + final_status["fembs"] = [False] * 4 - Args: - category: The name of the category (e.g., "NP04 CB", "APA1", etc.). This is an - entry from the WIB_DATA mapping and will be used as the table title. - wibs: A mapping of WIB numbers to their IP addresses for this category - status_map: A mapping of IP addresses to their online status (True/False/None) + except Exception: + # Catch-all for timeouts or subprocess errors + pass - Returns: - A Rich Table object representing the WIBs in this category and their status + return final_status - Raises: - None - """ - # Create a Rich Table with a title based on the category. +def make_wib_table(category: str, wibs: dict[str, str], results_map: dict) -> Table: table = Table(title=f"[magenta]{category}[/]", box=box.ROUNDED, border_style="dim") - - # Add columns for WIB number, address, and online status. table.add_column("WIB #", justify="center") - table.add_column("Address", style="dim white") - table.add_column("Online?", justify="center") + for i in range(4): + table.add_column(f"FEMB {i}", justify="center") - # For each relevant WIB, show the status. for wib_num, ip in wibs.items(): - # Look up the status of this IP in the status map. - res = status_map.get(ip) - - # WIB is online - if res is True: - status = "[bold green]✔[/]" + # Get the result, or if not checked yet, show dots + res = results_map.get(ip) - # WIB is offline - elif res is False: - status = "[bold red]✘[/]" - - # Status is unknown (ping not completed yet) + if res is None: + wib_text = f"[white]{wib_num}[/]" + femb_icons = ["[dim]...[/]"] * 4 else: - status = "[yellow]...[/]" + wib_style = "bold green" if res["online"] else "bold red" + wib_text = f"[{wib_style}]{wib_num}[/]" - # Add a row to the table for this WIB with its number, IP address, and status. - table.add_row(wib_num, ip, status) + femb_icons = [] + for state in res["fembs"]: + if state is True: + femb_icons.append("[bold green]✔[/]") + else: + femb_icons.append("[bold red]✘[/]") + table.add_row(wib_text, *femb_icons) return table -def generate_table(status_map: dict[str, bool | None]) -> Columns: - """ - Generates a Rich Columns object containing tables for each category of WIBs. +def generate_display(results_map: dict) -> Table: + grid = Table.grid(expand=True) + for group_name, sub_cats in WIB_DATA.items(): + grid.add_row(f"\n[bold cyan]{group_name}[/]") + tables = [ + make_wib_table(cat, wibs, results_map) for cat, wibs in sub_cats.items() + ] + grid.add_row(Columns(tables, equal=True, expand=False)) + return grid - Args: - status_map: A mapping of IP addresses to their online status - - Returns: - A Rich Columns object containing the WIB status tables for display - Raises: - None +def main(): """ + Prints the power status of the hardware defined in WIB_DATA. - tables: list[Table] = [ - make_wib_table(cat, wibs, status_map) for cat, wibs in WIB_DATA.items() - ] - return Columns(tables, equal=True, expand=False) - - -def main() -> None: + For each WIB, the script first checks if it's online via ping, then if it is online, + the FEMB power status is queried using the WIB library. Results are displayed in a + live-updating table format. """ - Main function to execute the NP0x HW status check and display results in a - live-updating table. - - This function initializes the console, parses out all the IP addresses, and pings - them concurrently while updating the display in real-time. It handles graceful - shutdown on keyboard interrupt and ensures that all threads are properly terminated. - Args: - None - - Returns: - None - - Raises: - Any exceptions that occur during the execution of the host checks will be - handled within the get_host_info function. - """ - - # Initialize the console for Rich output + # Set up the singular console to which all output is redirected. console = Console() - # Get the timezone-aware current time for display in the header. - now = datetime.now(pytz.timezone(tz)) - - # Print a blank line for spacing and a header to indicate the start of the scan - console.print("\n") + # Print header with timestamp + now = datetime.now(pytz.UTC) console.print( - f"[bold cyan]Checking NP0x hardware status at {now.strftime('%Y-%m-%d %H:%M:%S %Z')}[/]\n" + "\n[bold cyan]Checking NP0x hardware status at " + f"{now.strftime('%Y-%m-%d %H:%M:%S %Z')}[/]\n" ) - # Extract all IPs from the WIB_DATA for processing - all_ips: list[str] = [ip for cat in WIB_DATA.values() for ip in cat.values()] + # Get all the IP addresses from the WIB_DATA structure to check. This flattens the + # nested structure into a single list of IPs. + all_ips = [ + ip + for apparatus in WIB_DATA.values() + for resource in apparatus.values() + for ip in resource.values() + ] + + # Initialize results map as empty/None for all IPs to force dots initially + results = {ip: None for ip in all_ips} - # Initialize a results map to store the status of each IP, starting with None (unknown) - results_map: dict[str, bool | None] = {ip: None for ip in all_ips} + # Define a ThreadPoolExecutor to check hardware in parallel, and a mapping of + # futures to IPs + executor = ThreadPoolExecutor(max_workers=10) - # 1. Create executor outside a context manager to allow manual shutdown control - executor = ThreadPoolExecutor(max_workers=40) - futures = {executor.submit(ping_host, ip): ip for ip in all_ips} + # Submit all hardware checks to the executor and store the future-to-IP mapping for + # later reference + futures = {executor.submit(check_hardware, ip): ip for ip in all_ips} - # Use Rich's Live to create a live-updating table. The table will be refreshed as - # results come in from the concurrent checks. + # Use a Live context to update the display as results come in. As each future + # completes, the corresponding IP's result is updated in the results map, and the + # display is refreshed to show the new status. A final update is done after all + # futures complete to ensure the display is fully up to date. The try-except block + # allows for graceful interruption with Ctrl+C, ensuring the executor is shut down + # properly. try: with Live( - generate_table(results_map), console=console, auto_refresh=True + generate_display(results), console=console, refresh_per_second=5 ) as live: - # While we have commands executing, we wait for them to complete. Use a - # timeout to periodically refresh the table and check for completed futures. - while futures: - try: - # Update the table with results as they come in. Remove completed - # futures from the tracking dict to avoid re-processing. - for future in as_completed(futures, timeout=0.1): - ip = futures.pop(future) - results_map[ip] = future.result() - live.update(generate_table(results_map)) - - # No futures have completed, but refresh the table to keep it "live" - except TimeoutError: - live.update(generate_table(results_map)) - - # If all futures are done, break out of the loop to finish up - if not futures: - break - - # Handle graceful shutdown on keyboard interrupt. This allows the user to stop the - # scan early without leaving hanging threads. + for future in as_completed(futures): + ip = futures[future] + results[ip] = future.result() + live.update(generate_display(results)) + + # Final sweep + live.update(generate_display(results)) + except KeyboardInterrupt: pass - - # When exiting, all threads are properly shut down. finally: - executor.shutdown(wait=False, cancel_futures=True) + executor.shutdown(wait=False) - # Final display of results after all checks are complete. + # Print final status summary and any warnings about hardware communication if the + # WIB library is not available. console.print("\n[bold green]Scan complete.[/]") + # If the WIB library is not available, print a warning message to the user. This is + # done after the main display loop to ensure it doesn't interfere with the + # live-updating tables. The message provides guidance on how to resolve the issue if + # the library is not found, or informs the user that the WIB firmware repository is + # not present if that's the case. The warning is styled to stand out and is enclosed + # in horizontal lines for emphasis. + if not WIB_LIB_AVAILABLE: + console.print("-" * 40) + console.print("[bold yellow]Hardware Communication Warning:[/]") + if WIB_FW_SW_IFACE_PATH: + console.print( + f"Modules found but not loaded. Try running [red]make -o build/%.d python[/] in:\n[blue]{WIB_FW_SW_IFACE_PATH}[/]" + ) + else: + console.print( + "Couldnt check wib status. [italic]dune-wib-firmware[/italic] repo not found." + ) + console.print("-" * 40) + if __name__ == "__main__": main() From c6e5cea8ceb1c0e2ab0d235ebf9dc786e2d6bc5e Mon Sep 17 00:00:00 2001 From: PawelPlesniak Date: Wed, 1 Apr 2026 19:51:00 +0200 Subject: [PATCH 3/4] Docs --- src/drunc/apps/check_np0x_hw_status.py | 90 +++++++++++++++++++++++++- 1 file changed, 89 insertions(+), 1 deletion(-) diff --git a/src/drunc/apps/check_np0x_hw_status.py b/src/drunc/apps/check_np0x_hw_status.py index 2bb553dfa..3a7f62d0c 100644 --- a/src/drunc/apps/check_np0x_hw_status.py +++ b/src/drunc/apps/check_np0x_hw_status.py @@ -162,6 +162,28 @@ def setup_wib_path(): def check_hardware(ip: str) -> dict: + """ + Checks the hardware status of a given IP address by first pinging it to determine if + it is online, and if it is online, attempting to query its FEMB power status using + the WIB library. + + Args: + ip (str): The IP address of the hardware to check. + + Returns: + dict: A dictionary containing the online status and FEMB power status of the + hardware, with the following structure: + { + "online": bool, # True if the device is reachable, False otherwise + "fembs": list[bool] # A list of 4 booleans indicating the power status + of each FEMB (True for powered, False for not powered) + } + + Raises: + None: All exceptions are caught and handled within the function, with the final + status defaulting to "offline" and all FEMBs as "not powered" in case of any + errors (e.g. timeouts, subprocess errors, gRPC errors, etc.). + """ # Default state is 'In Progress' (None) final_status = {"online": False, "fembs": [False] * 4} @@ -203,15 +225,54 @@ def check_hardware(ip: str) -> dict: def make_wib_table(category: str, wibs: dict[str, str], results_map: dict) -> Table: + """ + Creates a Rich Table for a given category of WIBs, showing their online status and + FEMB power status. + + Args: + category (str): The name of the category (e.g. "NP02 CB", "CRP4", etc.) to be + displayed as the table title. + wibs (dict): A mapping of WIB numbers to their corresponding IP addresses for + this category. + results_map (dict): A mapping of IP addresses to their hardware status results, + where each result is a dictionary containing 'online' status and 'fembs' + status. + + Returns: + Table: A Rich Table object representing the status of the WIBs in this category, + with the WIB number, online status, and FEMB power status displayed in a + visually intuitive format (e.g. green checkmarks for online/powered, red Xs + for offline/unpowered, and dimmed dots for unknown status). + + Raises: + None + """ + + # Create a table with a title based on the category name. table = Table(title=f"[magenta]{category}[/]", box=box.ROUNDED, border_style="dim") + + # Add columns for WIB number and each of the 4 FEMBs, with centered text. The WIB + # number will be colored based on online status, and the FEMB columns will show + # icons based on their power status. table.add_column("WIB #", justify="center") for i in range(4): table.add_column(f"FEMB {i}", justify="center") + # Iterate through the WIBs in this category, adding a row for each. The WIB number + # is styled based on whether it's online (green) or offline (red), and the FEMB + # columns show a green checkmark if powered, a red X if not powered, or dimmed dots + # if the status is unknown (e.g. if the ping check hasn't completed yet). The + # results_map is used to get the current status for each IP, and if no result is + # available yet, the row will show the WIB number in white and the FEMB columns as + # dimmed dots to indicate that the status is still being checked. for wib_num, ip in wibs.items(): - # Get the result, or if not checked yet, show dots + # Get the result for this IP from the results_map. res = results_map.get(ip) + # If no result is available yet (res is None), show the WIB number in white and + # the FEMB columns as dimmed dots. Otherwise, style the WIB number based on + # online status and show the FEMB columns with green checkmarks or red Xs based + # on their power status. if res is None: wib_text = f"[white]{wib_num}[/]" femb_icons = ["[dim]...[/]"] * 4 @@ -231,6 +292,33 @@ def make_wib_table(category: str, wibs: dict[str, str], results_map: dict) -> Ta def generate_display(results_map: dict) -> Table: + """ + Generates the overall display grid for the current results. + + This function creates a grid layout using the Rich library, where each apparatus + group (e.g. "NP02 CB", "NP02", "NP04 CB", "NP04") is displayed in its own section. + For each group, it iterates through the subcategories and creates a table for each + using the make_wib_table function. The resulting tables are arranged in columns + within the grid. The display is designed to be updated live as results come in, + showing the current status of each WIB and its FEMBs based on the results_map + provided. + + Args: + results_map (dict): A mapping of IP addresses to their hardware status results, + where each result is a dictionary containing 'online' status and 'fembs' + status. + + Returns: + Table: A Rich Table object representing the current status display for all + hardware. + + Raises: + None + """ + + # Create a main grid to hold all apparatus groups. The grid is set to expand to fill + # the available space. Each group will be added as a row, with its own set of tables + # for the subcategories. grid = Table.grid(expand=True) for group_name, sub_cats in WIB_DATA.items(): grid.add_row(f"\n[bold cyan]{group_name}[/]") From 6c5e75a8390980f5d3bc6aa0dcd6e3bad33c74c1 Mon Sep 17 00:00:00 2001 From: PawelPlesniak Date: Fri, 24 Apr 2026 14:50:15 +0200 Subject: [PATCH 4/4] bump --- src/drunc/apps/check_np0x_hw_status.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/drunc/apps/check_np0x_hw_status.py b/src/drunc/apps/check_np0x_hw_status.py index 3a7f62d0c..ca7e6e751 100644 --- a/src/drunc/apps/check_np0x_hw_status.py +++ b/src/drunc/apps/check_np0x_hw_status.py @@ -406,7 +406,7 @@ def main(): console.print("[bold yellow]Hardware Communication Warning:[/]") if WIB_FW_SW_IFACE_PATH: console.print( - f"Modules found but not loaded. Try running [red]make -o build/%.d python[/] in:\n[blue]{WIB_FW_SW_IFACE_PATH}[/]" + f"Modules found but not loaded. Try running [red]pip install zmq[/] and [red]make -o build/%.d python[/] in:\n[blue]{WIB_FW_SW_IFACE_PATH}[/]" ) else: console.print(