From 85215ce15078b5eca135f5e3fa5b07adc7d6db8a Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Fri, 12 Jun 2026 14:48:29 +0200 Subject: [PATCH] feat: add PoE config and LAG/trunk membership support get_poe_status()/set_poe() implement PoE port config via the JetStream CLI's 'power inline' interface sub-commands. get_interfaces() now enriches LagN entries and their member ports with trunk_group/lag_members/lag_mode by parsing 'show link-aggregation summary'. New set_lag_members() adds/ removes members via channel-group/no channel-group. Co-Authored-By: Claude Sonnet 4.6 --- napalm_tplink_jetstream/tplink_jetstream.py | 183 +++++++++++++++++++- 1 file changed, 182 insertions(+), 1 deletion(-) diff --git a/napalm_tplink_jetstream/tplink_jetstream.py b/napalm_tplink_jetstream/tplink_jetstream.py index 9cd9d61..30ba94e 100644 --- a/napalm_tplink_jetstream/tplink_jetstream.py +++ b/napalm_tplink_jetstream/tplink_jetstream.py @@ -46,6 +46,11 @@ class TPLinkJetstreamDriver(SwitchDriver): # Netmiko device type for TP-Link Jetstream NETMIKO_DEVICE_TYPE = "tplink_jetstream" + # PoE priority mapping between the CLI's "low/high/critical" and the + # generic priority strings used across drivers. + _POE_PRIORITY_MAP = {"low": "PPP_LOW", "high": "PPP_HIGH", "critical": "PPP_CRITICAL"} + _POE_PRIORITY_MAP_REV = {v: k for k, v in _POE_PRIORITY_MAP.items()} + def __init__( self, hostname: str, @@ -348,7 +353,54 @@ class TPLinkJetstreamDriver(SwitchDriver): status_out = self._send_command("show interface status") config_out = self._send_command("show interface configuration") - return self._parse_interfaces(status_out, config_out, global_mtu) + interfaces = self._parse_interfaces(status_out, config_out, global_mtu) + self._add_lag_info(interfaces) + return interfaces + + def _add_lag_info(self, interfaces: Dict[str, Dict]) -> None: + """Enrich ``interfaces`` with LAG/trunk membership from ``show link-aggregation summary``. + + JetStream switches report one row per configured LAG, e.g.:: + + Group Type Ports + Lag1 LACP Gi1/0/1,Gi1/0/2 + Lag2 STATIC Gi1/0/3,Gi1/0/4 + """ + try: + output = self._send_command("show link-aggregation summary") + except Exception: + return + + for line in output.splitlines(): + m = re.match(r"^\s*(Lag\d+)\s+(\S+)\s+(.+)$", line, re.I) + if not m: + continue + lag, lag_type, ports_raw = m.group(1), m.group(2), m.group(3).strip() + if lag_type.lower() not in ("lacp", "static"): + continue + members = re.findall(r"(?:Gi|Te|Fa)\d+(?:/\d+)*", ports_raw, re.I) + if not members: + continue + lag_mode = "lacp" if lag_type.lower() == "lacp" else "trunk" + for member in members: + if member in interfaces: + interfaces[member]["trunk_group"] = lag + known_members = [m for m in members if m in interfaces] + if lag in interfaces: + interfaces[lag]["lag_members"] = members + interfaces[lag]["lag_mode"] = lag_mode + else: + interfaces[lag] = { + "is_up": any(interfaces[m]["is_up"] for m in known_members), + "is_enabled": any(interfaces[m]["is_enabled"] for m in known_members), + "description": "", + "last_flapped": -1.0, + "speed": sum(interfaces[m]["speed"] for m in known_members), + "mtu": interfaces[known_members[0]]["mtu"] if known_members else 1518, + "mac_address": "", + "lag_members": members, + "lag_mode": lag_mode, + } def _parse_interfaces(self, status_output: str, config_output: str = "", mtu: int = 1518) -> Dict[str, Dict]: """Parse tabular output of 'show interface status' and 'show interface configuration'.""" @@ -1615,6 +1667,135 @@ class TPLinkJetstreamDriver(SwitchDriver): self._exit_config_mode() self._save_config() + # ------------------------------------------------------------------ + # LAG / trunk membership + # ------------------------------------------------------------------ + + def set_lag_members(self, lag_name: str, members: List[str]) -> None: + """Set the full member-port list of a LAG (e.g. ``"Lag1"``). + + Diffs ``members`` against the LAG's current members (as reported by + ``get_interfaces()``) and issues ``channel-group``/``no + channel-group`` interface sub-commands for the difference. The LACP + vs. static mode of new members follows the LAG's existing mode + (``lag_mode``), defaulting to LACP (``mode active``). + """ + current = self.get_interfaces().get(lag_name, {}) + current_members = set(current.get("lag_members") or []) + desired = set(members) + sort_key = lambda s: [int(x) if x.isdigit() else x for x in re.split(r"(\d+)", s)] # noqa: E731 + to_remove = sorted(current_members - desired, key=sort_key) + to_add = sorted(desired - current_members, key=sort_key) + + if not to_remove and not to_add: + return + + m = re.match(r"[Ll]ag(\d+)", lag_name) + if not m: + raise ValueError(f"Invalid LAG interface name: {lag_name!r}") + group_id = m.group(1) + mode = current.get("lag_mode") or "lacp" + channel_mode = "active" if mode == "lacp" else "on" + + lines: List[str] = [] + for port in to_remove: + cli_iface = self._to_cli_interface(port) + lines.append(f"interface {cli_iface}") + lines.append(" no channel-group") + lines.append("exit") + for port in to_add: + cli_iface = self._to_cli_interface(port) + lines.append(f"interface {cli_iface}") + lines.append(f" channel-group {group_id} mode {channel_mode}") + lines.append("exit") + + self._enter_config_mode() + try: + errors = self._apply_config_lines("\n".join(lines)) + if errors: + raise CommandErrorException(f"set_lag_members({lag_name}) errors: {errors}") + finally: + self._exit_config_mode() + self._save_config() + + # ------------------------------------------------------------------ + # PoE + # ------------------------------------------------------------------ + + def get_poe_status(self) -> Dict[str, Dict]: + """Return PoE configuration per port from ``show power inline``. + + Expected ``show power inline`` columns (JetStream T-series):: + + Port Admin State Power Limit(W) Priority Power(W) Status + Gi1/0/1 Enable 30.0 Low 5.2 On + """ + output = self._send_command("show power inline") + + result: Dict[str, Dict] = {} + in_table = False + for line in output.splitlines(): + line_s = line.strip() + if re.match(r"^-{4,}", line_s): + in_table = True + continue + if not in_table or not line_s: + continue + parts = line_s.split() + if not parts or not re.match(r"^(Gi|Te|Fa)\d+(/\d+)*$", parts[0], re.I): + continue + + port = parts[0] + admin = parts[1] if len(parts) > 1 else "" + + limit_w = 0.0 + priority = "" + for tok in parts[1:]: + tl = tok.lower() + if tl in self._POE_PRIORITY_MAP: + priority = tl + elif limit_w == 0.0 and re.match(r"^\d+(\.\d+)?$", tok): + limit_w = float(tok) + + result[port] = { + "port_id": port, + "is_poe_enabled": admin.lower() in ("enable", "on"), + "poe_priority": self._POE_PRIORITY_MAP.get(priority, "PPP_LOW"), + "poe_allocation_method": "PPAM_VALUE", + "allocated_power_in_watts": limit_w, + "pre_standard_detect_enabled": False, + } + + return result + + def set_poe(self, interface: str, config: Dict) -> None: + """Update PoE configuration for a single port. + + Uses the ``power inline`` interface sub-commands (JetStream CLI). + """ + iface = self._to_cli_interface(interface) + lines = [f"interface {iface}"] + if "is_poe_enabled" in config: + lines.append("power inline supply" if config["is_poe_enabled"] else "no power inline supply") + if "poe_priority" in config: + priority = self._POE_PRIORITY_MAP_REV.get(config["poe_priority"], "low") + lines.append(f"power inline priority {priority}") + if "allocated_power_in_watts" in config and config.get("poe_allocation_method") == "PPAM_VALUE": + lines.append(f"power inline power-limit {config['allocated_power_in_watts']}") + lines.append("exit") + + if len(lines) <= 2: + return + + self._enter_config_mode() + try: + errors = self._apply_config_lines("\n".join(lines)) + if errors: + raise CommandErrorException(f"set_poe({interface}) errors: {errors}") + finally: + self._exit_config_mode() + self._save_config() + def ping( self, destination: str,