use standard logging interface, only

Author: jquast

gambaterm/audio.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import atexit
+import logging
 import os
 from typing import Generator, Iterator, TYPE_CHECKING
 from contextlib import contextmanager
@@ -16,6 +17,8 @@
     import miniaudio
     import samplerate
 
+logger = logging.getLogger(__name__)
+
 
 class AudioOut:
     output_rate: float = 48000.0  # Hz
@@ -66,24 +69,22 @@ def __init__(
         # Diagnostics variables
         self._underruns = 0
         self._overruns = 0
-        self._diag_enabled = bool(
-            os.environ.get("GAMBATERM_AUDIO_CSV")
-            or os.environ.get("GAMBATERM_AUDIO_LOG")
-        )
-        if self._diag_enabled:
+        self._frame_num = 0
+        self._csv_enabled = bool(os.environ.get("GAMBATERM_AUDIO_CSV"))
+        self._diag_fill_min = 1.0
+        self._diag_ratio_min = self.nominal_sampling_ratio
+        self._diag_ratio_max = self.nominal_sampling_ratio
+        if self._csv_enabled:
             self._diag_frames: list[dict] = []
-            self._diag_frame_num = 0
-            self._diag_fill_min = 1.0
-            self._diag_ratio_min = self.nominal_sampling_ratio
-            self._diag_ratio_max = self.nominal_sampling_ratio
-            atexit.register(self._dump_audio_stats)
+            atexit.register(self._dump_csv)
+        atexit.register(self._log_summary)
 
         # Controller configuration
         self.correction_min = 1 - self.correction_clamp
         self.correction_max = 1 + self.correction_clamp
 
-        # Batch variable-length emulator output to avoid starving the
-        # ring buffer as runFor() sometimes returns partial frames.
+        # Batch the variable-length emulator audio output, avoids starving the ring buffer, runFor()
+        # sometimes returns partial frames!
         self._acc_buf: npt.NDArray[np.float32] = np.empty((0, 2), dtype=np.float32)
 
         # Controller state
@@ -107,17 +108,8 @@ def start(self) -> miniaudio.PlaybackDevice:
         device.start(stream)
         return device
 
-    def _dump_audio_stats(self) -> None:
-        if (audio_log := os.environ.get("GAMBATERM_AUDIO_LOG")):
-            with open(audio_log, "w") as fout:
-                fout.write(f"underruns={self._underruns}\n")
-                fout.write(f"overruns={self._overruns}\n")
-                fout.write(f"fill_min={self._diag_fill_min:.4f}\n")
-                fout.write(f"ratio_min={self._diag_ratio_min:.8f}\n")
-                fout.write(f"ratio_max={self._diag_ratio_max:.8f}\n")
-                _rng = self._diag_ratio_max - self._diag_ratio_min
-                fout.write(f"ratio_range={_rng:.8f}\n")
-        if (diag_csv := os.environ.get("GAMBATERM_AUDIO_CSV")):
+    def _dump_csv(self) -> None:
+        if diag_csv := os.environ.get("GAMBATERM_AUDIO_CSV"):
             with open(diag_csv, "w") as fout:
                 fout.write("frame,input,acc,proc,output,fill\n")
                 for _df in self._diag_frames:
@@ -126,6 +118,19 @@ def _dump_audio_stats(self) -> None:
                         f"{_df['proc']},{_df['output']},{_df['fill']:.4f}\n"
                     )
 
+    def _log_summary(self) -> None:
+        logger.debug(
+            "Audio stats: underruns=%d overruns=%d "
+            "fill_min=%.4f ratio_min=%.8f ratio_max=%.8f "
+            "ratio_range=%.8f",
+            self._underruns,
+            self._overruns,
+            self._diag_fill_min,
+            self._diag_ratio_min,
+            self._diag_ratio_max,
+            self._diag_ratio_max - self._diag_ratio_min,
+        )
+
     @property
     def fill_fraction(self) -> float:
         # Ring buffer fill ratio (0-1.0)
@@ -162,52 +167,68 @@ def adapt_sample_rate(self) -> None:
         self._diag_track_ratio()
 
     def _diag_record_skip(self, input_len: int, acc_len: int) -> None:
-        if not self._diag_enabled:
-            return
         fill = self.fill_fraction
         self._diag_fill_min = min(self._diag_fill_min, fill)
