serial_reader.py

#!/usr/bin/env python3
"""
# Serial Reader for NerdMiner V2 ESP32-based miners.
# Connects to the serial port, reads lines, parses key metrics, and displays a live
# dashboard in the terminal using Rich. Supports three display modes:
# 1. Full dashboard with all metrics and log.
# 2. Compact view with key metrics only.
# 3. Time-focused view with large digital clock and recent log.
#
"""

import argparse
import re
import sys
import threading
import time
from collections import deque
from datetime import datetime

import serial
from rich import box
from rich.console import Console
from rich.layout import Layout
from rich.live import Live
from rich.panel import Panel
from rich.text import Text
from rich.align import Align

# ================= CONFIG =================
DEFAULT_PORT = "/dev/ttyACM0"
DEFAULT_BAUD = 115200
TIMEOUT = 0.5
MAX_LOG_LINES = 80
SNAPSHOT_INTERVAL = 10
REFRESH_PER_SECOND = 5

_PAT_TEMPERATURE = re.compile(r"Temperature:\s*([+-]?\d+)")
_PAT_MHASHES = re.compile(r"Total MHashes:\s*([+-]?\d+)")
_PAT_32SHARES = re.compile(r"32Bit\s*shares:\s*([+-]?\d+)")
_PAT_BLOCKS = re.compile(r"Valid blocks:\s*([+-]?\d+)")
_PAT_DIFF = re.compile(r"Best difficulty:\s*([0-9]*\.?[0-9]+)")

console = Console()


# ================= Share Tracker =================
class ShareTracker:
    def __init__(self):
        self.last_share_count = None
        self.last_share_time = None
        self.time_between_shares = None

    def update(self, current_count: int):
        now = time.time()
        if self.last_share_count is None:
            self.last_share_count = current_count
            self.last_share_time = now
            return
        if current_count > self.last_share_count:
            self.time_between_shares = now - self.last_share_time
            self.last_share_time = now
        self.last_share_count = current_count

    def get_display(self):
        return f"{self.time_between_shares:.2f} s" if self.time_between_shares else "N/A"


