#!/usr/bin/env python3 """ Plucker - Standalone Miner Diagnostic Tool Listens for UDP broadcasts from the Whatsminer IP Found button, collects diagnostics, and displays a formatted report (M60-only fleet). """ import os import sys import re import socket import struct import select import threading import logging import tkinter as tk from tkinter import scrolledtext, filedialog, messagebox from datetime import datetime from miner_apis import collect_whatsminer from report_builder import format_report logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", ) logger = logging.getLogger(__name__) # UDP ports PORT_WHATSMINER = 8888 # Whatsminer IP Found button (IPFOUND broadcast); NOT 4028 (that's the TCP API port) # Colors BG_DARK = "#1e1e1e" BG_WIDGET = "#2d2d2d" FG_TEXT = "#d4d4d4" FG_ACCENT = "#569cd6" FG_STATUS_OK = "#6a9955" FG_STATUS_WARN = "#d7ba7d" FG_STATUS_ERR = "#f44747" # --------------------------------------------------------------------------- # DPI awareness (Windows only) # --------------------------------------------------------------------------- def _set_dpi_awareness(): try: import ctypes ctypes.windll.shcore.SetProcessDpiAwareness(1) except Exception: pass # --------------------------------------------------------------------------- # MAC address parsing from UDP payloads # --------------------------------------------------------------------------- def _parse_mac_from_payload(payload): """Try multiple strategies to extract a MAC address from a UDP payload.""" if not payload: return None text = payload.decode("utf-8", errors="ignore") # Strategy 0: Whatsminer "MAC:" prefix. Must come first — a bare MAC regex # latches onto the "AC" in "MAC:" (e.g. "...10MAC:ab:cd:ef:ab:cd:ef" yields # "AC:ab:cd:ef:ab:cd", dropping the last octet). m = re.search(r"MAC:\s*(([0-9a-fA-F]{2}[:\-]){5}[0-9a-fA-F]{2})", text) if m: return m.group(1).upper().replace("-", ":") # Strategy 1: Standard text MAC (aa:bb:cc:dd:ee:ff or aa-bb-cc-dd-ee-ff) m = re.search(r"([0-9a-fA-F]{2}[:\-]){5}[0-9a-fA-F]{2}", text) if m: return m.group(0).upper().replace("-", ":") # Strategy 2: MAC without separators (12 hex chars) m = re.search(r"(?= 6: # Look for 6-byte sequences that look like a MAC (non-zero, first byte even for unicast) for offset in range(len(payload) - 5): candidate = payload[offset:offset + 6] if candidate != b"\x00" * 6 and candidate != b"\xff" * 6: mac = ":".join(f"{b:02X}" for b in candidate) return mac return None # --------------------------------------------------------------------------- # GUI Application # --------------------------------------------------------------------------- class PluckerApp: def __init__(self, root): self.root = root self.root.title("PLUCKER - Miner Diagnostic Tool") self.root.configure(bg=BG_DARK) self.root.minsize(700, 550) self.root.geometry("800x650") # Track scan thread self._scan_thread = None self._build_gui() self._start_listener() # ── GUI Layout ── def _build_gui(self): # Title bar title_frame = tk.Frame(self.root, bg=BG_DARK) title_frame.pack(fill=tk.X, padx=10, pady=(10, 0)) tk.Label( title_frame, text="PLUCKER", font=("Consolas", 20, "bold"), bg=BG_DARK, fg=FG_ACCENT, ).pack(side=tk.LEFT) self.status_label = tk.Label( title_frame, text="Listening...", font=("Consolas", 11), bg=BG_DARK, fg=FG_STATUS_OK, ) self.status_label.pack(side=tk.RIGHT) # Miner info bar info_frame = tk.Frame(self.root, bg=BG_WIDGET, relief=tk.FLAT, bd=1) info_frame.pack(fill=tk.X, padx=10, pady=(8, 0)) self.lbl_ip = tk.Label(info_frame, text="IP: --", font=("Consolas", 10), bg=BG_WIDGET, fg=FG_TEXT, anchor=tk.W) self.lbl_ip.pack(side=tk.LEFT, padx=10, pady=4) self.lbl_mac = tk.Label(info_frame, text="MAC: --", font=("Consolas", 10), bg=BG_WIDGET, fg=FG_TEXT, anchor=tk.W) self.lbl_mac.pack(side=tk.LEFT, padx=10, pady=4) self.lbl_fw = tk.Label(info_frame, text="Firmware: --", font=("Consolas", 10), bg=BG_WIDGET, fg=FG_TEXT, anchor=tk.W) self.lbl_fw.pack(side=tk.LEFT, padx=10, pady=4) # Report area self.report_text = scrolledtext.ScrolledText( self.root, wrap=tk.WORD, font=("Consolas", 10), bg=BG_WIDGET, fg=FG_TEXT, insertbackground=FG_TEXT, selectbackground=FG_ACCENT, relief=tk.FLAT, bd=0, state=tk.DISABLED, ) self.report_text.pack(fill=tk.BOTH, expand=True, padx=10, pady=8) # Button bar btn_frame = tk.Frame(self.root, bg=BG_DARK) btn_frame.pack(fill=tk.X, padx=10, pady=(0, 10)) btn_style = dict( font=("Consolas", 10), bg=BG_WIDGET, fg=FG_TEXT, activebackground=FG_ACCENT, activeforeground="white", relief=tk.FLAT, padx=12, pady=4, cursor="hand2", ) tk.Button(btn_frame, text="Copy to Clipboard", command=self._copy_report, **btn_style).pack(side=tk.LEFT, padx=(0, 6)) tk.Button(btn_frame, text="Save to File", command=self._save_report, **btn_style).pack(side=tk.LEFT, padx=(0, 6)) tk.Button(btn_frame, text="Clear", command=self._clear_report, **btn_style).pack(side=tk.LEFT) # Firewall hint self.hint_label = tk.Label( self.root, text="Hint: If no miners are detected, check your firewall allows UDP port 8888.", font=("Consolas", 8), bg=BG_DARK, fg="#666666", ) self.hint_label.pack(side=tk.BOTTOM, pady=(0, 4)) # ── Report text helpers ── def _set_report(self, text): self.report_text.config(state=tk.NORMAL) self.report_text.delete("1.0", tk.END) self.report_text.insert(tk.END, text) self.report_text.config(state=tk.DISABLED) self.report_text.see("1.0") def _append_report(self, text): self.report_text.config(state=tk.NORMAL) self.report_text.insert(tk.END, text) self.report_text.config(state=tk.DISABLED) def _set_status(self, text, color=FG_STATUS_OK): self.status_label.config(text=text, fg=color) def _set_info(self, ip=None, mac=None, fw=None): if ip is not None: self.lbl_ip.config(text=f"IP: {ip}") if mac is not None: self.lbl_mac.config(text=f"MAC: {mac}") if fw is not None: self.lbl_fw.config(text=f"Firmware: {fw}") # ── Button actions ── def _copy_report(self): self.report_text.config(state=tk.NORMAL) content = self.report_text.get("1.0", tk.END).strip() self.report_text.config(state=tk.DISABLED) if content: self.root.clipboard_clear() self.root.clipboard_append(content) self._set_status("Copied to clipboard", FG_STATUS_OK) def _save_report(self): self.report_text.config(state=tk.NORMAL) content = self.report_text.get("1.0", tk.END).strip() self.report_text.config(state=tk.DISABLED) if not content: return # Default to Desktop if available desktop = os.path.join(os.path.expanduser("~"), "Desktop") initial_dir = desktop if os.path.isdir(desktop) else os.path.expanduser("~") default_name = f"plucker_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt" filepath = filedialog.asksaveasfilename( initialdir=initial_dir, initialfile=default_name, defaultextension=".txt", filetypes=[("Text files", "*.txt"), ("All files", "*.*")], ) if filepath: with open(filepath, "w") as f: f.write(content) self._set_status(f"Saved: {os.path.basename(filepath)}", FG_STATUS_OK) def _clear_report(self): self._set_report("") self._set_info("--", "--", "--") self._set_status("Listening...", FG_STATUS_OK) # ── UDP Listener ── def _start_listener(self): thread = threading.Thread(target=self._udp_listener_loop, daemon=True) thread.start() def _udp_listener_loop(self): socks = [] ports = [PORT_WHATSMINER] for port in ports: try: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) try: s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) except AttributeError: pass # Windows doesn't support SO_REUSEPORT s.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) s.bind(("", port)) socks.append((s, port)) logger.info(f"Listening on UDP port {port}") except OSError as e: logger.error(f"Cannot bind UDP port {port}: {e}") self.root.after(0, lambda e=e, p=port: self._set_status( f"Port {p} bind failed - another instance running?", FG_STATUS_ERR )) if not socks: self.root.after(0, lambda: self._set_status( "Failed to bind any UDP port", FG_STATUS_ERR )) return logger.info(f"UDP listener active on {len(socks)} port(s)") while True: try: readable, _, _ = select.select([s for s, _ in socks], [], [], 1.0) for sock in readable: data, addr = sock.recvfrom(4096) ip = addr[0] port = sock.getsockname()[1] logger.info(f"UDP packet from {ip} on port {port} ({len(data)} bytes)") mac = _parse_mac_from_payload(data) self.root.after(0, self._on_miner_detected, ip, mac, port) except Exception as e: logger.error(f"UDP listener error: {e}") # ── Detection + Scan ── def _on_miner_detected(self, ip, mac, port): mac_str = mac or "detecting..." self._set_info(ip=ip, mac=mac_str, fw="detecting...") self._set_status("Detected - scanning...", FG_STATUS_WARN) self._set_report(f"Miner detected at {ip} (port {port})\nScanning...\n") # Cancel any existing scan self._scan_thread = threading.Thread( target=self._run_scan, args=(ip, mac), daemon=True ) self._scan_thread.start() def _run_scan(self, ip, mac): try: # Collect data (fleet is Whatsminer M60-only) self.root.after(0, self._set_info, None, None, "whatsminer") self.root.after(0, self._append_report, "Collecting data...\n") data = collect_whatsminer(ip) # Inject MAC from UDP if API didn't return one if mac and not data.get("mac"): data["mac"] = mac # Update MAC in info bar if we got it from API if data.get("mac"): self.root.after(0, self._set_info, None, data["mac"], None) # Step 3: Format report report = format_report(data) self.root.after(0, self._set_report, report) self.root.after(0, self._set_status, "Report ready", FG_STATUS_OK) except Exception as e: logger.exception(f"Scan failed for {ip}") self.root.after(0, self._set_report, f"Scan failed for {ip}:\n{e}") self.root.after(0, self._set_status, "Error", FG_STATUS_ERR) # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main(): _set_dpi_awareness() root = tk.Tk() app = PluckerApp(root) root.mainloop() if __name__ == "__main__": main()