-        self._diag_frames.append({
-            "frame": self._diag_frame_num,
-            "input": input_len,
-            "acc": acc_len,
-            "proc": 0,
-            "output": 0,
-            "fill": fill,
-        })
-        self._diag_frame_num += 1
+        frame = self._frame_num
+        self._frame_num += 1
+        logger.debug(
+            "skip frame=%d input=%d acc=%d fill=%.4f", frame, input_len, acc_len, fill
+        )
+        if self._csv_enabled:
+            self._diag_frames.append(
+                {
+                    "frame": frame,
+                    "input": input_len,
+                    "acc": acc_len,
+                    "proc": 0,
+                    "output": 0,
+                    "fill": fill,
+                }
+            )
 
     def _diag_record_process(
-        self, input_len: int, acc_len: int, output_len: int,
+        self,
+        input_len: int,
+        acc_len: int,
+        output_len: int,
     ) -> None:
-        if not self._diag_enabled:
-            return
         fill = self.fill_fraction
         self._diag_fill_min = min(self._diag_fill_min, fill)
-        self._diag_frames.append({
-            "frame": self._diag_frame_num,
-            "input": input_len,
-            "acc": acc_len,
-            "proc": 1,
-            "output": output_len,
-            "fill": fill,
-        })
-        self._diag_frame_num += 1
+        frame = self._frame_num
+        self._frame_num += 1
+        logger.debug(
+            "process frame=%d input=%d acc=%d output=%d fill=%.4f",
+            frame,
+            input_len,
+            acc_len,
+            output_len,
+            fill,
+        )
+        if self._csv_enabled:
+            self._diag_frames.append(
+                {
+                    "frame": frame,
+                    "input": input_len,
+                    "acc": acc_len,
+                    "proc": 1,
+                    "output": output_len,
+                    "fill": fill,
+                }
+            )
 
     def _diag_track_fill(self) -> None:
-        if not self._diag_enabled:
-            return
         self._diag_fill_min = min(self._diag_fill_min, self.fill_fraction)
 
     def _diag_track_ratio(self) -> None:
-        if not self._diag_enabled:
-            return
         self._diag_ratio_min = min(self._diag_ratio_min, self.sampling_ratio)
         self._diag_ratio_max = max(self._diag_ratio_max, self.sampling_ratio)
 
     def send(self, console: Console, audio: npt.NDArray[np.int16]) -> None:
         # Scale and remove DC offset
-        scaled = (audio.astype(np.float32) * self.audio_volume
-                  - console.AUDIO_OFFSET * self.audio_volume)
+        scaled = (
+            audio.astype(np.float32) * self.audio_volume
+            - console.AUDIO_OFFSET * self.audio_volume
+        )
 
         # Accumulate input to batch variable-length emulator frames into consistent chunks for the
         # resampler.  The emulator's runFor() may produce anywhere from a few hundred to tens of
@@ -246,6 +267,12 @@ def send(self, console: Console, audio: npt.NDArray[np.int16]) -> None:
         # Drop excess frames if we're overrun
         if frames > space:
             self._overruns += 1
+            logger.warning(
+                "Audio overrun: dropping %d of %d frames (fill=%.2f)",
+                frames - space,
+                frames,
+                self.fill_fraction,
+            )
             resampled = resampled[:space]
             frames = space
 
@@ -317,6 +344,12 @@ def _audio_stream(self) -> Generator[bytes, int, None]:
             # Log if we're underrunning
             if read_size < required_frames:
                 self._underruns += 1
+                logger.warning(
+                    "Audio underrun: requested %d, got %d (fill=%.2f)",
+                    required_frames,
+                    read_size,
+                    self.fill_fraction,
+                )
 
             # Send audio to output and get next required frames
             required_frames = yield result.tobytes()

gambaterm/main.py

@@ -2,9 +2,10 @@
 from __future__ import annotations
 
 import time
+import logging
 import argparse
 from pathlib import Path
-from typing import ContextManager, TYPE_CHECKING
+from typing import Any, ContextManager, Optional, TYPE_CHECKING
 import dataclasses
 from dataclasses import dataclass, field
 
@@ -23,6 +24,45 @@
 if TYPE_CHECKING:
     from typing import Self
 
