feat: add PoE config and power-cycle support for ProCurve REST API
Adds get_poe_status()/set_poe() to expose per-port PoE configuration (enable state, priority, allocation method, allocated power) and allow toggling it via the AOS-Switch REST API.
This commit is contained in:
@@ -287,6 +287,19 @@ class ProcurveApiClient:
|
|||||||
|
|
||||||
return {k: v for k, v in result.items() if v["ipv4"] or v["ipv6"]}
|
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:
|
def get_arp_table(self, vrf: str = "") -> list:
|
||||||
"""Return NAPALM ARP table from the REST API."""
|
"""Return NAPALM ARP table from the REST API."""
|
||||||
arp_data = self.get("arp")
|
arp_data = self.get("arp")
|
||||||
|
|||||||
@@ -896,6 +896,47 @@ class ProcurveDriver(SwitchDriver):
|
|||||||
self._exit_config_mode()
|
self._exit_config_mode()
|
||||||
self._save_config()
|
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(
|
def ping(
|
||||||
self,
|
self,
|
||||||
destination: str,
|
destination: str,
|
||||||
|
|||||||
Reference in New Issue
Block a user