#!/usr/bin/env python3 """ Report Builder - Formats raw miner data dicts into readable text reports. """ from datetime import datetime def _fmt_hashrate(val, unit="GH/s"): """Format a hashrate value, normalizing to TH/s when appropriate.""" if val is None: return "N/A" try: val = float(val) except (TypeError, ValueError): return str(val) if unit == "MH/s": if val >= 1_000_000: return f"{val / 1_000_000:.2f} TH/s" elif val >= 1_000: return f"{val / 1_000:.2f} GH/s" return f"{val:.0f} MH/s" elif unit == "GH/s": if val >= 1_000: return f"{val / 1_000:.2f} TH/s" return f"{val:.2f} GH/s" return f"{val} {unit}" def _fmt_uptime(seconds): """Format uptime in seconds to human-readable string.""" if seconds is None: return "N/A" try: seconds = int(seconds) except (TypeError, ValueError): return str(seconds) days, rem = divmod(seconds, 86400) hours, rem = divmod(rem, 3600) minutes, _ = divmod(rem, 60) parts = [] if days: parts.append(f"{days}d") if hours: parts.append(f"{hours}h") parts.append(f"{minutes}m") return " ".join(parts) def _fmt_temp(val): """Format a temperature value with warning if >80C.""" if val is None: return "N/A" try: val = float(val) except (TypeError, ValueError): return str(val) warn = " *** HIGH ***" if val > 80 else "" return f"{val:.1f} C{warn}" def _v(data, key, default="N/A"): """Safe value getter.""" val = data.get(key) if val is None: return default return val def _section(title): """Section header.""" return f"\n{'=' * 60}\n {title}\n{'=' * 60}" def _collect_errors(data): """Collect all *_error keys from data dict.""" errors = [] for key, val in sorted(data.items()): if key.endswith("_error") and val: errors.append(f" {key}: {val}") return errors def format_report(data): """ Format a raw miner data dict into a readable text report. Works with output from collect_whatsminer. """ fw_type = data.get("firmware_type", "unknown") lines = [] # Title lines.append("=" * 60) lines.append(" PLUCKER MINER DIAGNOSTIC REPORT") lines.append(f" Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") lines.append("=" * 60) # ── Identification ── lines.append(_section("IDENTIFICATION")) lines.append(f" IP Address: {_v(data, 'ip')}") lines.append(f" MAC Address: {_v(data, 'mac')}") lines.append(f" Model: {_v(data, 'model')}") lines.append(f" Serial: {_v(data, 'serial')}") lines.append(f" Hostname: {_v(data, 'hostname')}") lines.append(f" Firmware: {_v(data, 'firmware')}") lines.append(f" Firmware Type: {fw_type}") if fw_type == "braiins": if data.get("is_braiins_plus") is not None: lines.append(f" BOS+: {'Yes' if data.get('is_braiins_plus') else 'No'}") if data.get("psu_model"): lines.append(f" PSU Model: {data['psu_model']}") if data.get("psu_serial"): lines.append(f" PSU Serial: {data['psu_serial']}") if fw_type == "whatsminer": if data.get("psu_model"): lines.append(f" PSU Model: {data['psu_model']}") # ── Status ── lines.append(_section("STATUS")) is_sleeping = data.get("is_sleeping", False) lines.append(f" Uptime: {_fmt_uptime(data.get('uptime_seconds'))}") if fw_type == "braiins" and data.get("bosminer_uptime") is not None: lines.append(f" BOSminer Up: {_fmt_uptime(data.get('bosminer_uptime'))}") lines.append(f" Sleeping: {'YES' if is_sleeping else 'No'}") if fw_type == "braiins" and data.get("miner_status") is not None: status_map = {0: "Unspecified", 1: "Not Started", 2: "Normal", 3: "Paused", 4: "Suspended", 5: "Restricted"} lines.append(f" Miner Status: {status_map.get(data['miner_status'], data['miner_status'])}") if fw_type == "whatsminer" and data.get("sleep_reason"): lines.append(f" Sleep Reason: {data['sleep_reason']}") # ── Hashrate ── lines.append(_section("HASHRATE")) if fw_type == "stock": lines.append(f" 5s: {_fmt_hashrate(data.get('hashrate_ghs_5s'), 'GH/s')}") lines.append(f" Average: {_fmt_hashrate(data.get('hashrate_ghs_avg'), 'GH/s')}") lines.append(f" 30m: {_fmt_hashrate(data.get('hashrate_ghs_30m'), 'GH/s')}") if data.get("target_hashrate") is not None: lines.append(f" Target: {_fmt_hashrate(data.get('target_hashrate'), 'GH/s')}") elif fw_type == "braiins": lines.append(f" 15s: {_fmt_hashrate(data.get('hashrate_ghs_15s'), 'GH/s')}") lines.append(f" 1m: {_fmt_hashrate(data.get('hashrate_ghs_1m'), 'GH/s')}") lines.append(f" 15m: {_fmt_hashrate(data.get('hashrate_ghs_15m'), 'GH/s')}") lines.append(f" 1h: {_fmt_hashrate(data.get('hashrate_ghs_1h'), 'GH/s')}") lines.append(f" 24h: {_fmt_hashrate(data.get('hashrate_ghs_24h'), 'GH/s')}") if data.get("target_hashrate_ghs") is not None: lines.append(f" Target: {_fmt_hashrate(data.get('target_hashrate_ghs'), 'GH/s')}") elif fw_type == "whatsminer": lines.append(f" 5s: {_fmt_hashrate(data.get('hashrate_mhs_5s'), 'MH/s')}") lines.append(f" Average: {_fmt_hashrate(data.get('hashrate_mhs_avg'), 'MH/s')}") lines.append(f" 1m: {_fmt_hashrate(data.get('hashrate_mhs_1m'), 'MH/s')}") lines.append(f" 15m: {_fmt_hashrate(data.get('hashrate_mhs_15m'), 'MH/s')}") if data.get("target_hashrate_mhs") is not None: lines.append(f" Target: {_fmt_hashrate(data.get('target_hashrate_mhs'), 'MH/s')}") if data.get("frequency") is not None: lines.append(f" Frequency: {data['frequency']} MHz") # ── Power (Braiins/Whatsminer) ── if data.get("power_watts") is not None or data.get("power_limit") is not None: lines.append(_section("POWER")) if data.get("power_watts") is not None: lines.append(f" Consumption: {data['power_watts']} W") if data.get("power_limit") is not None: lines.append(f" Power Limit: {data['power_limit']} W") if data.get("power_mode") is not None: lines.append(f" Power Mode: {data['power_mode']}") if data.get("efficiency_jth") is not None: lines.append(f" Efficiency: {data['efficiency_jth']} J/TH") if data.get("efficiency") is not None and data.get("efficiency_jth") is None: lines.append(f" Efficiency: {data['efficiency']}") # ── Shares ── lines.append(_section("SHARES")) accepted = data.get("accepted") rejected = data.get("rejected") lines.append(f" Accepted: {_v(data, 'accepted')}") lines.append(f" Rejected: {_v(data, 'rejected')}") if data.get("stale") is not None: lines.append(f" Stale: {data['stale']}") if data.get("hw_errors") is not None: lines.append(f" HW Errors: {data['hw_errors']}") # Reject rate if data.get("pool_rejected_pct") is not None: lines.append(f" Reject Rate: {data['pool_rejected_pct']}%") elif accepted is not None and rejected is not None: try: total = float(accepted) + float(rejected) if total > 0: lines.append(f" Reject Rate: {float(rejected) / total * 100:.2f}%") except (TypeError, ValueError): pass if data.get("pool_stale_pct") is not None: lines.append(f" Stale Rate: {data['pool_stale_pct']}%") # ── Temperatures ── lines.append(_section("TEMPERATURES")) if fw_type == "stock": if data.get("board_temps"): for i, t in enumerate(data["board_temps"], 1): lines.append(f" Board {i}: {_fmt_temp(t)}") if data.get("chip_temp_min") is not None: lines.append(f" Chip Min: {_fmt_temp(data['chip_temp_min'])}") if data.get("chip_temp_max") is not None: lines.append(f" Chip Max: {_fmt_temp(data['chip_temp_max'])}") if data.get("chip_temp_avg") is not None: lines.append(f" Chip Avg: {_fmt_temp(data['chip_temp_avg'])}") elif fw_type == "braiins": if data.get("board_temp_max") is not None: lines.append(f" Board Max: {_fmt_temp(data['board_temp_max'])}") if data.get("chip_temp_max") is not None: lines.append(f" Chip Max: {_fmt_temp(data['chip_temp_max'])}") elif fw_type == "whatsminer": if data.get("board_temp") is not None: lines.append(f" Board: {_fmt_temp(data['board_temp'])}") if data.get("chip_temp_min") is not None: lines.append(f" Chip Min: {_fmt_temp(data['chip_temp_min'])}") if data.get("chip_temp_max") is not None: lines.append(f" Chip Max: {_fmt_temp(data['chip_temp_max'])}") if data.get("chip_temp_avg") is not None: lines.append(f" Chip Avg: {_fmt_temp(data['chip_temp_avg'])}") if data.get("env_temp") is not None: lines.append(f" Environment: {_fmt_temp(data['env_temp'])}") # ── Fans ── lines.append(_section("FANS")) if fw_type == "whatsminer": fan_in = data.get("fan_in") fan_out = data.get("fan_out") if fan_in is not None: warn = " *** STOPPED ***" if fan_in == 0 else "" lines.append(f" Fan In: {fan_in} RPM{warn}") if fan_out is not None: warn = " *** STOPPED ***" if fan_out == 0 else "" lines.append(f" Fan Out: {fan_out} RPM{warn}") else: fans = data.get("fans", []) if fans: for fan in fans: label = fan.get("position", fan.get("index", "?")) rpm = fan.get("rpm", "N/A") warn = " *** STOPPED ***" if rpm == 0 else "" extra = "" if fan.get("speed_ratio") is not None: extra = f" ({fan['speed_ratio'] * 100:.0f}%)" lines.append(f" Fan {label}: {rpm} RPM{extra}{warn}") else: lines.append(" No fan data available") # ── Hashboards ── hashboards = data.get("hashboards", []) lines.append(_section("HASHBOARDS")) if data.get("total_chips") is not None: lines.append(f" Total Chips: {data['total_chips']}") if hashboards: for hb in hashboards: idx = hb.get("index", hb.get("id", hb.get("name", "?"))) lines.append(f"\n --- Board {idx} ---") if hb.get("chips") is not None: lines.append(f" Chips: {hb['chips']}") if hb.get("enabled") is not None: lines.append(f" Enabled: {hb['enabled']}") if hb.get("status") is not None: lines.append(f" Status: {hb['status']}") # Hashrate if fw_type == "whatsminer": if hb.get("hashrate_mhs_5s") is not None: lines.append(f" Hashrate 5s: {_fmt_hashrate(hb['hashrate_mhs_5s'], 'MH/s')}") if hb.get("hashrate_mhs_avg") is not None: lines.append(f" Hashrate Avg: {_fmt_hashrate(hb['hashrate_mhs_avg'], 'MH/s')}") elif fw_type == "braiins": if hb.get("hashrate_ghs_5s") is not None: lines.append(f" Hashrate 5s: {_fmt_hashrate(hb['hashrate_ghs_5s'], 'GH/s')}") if hb.get("hashrate_ghs_avg") is not None: lines.append(f" Hashrate Avg: {_fmt_hashrate(hb['hashrate_ghs_avg'], 'GH/s')}") elif fw_type == "stock": if hb.get("hashrate") is not None: lines.append(f" Hashrate: {hb['hashrate']}") # Frequency if hb.get("frequency") is not None or hb.get("frequency_mhz") is not None: freq = hb.get("frequency_mhz") or hb.get("frequency") lines.append(f" Frequency: {freq} MHz") # Temps for tkey, tlabel in [("temp", "Temp"), ("temp_chip", "Chip Temp"), ("temp_board", "Board Temp"), ("temp_inlet", "Inlet"), ("temp_outlet", "Outlet"), ("chip_temp_min", "Chip Min"), ("chip_temp_max", "Chip Max"), ("chip_temp_avg", "Chip Avg")]: if hb.get(tkey) is not None: lines.append(f" {tlabel + ':':<14}{_fmt_temp(hb[tkey])}") # Extra stock fields if hb.get("hw_errors") is not None: lines.append(f" HW Errors: {hb['hw_errors']}") if hb.get("chip_status") is not None: lines.append(f" Chip Status: {hb['chip_status']}") if hb.get("serial") is not None: lines.append(f" Serial: {hb['serial']}") else: lines.append(" No hashboard data available") # ── Pool Configuration ── pools = data.get("pools", []) lines.append(_section("POOL CONFIGURATION")) if pools: for i, pool in enumerate(pools, 1): lines.append(f"\n --- Pool {i} ---") lines.append(f" URL: {pool.get('url', 'N/A')}") lines.append(f" Worker: {pool.get('user', 'N/A')}") if pool.get("status") is not None: lines.append(f" Status: {pool['status']}") if pool.get("enabled") is not None: lines.append(f" Enabled: {pool['enabled']}") else: lines.append(" No pool data available") # ── Active Errors ── errors = data.get("errors", data.get("error_codes", [])) if errors: lines.append(_section("ACTIVE ERRORS")) for err in errors: if isinstance(err, dict): msg = err.get("message", "") codes = err.get("codes", []) if msg: lines.append(f" {msg}") if codes: lines.append(f" Codes: {', '.join(str(c) for c in codes)}") else: lines.append(f" {err}") # ── PSU (Whatsminer) ── if fw_type == "whatsminer" and any(data.get(k) is not None for k in ["psu_power_in", "psu_voltage_in", "psu_fan_speed", "psu_temp"]): lines.append(_section("POWER SUPPLY")) if data.get("psu_model"): lines.append(f" Model: {data['psu_model']}") if data.get("psu_power_in") is not None: lines.append(f" Power In: {data['psu_power_in']} W") if data.get("psu_voltage_in") is not None: lines.append(f" Voltage In: {data['psu_voltage_in']} mV") if data.get("psu_fan_speed") is not None: lines.append(f" Fan Speed: {data['psu_fan_speed']} RPM") if data.get("psu_temp") is not None: lines.append(f" Temperature: {_fmt_temp(data['psu_temp'])}") # ── Data Collection Notes ── collection_errors = _collect_errors(data) if collection_errors: lines.append(_section("DATA COLLECTION NOTES")) lines.append(" Some API calls failed during data collection:") for e in collection_errors: lines.append(e) lines.append("\n" + "=" * 60) lines.append(" END OF REPORT") lines.append("=" * 60) return "\n".join(lines)