from flask import Flask, render_template, request, jsonify, redirect, url_for from apscheduler.schedulers.background import BackgroundScheduler from datetime import datetime import subprocess import re import database as db app = Flask(__name__) # OIDs for Sierra Wireless RV50 OID_BYTES_SENT = '.1.3.6.1.4.1.20542.9.1.1.2.284.0' OID_BYTES_RECV = '.1.3.6.1.4.1.20542.9.1.1.2.283.0' # Store last readings for rate calculation last_readings = {} scheduler = BackgroundScheduler() scheduler.start() def format_bytes(bytes_value): """Convert bytes to human-readable format""" if bytes_value is None: return "N/A", 0 for unit in ['B', 'KB', 'MB', 'GB', 'TB']: if abs(bytes_value) < 1024.0: return f"{bytes_value:.2f} {unit}", bytes_value bytes_value /= 1024.0 return f"{bytes_value:.2f} PB", bytes_value def snmp_get(ip, port, community, oid): """Execute snmpget command and return value""" try: cmd = [ 'snmpget', '-v', '2c', '-c', community, '-t', '5', '-r', '2', f'{ip}:{port}', oid ] result = subprocess.run(cmd, capture_output=True, text=True, timeout=10) if result.returncode == 0: # Parse output: "SNMPv2-SMI::enterprises.20542.9.1.1.2.284.0 = Counter64: 123456" match = re.search(r'Counter64:\s*(\d+)', result.stdout) if match: return int(match.group(1)) return None except Exception as e: print(f"SNMP error: {e}") return None def poll_target(target_id, ip, port, community): """Poll a target and store the reading""" global last_readings bytes_sent = snmp_get(ip, port, community, OID_BYTES_SENT) bytes_recv = snmp_get(ip, port, community, OID_BYTES_RECV) if bytes_sent is not None and bytes_recv is not None: db.add_reading(target_id, bytes_sent, bytes_recv) # Calculate rates (bytes per second) rate_sent = 0 rate_recv = 0 if target_id in last_readings: last_sent, last_recv, last_time = last_readings[target_id] time_diff = (datetime.now() - last_time).total_seconds() if time_diff > 0: rate_sent = (bytes_sent - last_sent) / time_diff rate_recv = (bytes_recv - last_recv) / time_diff last_readings[target_id] = (bytes_sent, bytes_recv, datetime.now()) return { 'bytes_sent': bytes_sent, 'bytes_recv': bytes_recv, 'rate_sent': rate_sent, 'rate_recv': rate_recv, 'success': True } return {'success': False} def scheduled_poll(): """Background job to poll all enabled targets""" targets = db.get_all_targets() for target in targets: if target['enabled']: poll_target(target['id'], target['ip_address'], target['port'], target['community']) @app.route('/') def index(): targets = db.get_all_targets() target_data = [] for target in targets: latest = db.get_latest_reading(target['id']) if latest: sent_formatted, _ = format_bytes(latest['bytes_sent']) recv_formatted, _ = format_bytes(latest['bytes_recv']) # Calculate rate if we have previous reading rate_sent = 0 rate_recv = 0 if target['id'] in last_readings: last_sent, last_recv, last_time = last_readings[target['id']] time_diff = (datetime.now() - last_time).total_seconds() if time_diff > 0 and latest['bytes_sent'] >= last_sent: rate_sent = (latest['bytes_sent'] - last_sent) / time_diff rate_recv = (latest['bytes_recv'] - last_recv) / time_diff rate_sent_fmt, _ = format_bytes(rate_sent) rate_recv_fmt, _ = format_bytes(rate_recv) else: sent_formatted = recv_formatted = rate_sent_fmt = rate_recv_fmt = "N/A" target_data.append({ **target, 'bytes_sent_fmt': sent_formatted, 'bytes_recv_fmt': recv_formatted, 'rate_sent_fmt': f"{rate_sent_fmt}/s", 'rate_recv_fmt': f"{rate_recv_fmt}/s", 'last_poll': latest['timestamp'] if latest else None }) return render_template('index.html', targets=target_data) @app.route('/add', methods=['POST']) def add_target(): name = request.form.get('name', 'Unnamed') ip_address = request.form.get('ip_address') port = int(request.form.get('port', 161)) community = request.form.get('community', 'public') period_minutes = int(request.form.get('period_minutes', 5)) if ip_address: target_id = db.add_target(name, ip_address, port, community, period_minutes) # Schedule polling for this target scheduler.add_job( scheduled_poll, 'interval', minutes=period_minutes, id=f'target_{target_id}', replace_existing=True ) # Do initial poll poll_target(target_id, ip_address, port, community) return redirect(url_for('index')) @app.route('/edit/', methods=['POST']) def edit_target(target_id): name = request.form.get('name', 'Unnamed') ip_address = request.form.get('ip_address') port = int(request.form.get('port', 161)) community = request.form.get('community', 'public') period_minutes = int(request.form.get('period_minutes', 5)) enabled = 1 if request.form.get('enabled') else 0 db.update_target(target_id, name, ip_address, port, community, period_minutes, enabled) # Update scheduler if enabled: scheduler.add_job( scheduled_poll, 'interval', minutes=period_minutes, id=f'target_{target_id}', replace_existing=True ) else: try: scheduler.remove_job(f'target_{target_id}') except: pass return redirect(url_for('index')) @app.route('/delete/', methods=['POST']) def delete_target(target_id): try: scheduler.remove_job(f'target_{target_id}') except: pass db.delete_target(target_id) return redirect(url_for('index')) @app.route('/poll/', methods=['POST']) def manual_poll(target_id): target = db.get_target(target_id) if target: result = poll_target(target['id'], target['ip_address'], target['port'], target['community']) return jsonify(result) return jsonify({'success': False, 'error': 'Target not found'}), 404 @app.route('/graph/') def graph(target_id): target = db.get_target(target_id) hours = request.args.get('hours', 24, type=int) readings = db.get_readings(target_id, hours) labels = [] sent_data = [] recv_data = [] for r in readings: labels.append(r['timestamp']) sent_data.append(r['bytes_sent']) recv_data.append(r['bytes_recv']) return render_template('graph.html', target=target, labels=labels, sent_data=sent_data, recv_data=recv_data, hours=hours) @app.route('/api/readings/') def api_readings(target_id): hours = request.args.get('hours', 24, type=int) readings = db.get_readings(target_id, hours) data = [] for r in readings: data.append({ 'timestamp': r['timestamp'], 'bytes_sent': r['bytes_sent'], 'bytes_recv': r['bytes_recv'] }) return jsonify(data) if __name__ == '__main__': db.init_db() # Start with initial poll for any existing targets targets = db.get_all_targets() for target in targets: if target['enabled']: scheduler.add_job( scheduled_poll, 'interval', minutes=target['period_minutes'], id=f'target_{target["id"]}', replace_existing=True ) poll_target(target['id'], target['ip_address'], target['port'], target['community']) app.run(debug=True, host='0.0.0.0', port=5500)