+_DEFAULT_LOGFMT = " ".join(("%(levelname)s", "%(filename)s:%(lineno)d", "%(message)s"))
+
+
+def make_logger(
+    name: str,
+    loglevel: str = "info",
+    logfile: Optional[str] = None,
+    logfmt: str = _DEFAULT_LOGFMT,
+    filemode: str = "a",
+) -> logging.Logger:
+    """Create and return a configured logger (following telnetlib3 pattern)."""
+    lvl = getattr(logging, loglevel.upper(), None)
+    if lvl is None:
+        lvl = logging.getLevelName(loglevel.upper())
+    _cfg: dict[str, Any] = {"format": logfmt}
+    if logfile:
+        _cfg["filename"] = logfile
+        _cfg["filemode"] = filemode
+    logging.basicConfig(**_cfg)
+    logging.getLogger().setLevel(lvl)
+    logging.getLogger(name).setLevel(lvl)
+    return logging.getLogger(name)
+
+
+def add_logging_arguments(parser: argparse.ArgumentParser) -> None:
+    parser.add_argument(
+        "--logfile", default=None, help="File path for log output (default: stderr)"
+    )
+    parser.add_argument(
+        "--loglevel",
+        default="warn",
+        help="Logging level: debug, info, warn, error (default: warn)",
+    )
+    parser.add_argument(
+        "--logfmt",
+        default=_DEFAULT_LOGFMT,
+        help="Log format string (default: LEVEL file:lineno message)",
+    )
+
 
 @dataclass
 class AppConfig:
@@ -151,11 +191,18 @@ def main(
     add_base_arguments(parser)
     add_input_file_arguments(parser)
     add_tuning_arguments(parser)
+    add_logging_arguments(parser)
     add_local_only_arguments(parser)
     console_cls.add_console_arguments(parser)
 
     # Parse arguments
     namespace = parser.parse_args(parser_args)
+    make_logger(
+        __name__,
+        loglevel=getattr(namespace, "loglevel", "warn"),
+        logfile=getattr(namespace, "logfile", None),
+        logfmt=getattr(namespace, "logfmt", _DEFAULT_LOGFMT),
+    )
     disable_audio = getattr(namespace, "disable_audio", False)
     args = LocalAppConfig.from_namespace(namespace)
 

gambaterm/ssh.py

@@ -38,6 +38,9 @@
     add_base_arguments,
     add_input_file_arguments,
     add_tuning_arguments,
+    add_logging_arguments,
+    make_logger,
+    _DEFAULT_LOGFMT,
     AppConfig,
 )
 from .console import Console, GameboyColor
@@ -533,6 +536,7 @@ def main(
     add_base_arguments(parser)
     add_input_file_arguments(parser)
     add_tuning_arguments(parser)
+    add_logging_arguments(parser)
     console_cls.add_console_arguments(parser)
     parser.add_argument(
         "--bind",
@@ -570,6 +574,12 @@ def main(
 
     # Parse arguments
     namespace = parser.parse_args(parser_args)
+    make_logger(
+        __name__,
+        loglevel=getattr(namespace, "loglevel", "warn"),
+        logfile=getattr(namespace, "logfile", None),
+        logfmt=getattr(namespace, "logfmt", _DEFAULT_LOGFMT),
+    )
     bind: str = namespace.__dict__.pop("bind")
     port: int = namespace.__dict__.pop("port")
     password: str = namespace.__dict__.pop("password")

gambaterm/telnet.py

@@ -32,6 +32,9 @@
     add_base_arguments,
     add_input_file_arguments,
     add_tuning_arguments,
+    add_logging_arguments,
+    make_logger,
+    _DEFAULT_LOGFMT,
     AppConfig,
 )
 from .console import Console, GameboyColor
@@ -444,6 +447,7 @@ def main(
     add_base_arguments(parser)
     add_input_file_arguments(parser)
     add_tuning_arguments(parser)
+    add_logging_arguments(parser)
     console_cls.add_console_arguments(parser)
     parser.add_argument(
         "--bind",
@@ -464,8 +468,7 @@ def main(
         "--robot-check",
         action="store_true",
         default=False,
-        help="reject bots by checking if client responds to "
-        "cursor position requests",
+        help="reject bots by checking if client responds to cursor position requests",
     )
     parser.add_argument(
         "--port",
@@ -488,6 +491,12 @@ def main(
     )
 
     namespace = parser.parse_args(parser_args)
+    make_logger(
+        __name__,
+        loglevel=getattr(namespace, "loglevel", "warn"),
+        logfile=getattr(namespace, "logfile", None),
+        logfmt=getattr(namespace, "logfmt", _DEFAULT_LOGFMT),
+    )
     bind: str = namespace.__dict__.pop("bind")
     port: int = namespace.__dict__.pop("port")
     robot_check: bool = namespace.__dict__.pop("robot_check")