#!/usr/bin/env python3 import cgi, sys sys.path.insert(0, '/opt/ngon/apps') sys.path.insert(0, '/var/www/html/ngon') from managers.auth_manager import AuthManager, generate_login_page_html from links import generate_dropdown_html, generate_dropdown_css, generate_dropdown_js _form = cgi.FieldStorage() _auth = AuthManager('miner_report') _auth_required, _should_exit, _headers = _auth.require_auth(_form) if _should_exit: print("Content-Type: application/json") if _headers: print(_headers) print("") if _auth_required: print('{"success": false, "error": "Authentication required"}') else: print('{"success": true}') sys.exit(0) if _auth_required: print("Content-Type: text/html") print("") print(generate_login_page_html("Miner Pod Report")) sys.exit(0) _user_access = AuthManager.get_user_access() """ Miner Report Viewer Web interface for viewing hourly miner pod snapshots from SQLite history. """ import cgi import json import sys import sqlite3 from datetime import datetime import cgitb cgitb.enable() sys.path.insert(0, '/opt/ngon') from apps.miners.error_categories import categorize_error_codes # Configuration DB_PATH = "/opt/ngon/data/miner_history.db" MASTER_CONFIG_PATH = "/opt/ngon/config/master_config.json" def load_pod_config(): """Load pod config (miner_count, miner_type) from master_config.json.""" try: with open(MASTER_CONFIG_PATH, 'r') as f: config = json.load(f) miner_specs = {} for miner_type, specs in config.get('misc', {}).get('miner_types', {}).items(): miner_specs[miner_type] = float(specs.get('spec_hashrate', 180.0)) pod_config = {} for site_name, site_data in config.get('sites', {}).items(): for group_name, group_data in site_data.get('generator_groups', {}).items(): for pod_name, pod_data in group_data.get('pods', {}).items(): miner_type = pod_data.get('miner_type', 'Unknown') spec_hs = miner_specs.get(miner_type, 180.0) pod_config[pod_name] = { 'miner_count': pod_data.get('miner_count', 0), 'miner_type': miner_type, 'spec_hashrate': spec_hs, 'low_hashrate_threshold': spec_hs * 0.8 } return pod_config except Exception as e: return {} def get_data_range(): """Get the min and max snapshot timestamps available in the DB.""" try: conn = sqlite3.connect(f'file:{DB_PATH}?immutable=1', uri=True) cur = conn.cursor() cur.execute("SELECT MIN(timestamp), MAX(timestamp) FROM miner_snapshots") row = cur.fetchone() conn.close() if row and row[0] and row[1]: return row[0], row[1] return None, None except Exception: return None, None def load_snapshot_data(target_dt, pod_config): """Load the nearest snapshot to target_dt, aggregated per pod with error categorization. target_dt: a datetime string like '2026-03-06T06:00' or a datetime object. Finds the closest snapshot timestamp to the target. """ if isinstance(target_dt, str): target_str = target_dt.replace('T', 'T') # already ISO-ish else: target_str = target_dt.isoformat() try: conn = sqlite3.connect(f'file:{DB_PATH}?immutable=1', uri=True) conn.row_factory = sqlite3.Row cur = conn.cursor() # Find the closest snapshot timestamp to the target # Check both directions and pick the nearest cur.execute(""" SELECT timestamp FROM ( SELECT DISTINCT timestamp FROM miner_snapshots WHERE timestamp <= ? ORDER BY timestamp DESC LIMIT 1 ) UNION ALL SELECT timestamp FROM ( SELECT DISTINCT timestamp FROM miner_snapshots WHERE timestamp > ? ORDER BY timestamp ASC LIMIT 1 ) """, (target_str, target_str)) candidates = cur.fetchall() if not candidates: conn.close() return None, "No snapshot data available" # Pick the closest snapshot_ts = min( [r['timestamp'] for r in candidates], key=lambda t: abs( datetime.fromisoformat(t).timestamp() - datetime.fromisoformat(target_str).timestamp() ) ) # Get all miners from that snapshot cur.execute(""" SELECT pod, miner_type, hs_rt, error_codes, mining_state, mac FROM miner_snapshots WHERE timestamp = ? """, (snapshot_ts,)) miners = cur.fetchall() conn.close() # Aggregate per pod pods_data = {} # Initialize all configured pods for pod_name, pc in pod_config.items(): pods_data[pod_name] = { 'pod_status': 'offline', 'miner_type': pc['miner_type'], 'miners_installed': pc['miner_count'], 'miners_online': 0, 'miners_hashing': 0, 'miners_low_hashrate': 0, 'miners_zero_hashrate': 0, 'miners_offline': 0, 'miners_sleeping': 0, 'miners_with_errors': 0, 'total_error_instances': 0, 'error_summary': { 'miners_with_fan_errors': 0, 'miners_with_power_errors': 0, 'miners_with_temp_sensor_errors': 0, 'miners_with_overtemp_protection': 0, 'miners_with_hashboard_errors': 0, 'miners_with_pool_errors': 0, 'miners_with_system_errors': 0, } } # Process each miner row for miner in miners: pod = miner['pod'] if pod not in pods_data: # Pod in DB but not in config - create entry pods_data[pod] = { 'pod_status': 'offline', 'miner_type': miner['miner_type'] or 'Unknown', 'miners_installed': 0, 'miners_online': 0, 'miners_hashing': 0, 'miners_low_hashrate': 0, 'miners_zero_hashrate': 0, 'miners_offline': 0, 'miners_sleeping': 0, 'miners_with_errors': 0, 'total_error_instances': 0, 'error_summary': { 'miners_with_fan_errors': 0, 'miners_with_power_errors': 0, 'miners_with_temp_sensor_errors': 0, 'miners_with_overtemp_protection': 0, 'miners_with_hashboard_errors': 0, 'miners_with_pool_errors': 0, 'miners_with_system_errors': 0, } } pd = pods_data[pod] hs_rt = miner['hs_rt'] or 0.0 mining_state = (miner['mining_state'] or '').lower() if mining_state == 'sleeping': pd['miners_sleeping'] += 1 else: pd['miners_online'] += 1 if hs_rt > 0: pd['miners_hashing'] += 1 # Check low hashrate pc = pod_config.get(pod, {}) threshold = pc.get('low_hashrate_threshold', 144.0) if hs_rt < threshold: pd['miners_low_hashrate'] += 1 else: pd['miners_zero_hashrate'] += 1 # Mark pod as online if any non-sleeping miner present pd['pod_status'] = 'online' # Error categorization error_codes_str = miner['error_codes'] or '' cat = categorize_error_codes(error_codes_str) if cat['has_errors']: pd['miners_with_errors'] += 1 pd['total_error_instances'] += cat['total_error_count'] for key in pd['error_summary']: pd['error_summary'][key] += cat[key] # Calculate offline counts for pod_name, pd in pods_data.items(): pd['miners_offline'] = max(0, pd['miners_installed'] - pd['miners_online'] - pd['miners_sleeping']) return {'pods': pods_data, 'generated_at': snapshot_ts}, None except Exception as e: return None, f"Database error: {str(e)}" def format_percentage(value, total): """Format a percentage value.""" if total == 0: return "0%" return f"{(value/total)*100:.1f}%" def get_status_color(pod_status, miners_hashing, miners_installed): """Get color class for pod status.""" if pod_status == "offline": return "offline" elif miners_installed == 0: return "offline" elif miners_hashing == 0: return "critical" elif miners_hashing / miners_installed < 0.5: return "warning" else: return "healthy" def render_pod_table(pods_data): """Render the main pod table.""" html = """ """ # Sort pods by name sorted_pods = sorted(pods_data.items(), key=lambda x: x[0]) for pod_name, pod_data in sorted_pods: status_class = get_status_color( pod_data['pod_status'], pod_data['miners_hashing'], pod_data['miners_installed'] ) errors = pod_data['error_summary'] total_errors = sum(errors.values()) html += f""" """ html += """
Pod Name Type Status Installed Online Hashing Low Hash Zero Hash Offline Sleeping Miners w/ Errors Total Errors Error Summary
Fan Power Temp Sensor Over Temp Board Pool System
{pod_name} {pod_data['miner_type']} {pod_data['pod_status']} {pod_data['miners_installed']} {pod_data['miners_online']} {pod_data['miners_hashing']} {pod_data['miners_low_hashrate']} {pod_data['miners_zero_hashrate']} {pod_data['miners_offline']} {pod_data.get('miners_sleeping', 0)} {pod_data.get('miners_with_errors', 0)} {pod_data.get('total_error_instances', 0)} {errors['miners_with_fan_errors']} {errors['miners_with_power_errors']} {errors.get('miners_with_temp_sensor_errors', 0)} {errors.get('miners_with_overtemp_protection', 0)} {errors['miners_with_hashboard_errors']} {errors['miners_with_pool_errors']} {errors['miners_with_system_errors']}
""" return html def main(): print("Content-Type: text/html\n") form = cgi.FieldStorage() selected_dt = form.getvalue('dt', '') pod_config = load_pod_config() data_min, data_max = get_data_range() # Load report data if selected_dt: report_data, error = load_snapshot_data(selected_dt, pod_config) elif data_max: report_data, error = load_snapshot_data(data_max, pod_config) else: report_data, error = None, "No snapshot data available in database" print(""" Miner Pod Reports - NGON Mining """) print(f"""