# ================= Dashboard =================
class Dashboard:
    def __init__(self):
        self.temp = None
        self.mhashes = None
        self.shares_32 = None
        self.blocks = None
        self.difficulty = None
        self.log_lines = deque(maxlen=MAX_LOG_LINES)

        self.share_tracker = ShareTracker()
        self.snapshots = deque()
        self.last_snapshot_time = 0

    def ingest_line(self, line: str):
        line = line.strip()
        if not line:
            return
        self.log_lines.append(line)

        m = _PAT_TEMPERATURE.search(line)
        if m:
            self.temp = int(m.group(1))

        m = _PAT_MHASHES.search(line)
        if m:
            self.mhashes = int(m.group(1))
            self._maybe_snapshot()

        m = _PAT_32SHARES.search(line)
        if m:
            self.shares_32 = int(m.group(1))
            self.share_tracker.update(self.shares_32)

        m = _PAT_BLOCKS.search(line)
        if m:
            self.blocks = int(m.group(1))

        m = _PAT_DIFF.search(line)
        if m:
            self.difficulty = float(m.group(1))

    def _maybe_snapshot(self):
        now = time.time()
        if now - self.last_snapshot_time >= SNAPSHOT_INTERVAL:
            if self.mhashes is not None:
                self.snapshots.append((now, self.mhashes))
            self.last_snapshot_time = now

    def _average_hashrate(self, seconds: int):
        cutoff = time.time() - seconds
        relevant = [(t, h) for t, h in self.snapshots if t >= cutoff]
        if len(relevant) < 2:
            return 0.0
        start_time, start_val = relevant[0]
        end_time, end_val = relevant[-1]
        delta_h = end_val - start_val
        delta_t = end_time - start_time
        return (delta_h * 1000) / delta_t if delta_t > 0 else 0.0

    # ---- Panels ----
    def _panel_style_for_temp(self):
        if self.temp is None:
            return "white"
        if self.temp >= 75:
            return "bold red"
        if self.temp >= 65:
            return "yellow"
        return "green"

    def _render_top_panels(self):
        panels = [
            Panel(Text(str(self.temp or "N/A"), justify="center", style=self._panel_style_for_temp()), title="Temp C", box=box.ROUNDED),
            Panel(Text(str(self.mhashes or "N/A"), justify="center", style="green"), title="Total MHashes", box=box.ROUNDED),
            Panel(Text(str(self.shares_32 or "N/A"), justify="center", style="cyan"), title="32BitShares", box=box.ROUNDED),
            Panel(Text(str(self.blocks or "N/A"), justify="center", style="magenta"), title="Valid Blocks", box=box.ROUNDED),
            Panel(Text(f"{self.difficulty:.4f}" if self.difficulty else "N/A", justify="center"), title="Difficulty", box=box.ROUNDED),
            Panel(Text(self.share_tracker.get_display(), justify="center", style="yellow"), title="Time Since 32BitShare", box=box.ROUNDED),
            Panel(Text(f"{self._average_hashrate(3600):.2f} kH/s", justify="center", style="green"), title="1h Avg", box=box.ROUNDED),
            Panel(Text(f"{self._average_hashrate(86400):.2f} kH/s", justify="center", style="green"), title="24h Avg", box=box.ROUNDED),
            Panel(Text(f"{self._average_hashrate(604800):.2f} kH/s", justify="center", style="green"), title="7d Avg", box=box.ROUNDED),
        ]
        return panels

    def _render_compact(self):
        panels = [
            Panel(Text(str(self.blocks or "N/A"), justify="center", style="magenta"), title="Valid Blocks", box=box.ROUNDED),
            Panel(Text(f"{self.difficulty:.4f}" if self.difficulty else "N/A", justify="center"), title="Difficulty", box=box.ROUNDED),
        ]
        return panels

    

    _DIGITS = {
        "0": [" ███ ",
            "█   █",
            "█   █",
            "█   █",
            " ███ "],
        "1": ["  █  ",
            " ██  ",
            "  █  ",
            "  █  ",
            " ███ "],
        "2": [" ███ ",
            "    █",
            " ███ ",
            "█    ",
            "█████"],
        "3": ["████ ",
            "    █",
            " ███ ",
            "    █",
            "████ "],
        "4": ["█  █ ",
            "█  █ ",
            "█████",
            "   █ ",
            "   █ "],
        "5": ["█████",
            "█    ",
            "████ ",
            "    █",
            "████ "],
        "6": [" ███ ",
            "█    ",
            "████ ",
            "█   █",
            " ███ "],
        "7": ["█████",
            "    █",
            "   █ ",
            "  █  ",
            "  █  "],
        "8": [" ███ ",
            "█   █",
            " ███ ",
            "█   █",
            " ███ "],
        "9": [" ███ ",
            "█   █",
            " ████",
            "    █",
            " ███ "],
        ":": ["     ",
            "  █  ",
            "     ",
            "  █  ",
            "     "],
    }

    def _render_large_time(self):
        now_str = datetime.now().strftime("%H:%M:%S")
        rows = [""] * 5
        for c in now_str:
            digit = self._DIGITS.get(c, ["     "] * 5)
            for i in range(5):
                rows[i] += digit[i] + "  "
        return "\n".join(rows)

    def _render_time_focused(self):
        # Large digital clock
        clock_text = self._render_large_time()
        clock_panel = Panel(Align.center(Text(clock_text, style="bold cyan")), title="Time", box=box.ROUNDED)

        metrics_panel = Panel(Text(f"Valid Blocks: {self.blocks or 'N/A'}\n"
                                   f"Total MHashes: {self.mhashes or 'N/A'}\n"
                                   f"Difficulty: {self.difficulty or 'N/A'}\n"
                                   f"32BitShares: {self.shares_32 or 'N/A'}\n"
                                   f"Time since 32BitShare: {self.share_tracker.get_display()}\n",
                                   justify="center"), title="Metrics", box=box.ROUNDED)
        # Recent log (fill remaining space, show many lines; layout will crop as needed)
        recent_log_panel = self._render_recent_log(lines=MAX_LOG_LINES)
        return [clock_panel, metrics_panel, recent_log_panel]

    def _render_log(self):
        text = "\n".join(list(self.log_lines)[-MAX_LOG_LINES:])
        return Panel(text or "(no lines yet)", title="Serial Log", box=box.ROUNDED, height=12)

    def _render_recent_log(self, lines: int | None = None):
        """Render a recent log panel with the last N lines (colorized). If lines is None, show all available up to MAX_LOG_LINES."""
        count = min(len(self.log_lines), lines if isinstance(lines, int) else MAX_LOG_LINES)
        recent = list(self.log_lines)[-count:] if self.log_lines else []

        # Build a colorized Text block
        from rich.text import Text as RichText
        rich_text = RichText()
        for ln in recent:
            style = None
            if _PAT_TEMPERATURE.search(ln):
                # Color temperature line based on parsed value if possible
                try:
                    val = int(_PAT_TEMPERATURE.search(ln).group(1))
                    style = "bold red" if val >= 75 else ("yellow" if val >= 65 else "green")
                except Exception:
                    style = "yellow"
            elif _PAT_32SHARES.search(ln):
                style = "cyan"
            elif _PAT_BLOCKS.search(ln):
                style = "magenta"
            elif _PAT_MHASHES.search(ln):
                style = "green"
            elif _PAT_DIFF.search(ln):
                style = "bright_white"
            else:
                style = "white"
            rich_text.append(ln, style=style)
            rich_text.append("\n")

        return Panel(rich_text if recent else ("(no lines yet)"), title="Recent Log", box=box.ROUNDED)

    def render(self, screen_type=1):
        layout = Layout()
        if screen_type == 1:
            layout.split_column(
                Layout(name="top_panels", size=9),
                Layout(name="bottom")
            )
            panels = self._render_top_panels()
            layout["top_panels"].split_row(*panels)
            layout["bottom"].update(self._render_log())
        elif screen_type == 2:
            layout.update(Layout())
            panels = self._render_compact()
            layout.split_row(*panels)
        elif screen_type == 3:
            layout.update(Layout())
            panels = self._render_time_focused()
            layout.split_column(*panels)
        return layout


