alexandre.pires
d069fe3585
- Implemented a new module for cert-checker with Kubernetes resources including ServiceAccount, Role, RoleBinding, ConfigMap, and CronJob. - Added Python script to check certificate expiry and restart deployments if necessary. - Configured environment variables for deployment name, namespace, and service details. - Included requirements for the Python environment. feat(devolo-exporter): introduce Devolo exporter for Prometheus metrics - Created a new module for Devolo exporter with Kubernetes resources including ServiceAccount, ConfigMap, Deployment, and Service. - Developed Python scripts to scrape metrics from Devolo devices and expose them to Prometheus. - Configured environment variables for Devolo device IP, model, and ports. - Added requirements for the Python environment.
66 lines
2.0 KiB
Python
66 lines
2.0 KiB
Python
import json
|
|
import websocket
|
|
import time
|
|
import requests
|
|
|
|
# Step 1: Find Chrome DevTools WebSocket endpoint
|
|
targets = requests.get("http://localhost:9222/json").json()
|
|
ws_url = targets[0]["webSocketDebuggerUrl"]
|
|
|
|
# Step 2: Connect to the DevTools Protocol
|
|
ws = websocket.create_connection(ws_url)
|
|
print(f"Connected to {ws_url}")
|
|
|
|
# Helper: receive until we get the response for a given id
|
|
def recv_until_id(expected_id, timeout=10):
|
|
ws.settimeout(timeout)
|
|
while True:
|
|
msg = json.loads(ws.recv())
|
|
# Log protocol events to help debugging
|
|
if "id" not in msg:
|
|
print(msg)
|
|
continue
|
|
if msg["id"] == expected_id:
|
|
return msg
|
|
# Optionally log out-of-order responses
|
|
print(msg)
|
|
|
|
# Step 3: Enable runtime
|
|
ws.send(json.dumps({"id": 1, "method": "Runtime.enable"}))
|
|
recv_until_id(1)
|
|
|
|
# Step 4: Navigate to the page and wait a bit
|
|
ws.send(json.dumps({"id": 2, "method": "Page.navigate", "params": {"url": "http://192.168.0.199/#/powerline"}}))
|
|
recv_until_id(2)
|
|
time.sleep(2) # give the page some time to render
|
|
|
|
# Step 5: Evaluate JavaScript to extract device data, and return by value
|
|
script = """
|
|
Array.from(document.querySelectorAll('tr[id^="plc-"]')).map(tr => ({
|
|
id: tr.querySelector('[id^="plc-id-"]')?.innerText?.trim(),
|
|
mac: tr.querySelector('[id^="plc-mac-"]')?.innerText?.trim(),
|
|
tx: tr.querySelector('[id^="plc-tx-"]')?.innerText?.trim(),
|
|
rx: tr.querySelector('[id^="plc-rx-"]')?.innerText?.trim()
|
|
}))
|
|
"""
|
|
ws.send(json.dumps({
|
|
"id": 3,
|
|
"method": "Runtime.evaluate",
|
|
"params": {
|
|
"expression": script,
|
|
"returnByValue": True,
|
|
"awaitPromise": True
|
|
}
|
|
}))
|
|
response = recv_until_id(3)
|
|
result = response.get("result", {}).get("result", {}).get("value")
|
|
|
|
print("\n📡 Device Table:")
|
|
if result:
|
|
for dev in result:
|
|
print(f"ID: {dev['id']:<10} MAC: {dev['mac']:<20} TX: {dev['tx']:<6} RX: {dev['rx']}")
|
|
else:
|
|
print("No devices found — maybe the table hasn't loaded yet?")
|
|
|
|
ws.close()
|