# -*- coding: utf-8 -*- # Licensed under the Apache License, Version 2.0 """CLI output parsers for HPE/Aruba ProCurve switches. All functions are pure (no side effects) and operate on pre-collected command output strings. The same parsers are shared by the SSH and Telnet backends. """ import re from typing import Dict, List, Optional # --------------------------------------------------------------------------- # MAC address helpers # --------------------------------------------------------------------------- def procurve_mac_to_std(mac: str) -> str: """Convert ProCurve MAC format ``a1b2c3-d4e5f6`` to ``a1:b2:c3:d4:e5:f6``.""" clean = mac.replace("-", "").replace(":", "").replace(".", "").lower() if len(clean) != 12: return mac return ":".join(clean[i : i + 2] for i in range(0, 12, 2)) def _kv(output: str, key: str) -> str: """Extract value from ``key : value`` style ProCurve output (case-insensitive).""" for line in output.splitlines(): if re.search(re.escape(key), line, re.IGNORECASE): parts = line.split(":", 1) if len(parts) == 2: return parts[1].strip() return "" # --------------------------------------------------------------------------- # show system-information # --------------------------------------------------------------------------- def parse_system_info(output: str) -> Dict: """Parse ``show system-information`` output. Returns a dict with keys: hostname, os_version, serial_number, model, base_mac, uptime_seconds. Handles both older ProCurve style (two-column layout) and newer Aruba style. Example older output:: System Name : myswitch Software revision : R.11.27 Base MAC Addr : a1b2c3-d4e5f6 Serial Number : SGXXXXXXXXXX Up Time : 5 days 3:14 Example newer output:: System Name : ArubaSW Firmware revision: YA.16.04.0006 Serial Number : SGXXXXXXXXXX """ # Use a regex to extract key:value pairs from each line. # ProCurve CLI may pack two k/v pairs on the same line with many spaces # between them, e.g.: # " Software revision : R.11.27 Base MAC Addr : a1b2c3-d4e5f6" # We capture all k:v pairs on every line. _KV_RE = re.compile( r"([A-Za-z][A-Za-z0-9 ()/#-]{1,40}?)\s*:\s*(\S.*?)(?=\s{3,}[A-Za-z]|$)" ) kv_map: Dict[str, str] = {} for raw_line in output.splitlines(): for m in _KV_RE.finditer(raw_line): k = m.group(1).strip() v = m.group(2).strip() if k and v: kv_map[k] = v def get(key: str) -> str: for k, v in kv_map.items(): if key.lower() in k.lower(): return v return "" hostname = get("System Name") os_version = ( get("Software revision") or get("Firmware revision") or get("Software Revision") or get("Firmware Version") ) serial = get("Serial Number") or get("Serial Num") base_mac = procurve_mac_to_std(get("Base MAC Addr") or get("MAC Address") or "") model_raw = get("System Model") or get("product_model") or "" # Extract HPE/HP J-code part numbers (e.g. "J9298A") from the model string. # The J-code is the PN; the remaining text is the clean model name. # Examples: # "HP J9298A Switch 2520G-8-PoE" → model="HP 2520G-8-PoE Switch", pn="J9298A" # "HP2530-8G Switch(J9777A)" → model="HP2530-8G Switch", pn="J9777A" _PN_RE = re.compile(r"\b(J\d{4}[A-Z]{1,2})\b") pn_match = _PN_RE.search(model_raw) part_number = pn_match.group(1) if pn_match else "" if pn_match: clean = _PN_RE.sub("", model_raw) clean = re.sub(r"\(\s*\)", "", clean) clean = re.sub(r"\s{2,}", " ", clean).strip(" -()") model = clean if clean else model_raw else: model = model_raw # Parse uptime — ProCurve formats: "5 days 3:14" or "0 days 0:05" # Also: "5 day(s), 3:14:00" uptime_seconds = _parse_procurve_uptime(get("Up Time") or get("Uptime") or "") return { "hostname": hostname, "os_version": os_version, "serial_number": serial, "base_mac": base_mac, "model": model, "part_number": part_number, "uptime_seconds": uptime_seconds, } def _parse_procurve_uptime(uptime_str: str) -> float: """Convert ProCurve uptime string to total seconds. Handles formats:: 5 days 3:14 → 5*86400 + 3*3600 + 14*60 0 days 0:05:03 → 303 5 day(s), 3:14:00 → same 1 hr 5 min 3 sec → some devices """ total = 0.0 # Format: "N days H:MM" or "N days H:MM:SS" m = re.search(r"(\d+)\s+days?\s+(\d+):(\d+)(?::(\d+))?", uptime_str, re.I) if m: total = ( int(m.group(1)) * 86400 + int(m.group(2)) * 3600 + int(m.group(3)) * 60 + int(m.group(4) or 0) ) return float(total) # Format: "Xd Xh Xm Xs" or "X day(s), X hour(s), ..." days = re.search(r"(\d+)\s+day", uptime_str, re.I) hours = re.search(r"(\d+)\s+h(?:r|our)", uptime_str, re.I) minutes = re.search(r"(\d+)\s+min", uptime_str, re.I) seconds = re.search(r"(\d+)\s+sec", uptime_str, re.I) if days or hours or minutes or seconds: total = ( int(days.group(1) if days else 0) * 86400 + int(hours.group(1) if hours else 0) * 3600 + int(minutes.group(1) if minutes else 0) * 60 + int(seconds.group(1) if seconds else 0) ) return float(total) # --------------------------------------------------------------------------- # show version # --------------------------------------------------------------------------- def parse_version(output: str) -> str: """Extract firmware version from ``show version`` or ``show system-information``. ProCurve ``show version`` example:: HP J9565A 2520G-8-PoE Switch Software revision : R.11.27 Aruba image stamp format:: Image stamp: /ws/.../WC.16.11.0004 WC.16.11.0004 """ # "Software revision : R.11.27" style m = re.search(r"(?:software|firmware)\s+revision\s*:\s*(\S+)", output, re.I) if m: return m.group(1) # Bare revision on its own line (image stamp format) for line in output.splitlines(): line = line.strip() if re.match(r"^[A-Z]{2}\.\d+\.\d+", line): return line return "" def parse_model_from_version(output: str) -> str: """Extract model from ``show version`` first line. Example: ``HP J9565A 2520G-8-PoE Switch`` → ``2520G-8-PoE`` """ first = output.strip().splitlines()[0] if output.strip() else "" m = re.search(r"HP\s+\S+\s+(\S+)\s+Switch", first, re.I) if m: return m.group(1) # Aruba format: "Aruba 2530-8-PoE+ Switch" m = re.search(r"(?:Aruba|HP)\s+(\d\S+)\s+(?:Switch|)", first, re.I) if m: return m.group(1) return "" # --------------------------------------------------------------------------- # show interfaces brief / show port status # --------------------------------------------------------------------------- # ProCurve "show interfaces brief" tabular header line reference: # Port Type | Enabled Link MDI Mode FlwCtrl BCast # ------ ------------ + ------- ------ ----- ----------- -------- ----- # 1 100/1000T | Yes Down Auto Unknown off 0 _INTF_BRIEF_RE = re.compile( r"^\s*(\S+)\s+" # Port r"(\S+)\s+\|" # Type | r"\s+(Yes|No)\s+" # Enabled r"(Up|Down)\s+" # Link r"\S+\s+" # MDI (ignored) r"(\S+)", # Mode (speed/duplex) re.IGNORECASE, ) def parse_interfaces_brief(output: str) -> Dict[str, Dict]: """Parse ``show interfaces brief`` tabular output. Returns dict keyed by port name with sub-keys: is_up, is_enabled, speed, duplex, description. """ interfaces: Dict[str, Dict] = {} for line in output.splitlines(): m = _INTF_BRIEF_RE.match(line) if not m: continue port, itype, enabled, link, mode = ( m.group(1), m.group(2), m.group(3), m.group(4), m.group(5) ) speed = 0.0 duplex = "" # Mode examples: "1000FDx", "100HDx", "Unknown", "Auto" sm = re.match(r"(\d+)(FDx|HDx)?", mode, re.I) if sm: speed = float(sm.group(1)) duplex = "full" if (sm.group(2) or "").lower() == "fdx" else "half" interfaces[port] = { "is_up": link.lower() == "up", "is_enabled": enabled.lower() == "yes", "speed": speed, "duplex": duplex, "description": "", "last_flapped": -1.0, "mtu": -1, "mac_address": "", } return interfaces # --------------------------------------------------------------------------- # show interfaces (per-port detail for MAC / description) # --------------------------------------------------------------------------- def parse_interface_detail(output: str, port: str) -> Dict: """Parse ``show interfaces `` for MAC address and description.""" mac = procurve_mac_to_std(_kv(output, "MAC address")) name = _kv(output, "Name") or _kv(output, "Description") return {"mac_address": mac, "description": name} # --------------------------------------------------------------------------- # show ip / show ip interface # --------------------------------------------------------------------------- def parse_ip_interfaces(output: str) -> Dict[str, Dict]: """Parse ``show ip`` or ``show ip interface brief`` output. Returns NAPALM-compatible dict:: {"VLAN1": {"ipv4": {"192.168.0.1": {"prefix_length": 24}}}} ProCurve ``show ip`` multi-VLAN format:: IP Configuration for VLAN 1: IP Config : Manual IP Address : 192.168.0.1 Subnet Mask : 255.255.255.0 IP Configuration for VLAN 99: ... """ result: Dict[str, Dict] = {} current_vlan = "" current_ip = "" for line in output.splitlines(): # Detect VLAN section header: "IP Configuration for VLAN X" or "... VLAN 1:" vm = re.search(r"IP\s+Config(?:uration)?\s+for\s+(?:VLAN\s+)?(\S+)", line, re.I) if vm: raw_id = vm.group(1).rstrip(":") # Strip a leading "VLAN" prefix if the regex captured it (e.g. "VLAN1"→"1") num = re.sub(r"^VLAN", "", raw_id, flags=re.I) current_vlan = f"VLAN{num}" result.setdefault(current_vlan, {"ipv4": {}, "ipv6": {}}) continue # Single VLAN output (no header): detect from "IP Address" line if re.search(r"IP\s+Address\s*:", line, re.I): ip = line.split(":", 1)[-1].strip() if ip and ip not in ("0.0.0.0", ""): current_ip = ip if not current_vlan: current_vlan = "VLAN1" result.setdefault(current_vlan, {"ipv4": {}, "ipv6": {}}) continue if re.search(r"Subnet\s+Mask\s*:", line, re.I) and current_ip: mask = line.split(":", 1)[-1].strip() try: import netaddr prefix_len = netaddr.IPNetwork(f"{current_ip}/{mask}").prefixlen except Exception: prefix_len = 24 if current_vlan: result[current_vlan]["ipv4"][current_ip] = {"prefix_length": prefix_len} current_ip = "" continue # Clean up empty entries return {k: v for k, v in result.items() if v["ipv4"] or v["ipv6"]} # --------------------------------------------------------------------------- # show arp # --------------------------------------------------------------------------- _ARP_RE = re.compile( r"(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})" # IP r"\s+" r"([0-9a-fA-F]{6}-[0-9a-fA-F]{6}|" # MAC ProCurve hex-dash format r"[0-9a-fA-F:]{17})" # or standard colon format r"\s+(\S+)" # Type r"(?:\s+(\S+))?", # Port (optional) ) def parse_arp_table(output: str) -> List[Dict]: """Parse ``show arp`` output into NAPALM ARP table format.""" table = [] for line in output.splitlines(): m = _ARP_RE.search(line) if not m: continue ip, mac, entry_type, port = ( m.group(1), m.group(2), m.group(3), m.group(4) or "" ) table.append( { "interface": port, "mac": procurve_mac_to_std(mac), "ip": ip, "age": -1.0, } ) return table # --------------------------------------------------------------------------- # show mac-address # --------------------------------------------------------------------------- _MAC_RE = re.compile( r"([0-9a-fA-F]{6}-[0-9a-fA-F]{6}|[0-9a-fA-F:]{17})" # MAC r"\s+(\S+)" # Port r"\s+(\d+)" # VLAN r"\s+(\w+)", # Type ) def parse_mac_table(output: str) -> List[Dict]: """Parse ``show mac-address`` into NAPALM MAC table format.""" table = [] for line in output.splitlines(): m = _MAC_RE.search(line) if not m: continue mac, port, vlan, entry_type = m.group(1), m.group(2), m.group(3), m.group(4) table.append( { "mac": procurve_mac_to_std(mac), "interface": port, "vlan": int(vlan), "static": entry_type.lower() in ("static", "permanent"), "active": True, "moves": -1, "last_move": -1.0, } ) return table # --------------------------------------------------------------------------- # show lldp info remote-device # --------------------------------------------------------------------------- def parse_lldp_neighbors(output: str) -> Dict[str, List[Dict]]: """Parse ``show lldp info remote-device`` into NAPALM neighbors dict. Output format:: LocalPort | ChassisId PortId SysName ... --------- + ------------------- ------------ ------------ ... 1 | a1 b2 c3 d4 e5 f6 Gi 0/1 otherswitch ... """ neighbors: Dict[str, List[Dict]] = {} in_table = False for line in output.splitlines(): # Skip header / separator lines if re.match(r"\s*-+\s*\+", line) or re.match(r"\s*LocalPort", line, re.I): in_table = True continue if not in_table: continue if not line.strip(): continue # Format: " 1 | aa bb cc ... Gi0/1 hostname ..." m = re.match(r"\s*(\S+)\s*\|\s*(.+)", line) if not m: continue local_port = m.group(1) rest = m.group(2).strip() # ChassisId: may be space-separated octets (aa bb cc dd ee ff) or MAC # Then PortId, then SysName # Split by 2+ spaces to separate columns cols = re.split(r"\s{2,}", rest) if len(cols) < 3: continue chassis_raw = cols[0].strip() port_id = cols[1].strip() sys_name = cols[2].strip() # Normalise chassis ID: space-separated octets → colon MAC chassis = re.sub(r"\s+", "", chassis_raw) if re.match(r"^[0-9a-fA-F]{12}$", chassis): chassis = ":".join(chassis[i : i + 2] for i in range(0, 12, 2)) neighbors.setdefault(local_port, []).append( {"hostname": sys_name, "port": port_id} ) return neighbors def parse_lldp_neighbors_detail(output: str) -> Dict[str, List[Dict]]: """Parse ``show lldp info remote-device detail`` output. Returns NAPALM lldp_neighbors_detail format. """ neighbors: Dict[str, List[Dict]] = {} current_port = "" entry: Dict = {} def _save(): if current_port and entry: neighbors.setdefault(current_port, []).append(entry.copy()) for line in output.splitlines(): # New port section: "LLDP Remote Device Information Detail for Port N" m = re.search(r"for\s+[Pp]ort\s+(\S+)", line) if m: _save() current_port = m.group(1) entry = { "parent_interface": "", "remote_chassis_id": "", "remote_port": "", "remote_port_description": "", "remote_system_name": "", "remote_system_description": "", "remote_system_capab": [], "remote_system_enable_capab": [], } continue if not entry: continue low = line.lower() if "chassis id" in low: val = line.split(":", 1)[-1].strip() entry["remote_chassis_id"] = procurve_mac_to_std( re.sub(r"\s+", "", val) ) if re.search(r"[0-9a-fA-F ]{14,}", val) else val elif "port id" in low and "descr" not in low: entry["remote_port"] = line.split(":", 1)[-1].strip() elif "port description" in low: entry["remote_port_description"] = line.split(":", 1)[-1].strip() elif "system name" in low: entry["remote_system_name"] = line.split(":", 1)[-1].strip() elif "system description" in low: entry["remote_system_description"] = line.split(":", 1)[-1].strip() elif "system capabilities" in low and "enabled" not in low: entry["remote_system_capab"] = _parse_capabilities( line.split(":", 1)[-1].strip() ) elif "enabled capabilities" in low or ("system capabilities" in low and "enabled" in low): entry["remote_system_enable_capab"] = _parse_capabilities( line.split(":", 1)[-1].strip() ) _save() return neighbors def _parse_capabilities(cap_str: str) -> List[str]: """Convert ProCurve capability abbreviations to NAPALM strings.""" _MAP = { "B": "bridge", "R": "router", "T": "telephone", "W": "wlan-access-point", "H": "host", "D": "docsis-cable-device", "O": "other", "S": "station", "C": "customer-vlan-bridge", } result = [] for token in re.split(r"[,\s]+", cap_str): token = token.strip() if not token: continue result.append(_MAP.get(token, token.lower())) return result # --------------------------------------------------------------------------- # show ntp associations / show timesync # --------------------------------------------------------------------------- def parse_ntp_servers(output: str) -> Dict[str, Dict]: """Parse ``show ntp associations`` or ``show ntp status`` output. Returns dict of {server_ip: {}} as per NAPALM get_ntp_servers. """ servers: Dict[str, Dict] = {} for line in output.splitlines(): # " 10.0.0.1 sys.peer 2 ..." or "NTP server: 10.0.0.1" m = re.search(r"(\d{1,3}(?:\.\d{1,3}){3})", line) if m: ip = m.group(1) # Skip gateway / default-gateway entries if "gateway" in line.lower() or "default" in line.lower(): continue servers[ip] = {} return servers # --------------------------------------------------------------------------- # show environment # --------------------------------------------------------------------------- def parse_environment(output: str) -> Dict: """Parse ``show environment`` output into NAPALM environment dict.""" env: Dict = { "fans": {}, "temperature": {}, "power": {}, "cpu": [{"%usage": 0.0}], "memory": {"available_ram": -1, "used_ram": -1}, } # CPU utilisation from show system-information m = re.search(r"CPU\s+Util(?:ization)?\s*[(%:]+\s*(\d+)", output, re.I) if m: env["cpu"][0]["%usage"] = float(m.group(1)) # Memory total_m = re.search(r"Memory\s*[-–]\s*Total\s*[:\s]+(\d+)", output, re.I) free_m = re.search(r"Free\s*[:\s]+(\d+)", output, re.I) if total_m: env["memory"]["available_ram"] = int(total_m.group(1)) if free_m: env["memory"]["used_ram"] = ( env["memory"]["available_ram"] - int(free_m.group(1)) if env["memory"]["available_ram"] > 0 else -1 ) # Temperature sensors: "Sensor 1: 38C" or "Temperature: 38C" for m in re.finditer( r"(?:Sensor\s+(\d+)|Temperature(?:\s+Sensor)?)\s*[:\s]+(\d+)\s*[Cc°]", output, ): sensor_id = f"sensor_{m.group(1) or '1'}" temp = float(m.group(2)) env["temperature"][sensor_id] = { "temperature": temp, "is_alert": False, "is_critical": False, } # Fan status for m in re.finditer( r"Fan\s*(\d+)\s*[:\s]+(\S+)", output, re.I ): fan_id = f"fan_{m.group(1)}" status = m.group(2).lower() env["fans"][fan_id] = {"status": status in ("ok", "good", "normal", "operating")} return env # --------------------------------------------------------------------------- # show local-users # --------------------------------------------------------------------------- def parse_users(output: str) -> Dict[str, Dict]: """Parse ``show local-users`` or ``show users`` output. Returns NAPALM users dict. """ users: Dict[str, Dict] = {} for line in output.splitlines(): # " manager Operator Enabled ..." or " manager Manager ..." m = re.match( r"\s+(\w+)\s+(\w+(?:\s+\w+)?)\s+(Enabled|Disabled)", line, re.I, ) if not m: continue uname, level, _state = m.group(1), m.group(2).lower(), m.group(3) # Map privilege level if "manager" in level or "admin" in level: priv = 15 elif "operator" in level: priv = 5 else: priv = 1 users[uname] = { "level": priv, "password": "", "sshkeys": [], } return users # --------------------------------------------------------------------------- # show snmp-server / show snmp # --------------------------------------------------------------------------- def parse_snmp_info(output: str) -> Dict: """Parse ``show snmp-server`` output into NAPALM snmp_information dict.""" communities: Dict[str, Dict] = {} contact = _kv(output, "System Contact") or _kv(output, "Contact") location = _kv(output, "System Location") or _kv(output, "Location") chassis_id = "" for line in output.splitlines(): # " community_name Manager Restricted" (read only) # " community_name Operator Unrestricted" (read write) m = re.match(r"\s+(\S+)\s+(Manager|Operator)\s+(Restricted|Unrestricted)", line, re.I) if m: comm, _role, access = m.group(1), m.group(2), m.group(3) communities[comm] = { "acl": "", "mode": "ro" if access.lower() == "restricted" else "rw", } return { "chassis_id": chassis_id, "community": communities, "contact": contact, "location": location, } # --------------------------------------------------------------------------- # VLAN helpers # --------------------------------------------------------------------------- def expand_procurve_ports(port_str: str) -> List[str]: """Expand a ProCurve port-list string into a flat list of port names. Handles: - Numeric ranges: ``1-4,6,8`` → ``["1","2","3","4","6","8"]`` - Alphanumeric ranges: ``A1-A8`` → ``["A1","A2",..."A8"]`` - Trunk ports: ``Trk1,Trk2`` → ``["Trk1","Trk2"]`` - Mixed: ``1-3,Trk1,A1`` → ``["1","2","3","Trk1","A1"]`` """ ports: List[str] = [] for token in port_str.split(","): token = token.strip() if not token: continue # Try alphanumeric range: e.g. "A1-A8" (prefix + digits) alpha_range = re.match(r"^([A-Za-z]+)(\d+)-([A-Za-z]+)(\d+)$", token) if alpha_range: prefix = alpha_range.group(1) start = int(alpha_range.group(2)) end = int(alpha_range.group(4)) ports.extend(f"{prefix}{i}" for i in range(start, end + 1)) continue # Try pure numeric range: e.g. "1-4" num_range = re.match(r"^(\d+)-(\d+)$", token) if num_range: start = int(num_range.group(1)) end = int(num_range.group(2)) ports.extend(str(i) for i in range(start, end + 1)) continue # Single port (numeric, Trk, alphanumeric stacking port etc.) ports.append(token) return ports def parse_vlans_from_running_config(config_text: str) -> Dict[int, Dict]: """Parse VLAN blocks from ProCurve ``show running-config`` output. Extracts ``name``, ``tagged``, and ``untagged`` port lists per VLAN. Typical running-config VLAN block:: vlan 1 name "DEFAULT_VLAN" untagged 1-4,6-8 ip address 192.168.0.1 255.255.255.0 exit vlan 10 name "MGMT" tagged 1,2,5 exit Returns extended NAPALM-compatible dict:: { 1: {"name": "DEFAULT_VLAN", "interfaces": ["1","2","3","4","6","7","8","5"], "tagged": [], "untagged": ["1","2","3","4","6","7","8"]}, 10: {"name": "MGMT", "interfaces": ["1","2","5"], "tagged": ["1","2","5"], "untagged": []}, } """ vlans: Dict[int, Dict] = {} current_id: Optional[int] = None for line in config_text.splitlines(): stripped = line.strip() # Start of a vlan block: "vlan " m = re.match(r"^vlan\s+(\d+)\s*$", stripped, re.I) if m: current_id = int(m.group(1)) vlans[current_id] = {"name": f"VLAN{current_id}", "tagged": [], "untagged": []} continue if current_id is None: continue # End of vlan block if stripped.lower() == "exit": current_id = None continue # name "VLAN-Name" (with or without quotes) nm = re.match(r'^name\s+"?([^"]+)"?\s*$', stripped, re.I) if nm: vlans[current_id]["name"] = nm.group(1).strip() continue # untagged utm = re.match(r"^untagged\s+(.+)$", stripped, re.I) if utm: vlans[current_id]["untagged"] = expand_procurve_ports(utm.group(1)) continue # tagged tgm = re.match(r"^tagged\s+(.+)$", stripped, re.I) if tgm: vlans[current_id]["tagged"] = expand_procurve_ports(tgm.group(1)) continue # Populate "interfaces" as union of tagged + untagged (NAPALM standard key) for vdata in vlans.values(): seen: List[str] = [] for p in vdata["untagged"] + vdata["tagged"]: if p not in seen: seen.append(p) vdata["interfaces"] = seen return vlans def parse_vlans_brief(output: str) -> Dict[int, str]: """Parse ``show vlans`` brief table for VLAN IDs and names only. Used as a discovery step when running-config is unavailable. Example output:: VLAN ID Name | Status Voice Jumbo ------- ---------------- + ---------- ----- ----- 1 DEFAULT_VLAN | Port-based No No 10 MANAGEMENT | Port-based No No Returns ``{vlan_id: name}`` dict. """ vlans: Dict[int, str] = {} in_table = False for line in output.splitlines(): if re.match(r"\s*-{4,}", line): in_table = True continue if not in_table: continue m = re.match(r"\s*(\d+)\s+(\S+)", line) if m: vlans[int(m.group(1))] = m.group(2) return vlans def parse_vlan_ports_detail(output: str) -> Dict[str, List[str]]: """Parse ``show vlans `` per-VLAN port detail output. Returns ``{"tagged": [...], "untagged": [...]}`` for ONE vlan. Example output:: Port Information Mode Unknown VLAN Status ---------------- -------- ------------ ------- 1 Untagged Learn Up 2 Tagged Learn Up Trk1 Tagged Learn Up """ tagged: List[str] = [] untagged: List[str] = [] in_table = False for line in output.splitlines(): if re.match(r"\s*-{4,}", line): in_table = True continue if not in_table: continue # Columns: Port | Mode | Unknown VLAN | Status m = re.match(r"\s*(\S+)\s+(Tagged|Untagged)\s+", line, re.I) if m: port, mode = m.group(1), m.group(2).lower() if mode == "tagged": tagged.append(port) else: untagged.append(port) return {"tagged": tagged, "untagged": untagged} def derive_port_pvids(vlans: Dict[int, Dict]) -> Dict[str, int]: """Derive PVID (native/access VLAN) for each port from the ``get_vlans`` result. On ProCurve a port is untagged in exactly ONE VLAN — that is its PVID. A port that only appears in ``tagged`` lists has no PVID (trunk-only, PVID = 1 by default). Returns ``{"port": pvid_int, ...}`` mapping. Example:: # Port 1 is untagged in VLAN 1 → PVID 1 # Port 5 only tagged in VLAN 10 → PVID 1 (default) derive_port_pvids({ 1: {"untagged": ["1","2","3"], "tagged": []}, 10: {"untagged": [], "tagged": ["5"]}, }) # → {"1": 1, "2": 1, "3": 1, "5": 1} """ pvids: Dict[str, int] = {} # First pass: assign PVID from untagged membership for vid, vdata in sorted(vlans.items()): for port in vdata.get("untagged", []): pvids[port] = vid # Second pass: ports that only appear in tagged lists get PVID 1 (default) for vdata in vlans.values(): for port in vdata.get("tagged", []): if port not in pvids: pvids[port] = 1 return pvids