Initial homelab infrastructure
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
+113
@@ -0,0 +1,113 @@
|
||||
from flask import Flask, request, jsonify
|
||||
from wakeonlan import send_magic_packet
|
||||
import logging
|
||||
import os
|
||||
import subprocess
|
||||
|
||||
app = Flask(__name__)
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
|
||||
# Format: name:mac:ip,name2:mac2:ip2
|
||||
# Example: desktop:AA:BB:CC:DD:EE:FF:192.168.1.10
|
||||
# Set WOL_DEVICES in .env – never commit real MACs/IPs
|
||||
_WOL_DEVICES_RAW = os.getenv("WOL_DEVICES", "")
|
||||
BROADCAST_IP = os.getenv("WOL_BROADCAST_IP", "255.255.255.255")
|
||||
WOL_PACKET_COUNT = int(os.getenv("WOL_PACKET_COUNT", "3"))
|
||||
|
||||
|
||||
def _parse_devices() -> tuple[dict[str, str], dict[str, str]]:
|
||||
"""Parse WOL_DEVICES env into devices (name->mac) and device_ips (name->ip).
|
||||
Format: name:mac:ip (mac has colons, so use name:XX:XX:XX:XX:XX:XX:ip)
|
||||
Example: desktop:AA:BB:CC:DD:EE:FF:192.168.1.10
|
||||
"""
|
||||
devices: dict[str, str] = {}
|
||||
device_ips: dict[str, str] = {}
|
||||
for entry in _WOL_DEVICES_RAW.split(","):
|
||||
entry = entry.strip()
|
||||
if not entry:
|
||||
continue
|
||||
parts = entry.split(":")
|
||||
if len(parts) >= 8:
|
||||
name = parts[0]
|
||||
mac = ":".join(parts[1:7])
|
||||
ip = parts[7]
|
||||
devices[name] = mac
|
||||
device_ips[name] = ip
|
||||
elif len(parts) == 3:
|
||||
name, mac, ip = parts
|
||||
devices[name] = mac
|
||||
device_ips[name] = ip
|
||||
return devices, device_ips
|
||||
|
||||
|
||||
DEVICES, DEVICE_IPS = _parse_devices()
|
||||
|
||||
|
||||
def _ping_host(ip: str) -> bool:
|
||||
try:
|
||||
subprocess.check_output(
|
||||
["ping", "-c", "1", "-W", "1", ip],
|
||||
stderr=subprocess.DEVNULL,
|
||||
)
|
||||
return True
|
||||
except (subprocess.CalledProcessError, FileNotFoundError):
|
||||
return False
|
||||
|
||||
|
||||
@app.route("/wake/<device>", methods=["POST"])
|
||||
def wake_device(device):
|
||||
if device not in DEVICES:
|
||||
return jsonify({"error": "Device not found"}), 404
|
||||
|
||||
mac_address = DEVICES[device]
|
||||
try:
|
||||
for _ in range(max(1, WOL_PACKET_COUNT)):
|
||||
send_magic_packet(mac_address, ip_address=BROADCAST_IP)
|
||||
app.logger.info(
|
||||
f"Magic packet(s) sent to {device} ({mac_address}) via {BROADCAST_IP}"
|
||||
)
|
||||
return jsonify({"message": f"Wake-on-LAN packet sent to {device}"}), 200
|
||||
except Exception as e:
|
||||
app.logger.exception(f"Error sending magic packet to {device}")
|
||||
return jsonify({"error": f"Failed to send wake packet: {str(e)}"}), 500
|
||||
|
||||
|
||||
@app.route("/devices", methods=["GET"])
|
||||
def list_devices():
|
||||
return jsonify({"devices": list(DEVICES.keys())})
|
||||
|
||||
|
||||
@app.route("/status", methods=["GET"])
|
||||
def device_status():
|
||||
"""Return devices with MAC, IP, and online status."""
|
||||
status = {
|
||||
name: {
|
||||
"mac": mac,
|
||||
"ip": DEVICE_IPS.get(name, ""),
|
||||
"online": _ping_host(DEVICE_IPS.get(name, "")),
|
||||
}
|
||||
for name, mac in DEVICES.items()
|
||||
}
|
||||
return jsonify({"devices": status})
|
||||
|
||||
|
||||
@app.route("/debug/<device>", methods=["GET"])
|
||||
def debug_device(device):
|
||||
if device not in DEVICES:
|
||||
return jsonify({"error": "Device not found"}), 404
|
||||
return jsonify({
|
||||
"device": device,
|
||||
"mac_address": DEVICES[device],
|
||||
"ip": DEVICE_IPS.get(device, ""),
|
||||
"broadcast_ip": BROADCAST_IP,
|
||||
"online": _ping_host(DEVICE_IPS.get(device, "")),
|
||||
})
|
||||
|
||||
|
||||
@app.route("/health", methods=["GET"])
|
||||
def health_check():
|
||||
return jsonify({"status": "healthy"}), 200
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
app.run(host="0.0.0.0", port=5000, debug=False)
|
||||
Reference in New Issue
Block a user