67 lines
2.1 KiB
Python
67 lines
2.1 KiB
Python
from flask import Flask, render_template, jsonify
|
|
import subprocess
|
|
import platform
|
|
import json
|
|
import threading
|
|
import time
|
|
|
|
app = Flask(__name__)
|
|
|
|
def ping(host):
|
|
param = '-n' if platform.system().lower() == 'windows' else '-c'
|
|
command = ['ping', param, '1', host]
|
|
return subprocess.call(command, stdout=subprocess.DEVNULL) == 0
|
|
|
|
def load_config():
|
|
with open('config.json') as f:
|
|
return json.load(f)
|
|
|
|
device_status = {}
|
|
def update_status():
|
|
while True:
|
|
config = load_config()
|
|
for device in config['devices']:
|
|
current_status = ping(device['ip'])
|
|
previous_status = device_status.get(device['name'], True)
|
|
|
|
if current_status != previous_status:
|
|
diagnostics = run_diagnostics(device)
|
|
send_webhook(device, current_status, diagnostics)
|
|
|
|
device_status[device['name']] = current_status
|
|
time.sleep(config['check_interval'])
|
|
|
|
def send_webhook(device, status, diagnostics):
|
|
config = load_config()
|
|
webhook_data = {
|
|
"device": device,
|
|
"status": status,
|
|
"timestamp": datetime.now().isoformat(),
|
|
"diagnostics": diagnostics
|
|
}
|
|
requests.post(config['webhook_url'], json=webhook_data)
|
|
|
|
def run_diagnostics(device):
|
|
diagnostics = {}
|
|
if device['connection_type'] == 'fiber':
|
|
# Add your fiber diagnostic commands here
|
|
diagnostics['optical_power'] = subprocess.getoutput('optical-power-check ' + device['ip'])
|
|
diagnostics['light_levels'] = subprocess.getoutput('light-level-check ' + device['ip'])
|
|
else:
|
|
# Add your copper diagnostic commands here
|
|
diagnostics['cable_test'] = subprocess.getoutput('ethtool ' + device['ip'])
|
|
diagnostics['signal_quality'] = subprocess.getoutput('signal-quality-check ' + device['ip'])
|
|
return diagnostics
|
|
@app.route('/')
|
|
def home():
|
|
return render_template('index.html')
|
|
|
|
@app.route('/api/status')
|
|
def status():
|
|
return jsonify(device_status)
|
|
|
|
if __name__ == '__main__':
|
|
status_thread = threading.Thread(target=update_status, daemon=True)
|
|
status_thread.start()
|
|
app.run(debug=True)
|