NGON Mining - Miner Pod Report

{generate_dropdown_html(_user_access)}
Hourly mining operation status and error analysis
""") if error: print(f'
Error: {error}
') elif report_data: # Render date/time picker snapshot_ts = report_data.get('generated_at', '') # Format current snapshot time for the input value if snapshot_ts: try: picker_value = datetime.fromisoformat(snapshot_ts).strftime('%Y-%m-%dT%H:%M') except Exception: picker_value = selected_dt or '' else: picker_value = selected_dt or '' min_attr = f'min="{datetime.fromisoformat(data_min).strftime("%Y-%m-%dT%H:%M")}"' if data_min else '' max_attr = f'max="{datetime.fromisoformat(data_max).strftime("%Y-%m-%dT%H:%M")}"' if data_max else '' print(f'''
Latest
''') # Calculate summary statistics pods_data = report_data.get('pods', {}) total_pods = len(pods_data) online_pods = sum(1 for pod in pods_data.values() if pod['pod_status'] == 'online') total_installed = sum(pod['miners_installed'] for pod in pods_data.values()) total_hashing = sum(pod['miners_hashing'] for pod in pods_data.values()) total_errors = sum(sum(pod['error_summary'].values()) for pod in pods_data.values()) # Render summary cards print('
') print(f'''

Total Pods

{total_pods}

Pods Online

{online_pods}

Miners Installed

{total_installed:,}

Miners Hashing

{total_hashing:,}

Miners with Errors

{total_errors:,}

Hash Rate %

{format_percentage(total_hashing, total_installed)}
''') print('
') # Render main table (wrapped so it gets the panel/border/radius) print('
') print(render_pod_table(pods_data)) print('
') # Render timestamp generated_at = report_data.get('generated_at', '') if generated_at: try: dt = datetime.fromisoformat(generated_at.replace('Z', '+00:00')) formatted_time = dt.strftime('%Y-%m-%d %H:%M:%S UTC') print(f'
Snapshot taken: {formatted_time}
') except Exception: print(f'
Snapshot taken: {generated_at}
') else: print('
No report data available
') print(f""" """) if __name__ == "__main__": main()