c1b89ad6a9
TelegramCfg gains allowed_chat_ids (default: [chat_id]), poll_timeout_s=30, auto_poll_interval_s=180. _from_dict reads from TOML; absent key defaults to primary chat_id so existing configs need no changes. Detector.step(ts, frame=None): when frame is provided the capture() call is skipped — async loop pre-captures once, shares frame between canary+detection. DetectionResult.dot_pos_abs carries absolute (x,y) for price overlay. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
135 lines
4.1 KiB
Python
135 lines
4.1 KiB
Python
"""Per-cycle dot detector with debounce and rolling window."""
|
|
from __future__ import annotations
|
|
|
|
from collections import deque
|
|
from dataclasses import dataclass
|
|
from typing import Callable
|
|
|
|
import numpy as np
|
|
|
|
from .config import Config
|
|
from .vision import (
|
|
ColorMatch,
|
|
classify_pixel,
|
|
crop_roi,
|
|
find_rightmost_dot,
|
|
pixel_rgb,
|
|
)
|
|
|
|
ScreenCapture = Callable[[], "np.ndarray | None"]
|
|
|
|
|
|
@dataclass
|
|
class DetectionResult:
|
|
ts: float
|
|
window_found: bool
|
|
dot_found: bool
|
|
rgb: tuple[int, int, int] | None
|
|
match: ColorMatch | None # None if no dot
|
|
accepted: bool # post-debounce; True only when match repeats debounce_depth times
|
|
color: str | None # accepted color name (UNKNOWN excluded)
|
|
dot_pos_abs: tuple[int, int] | None = None # absolute (x, y) in frame; set when dot_found
|
|
|
|
|
|
class Detector:
|
|
"""Capture → crop → find dot → classify → debounce → emit."""
|
|
|
|
def __init__(
|
|
self,
|
|
cfg: Config,
|
|
capture: ScreenCapture,
|
|
bg_rgb: tuple[int, int, int] | None = None,
|
|
bg_tol: float | None = None,
|
|
) -> None:
|
|
self._cfg = cfg
|
|
self._capture = capture
|
|
# Prefer config-defined background; fall back to dark-grey default.
|
|
if "background" in cfg.colors:
|
|
spec = cfg.colors["background"]
|
|
self._bg_rgb = bg_rgb if bg_rgb is not None else spec.rgb
|
|
self._bg_tol = bg_tol if bg_tol is not None else spec.tolerance
|
|
else:
|
|
self._bg_rgb = bg_rgb if bg_rgb is not None else (18, 18, 18)
|
|
self._bg_tol = bg_tol if bg_tol is not None else 15.0
|
|
# Palette excludes "background" key
|
|
self._palette: dict[str, tuple[tuple[int, int, int], float]] = {
|
|
name: (spec.rgb, spec.tolerance)
|
|
for name, spec in cfg.colors.items()
|
|
if name != "background"
|
|
}
|
|
# maxlen enforces "last N" automatically
|
|
self._debounce: deque[str | None] = deque(maxlen=cfg.debounce_depth)
|
|
self._rolling: deque[DetectionResult] = deque(maxlen=20)
|
|
|
|
def step(self, ts: float, frame=None) -> DetectionResult:
|
|
"""Run one detection tick.
|
|
|
|
frame: pre-captured BGR ndarray (from asyncio.to_thread capture). When
|
|
None (default), calls self._capture() — preserving the sync-loop behaviour.
|
|
"""
|
|
if frame is None:
|
|
frame = self._capture()
|
|
|
|
if frame is None:
|
|
self._debounce.append(None)
|
|
r = DetectionResult(
|
|
ts=ts,
|
|
window_found=False,
|
|
dot_found=False,
|
|
rgb=None,
|
|
match=None,
|
|
accepted=False,
|
|
color=None,
|
|
)
|
|
self._rolling.append(r)
|
|
return r
|
|
|
|
roi_img = crop_roi(frame, self._cfg.dot_roi)
|
|
dot_pos = find_rightmost_dot(roi_img, self._bg_rgb, self._bg_tol)
|
|
|
|
if dot_pos is None:
|
|
self._debounce.append(None)
|
|
r = DetectionResult(
|
|
ts=ts,
|
|
window_found=True,
|
|
dot_found=False,
|
|
rgb=None,
|
|
match=None,
|
|
accepted=False,
|
|
color=None,
|
|
)
|
|
self._rolling.append(r)
|
|
return r
|
|
|
|
x, y = dot_pos
|
|
rgb = pixel_rgb(roi_img, x, y)
|
|
match = classify_pixel(rgb, self._palette)
|
|
self._debounce.append(match.name)
|
|
|
|
accepted = False
|
|
color: str | None = None
|
|
if (
|
|
len(self._debounce) == self._cfg.debounce_depth
|
|
and all(m == match.name for m in self._debounce)
|
|
and match.name != "UNKNOWN"
|
|
):
|
|
accepted = True
|
|
color = match.name
|
|
|
|
r = DetectionResult(
|
|
ts=ts,
|
|
window_found=True,
|
|
dot_found=True,
|
|
rgb=rgb,
|
|
match=match,
|
|
accepted=accepted,
|
|
color=color,
|
|
dot_pos_abs=(self._cfg.dot_roi.x + x, self._cfg.dot_roi.y + y),
|
|
)
|
|
self._rolling.append(r)
|
|
return r
|
|
|
|
@property
|
|
def rolling(self) -> list[DetectionResult]:
|
|
return list(self._rolling)
|