diff --git a/napalm_procurve/api_client.py b/napalm_procurve/api_client.py index 34a9f67..f2ac1a8 100644 --- a/napalm_procurve/api_client.py +++ b/napalm_procurve/api_client.py @@ -287,6 +287,19 @@ class ProcurveApiClient: return {k: v for k, v in result.items() if v["ipv4"] or v["ipv6"]} + # ------------------------------------------------------------------ + # PoE + # ------------------------------------------------------------------ + + def get_poe_ports(self) -> Dict[str, Dict]: + """Return PoE configuration for all ports, keyed by port id.""" + data = self.get("poe/ports") + return {p["port_id"]: p for p in data.get("port_poe", []) if p.get("port_id")} + + def set_port_poe(self, port_id: str, payload: Dict[str, Any]) -> requests.Response: + """PUT PoE configuration changes for a single port.""" + return self.put(f"ports/{port_id}/poe", json=payload) + def get_arp_table(self, vrf: str = "") -> list: """Return NAPALM ARP table from the REST API.""" arp_data = self.get("arp") diff --git a/napalm_procurve/procurve.py b/napalm_procurve/procurve.py index d17b945..1e8ff39 100644 --- a/napalm_procurve/procurve.py +++ b/napalm_procurve/procurve.py @@ -896,6 +896,47 @@ class ProcurveDriver(SwitchDriver): self._exit_config_mode() self._save_config() + # ------------------------------------------------------------------ + # PoE + # ------------------------------------------------------------------ + + def get_poe_status(self) -> Dict[str, Dict]: + """Return PoE configuration for all ports, keyed by interface name. + + Only available via the REST API transport. + """ + if self._transport == "api": + return self._api.get_poe_ports() + raise NotImplementedError("PoE status is only available via the REST API") + + def set_poe(self, interface: str, config: Dict) -> None: + """Update PoE configuration for a single interface.""" + if self._transport != "api": + raise NotImplementedError("PoE configuration is only available via the REST API") + + payload: dict = {} + for key in ( + "is_poe_enabled", + "poe_priority", + "poe_allocation_method", + "pre_standard_detect_enabled", + ): + if key in config: + payload[key] = config[key] + if "allocated_power_in_watts" in config: + method = config.get("poe_allocation_method") or payload.get("poe_allocation_method") + if method == "PPAM_VALUE": + payload["allocated_power_in_watts"] = config["allocated_power_in_watts"] + + if not payload: + return + + resp = self._api.set_port_poe(interface, payload) + if not resp.ok: + raise ConnectionException( + f"set_poe({interface}): ports/poe PUT HTTP {resp.status_code} – {resp.text[:200]}" + ) + def ping( self, destination: str,