# ================= Serial reader =================
def serial_iter(port: str, baud: int):
    ser = serial.Serial(port, baud, timeout=TIMEOUT)
    try:
        while True:
            raw = ser.readline()
            if not raw:
                yield None
                continue
            try:
                line = raw.decode("utf-8", errors="replace")
            except Exception:
                line = raw.decode(errors="ignore")
            yield line.rstrip("\n\r")
    finally:
        try:
            ser.close()
        except Exception:
            pass


# ================= Keypress thread =================
def keypress_listener(screen_mode_ref):
    import termios, tty
    stdin_fd = sys.stdin.fileno()
    old_settings = termios.tcgetattr(stdin_fd)
    try:
        tty.setcbreak(stdin_fd)
        while True:
            ch = sys.stdin.read(1)
            if ch == " ":
                screen_mode_ref[0] = (screen_mode_ref[0] % 3) + 1
    finally:
        termios.tcsetattr(stdin_fd, termios.TCSADRAIN, old_settings)


# ================= Main =================
def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--port", default=DEFAULT_PORT)
    parser.add_argument("--baud", type=int, default=DEFAULT_BAUD)
    parser.add_argument("--display_type", type=int, default=1, choices=[1, 2, 3])
    args = parser.parse_args()

    dash = Dashboard()
    serial_lines = serial_iter(args.port, args.baud)

    screen_mode_ref = [args.display_type]

    # start keypress listener
    threading.Thread(target=keypress_listener, args=(screen_mode_ref,), daemon=True).start()

    with Live(dash.render(screen_mode_ref[0]), console=console, refresh_per_second=REFRESH_PER_SECOND) as live:
        for raw in serial_lines:
            if raw is not None:
                dash.ingest_line(raw)
            live.update(dash.render(screen_mode_ref[0]))


if __name__ == "__main__":
    main()