NAPALM/napalm-tplink-jetstream
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add PoE config and LAG/trunk membership support

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
Christian Manivong
2026-06-12 14:48:29 +02:00
co-authored by Claude Sonnet 4.6
parent c53b21ad65
commit 85215ce150
1 changed files with 182 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
napalm_tplink_jetstream/tplink_jetstream.py
+182 -1
View File
@@ -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,