feat(config,detector): TelegramCfg polling fields + Detector.step optional frame

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>
This commit is contained in:
Claude Agent
2026-04-17 10:17:17 +00:00
parent fd04fcd5e6
commit c1b89ad6a9
2 changed files with 20 additions and 4 deletions

View File

@@ -78,6 +78,9 @@ class DiscordCfg:
class TelegramCfg:
bot_token: str
chat_id: str
allowed_chat_ids: tuple[str, ...] = ()
poll_timeout_s: int = 30
auto_poll_interval_s: int = 180
def __post_init__(self) -> None:
if not self.bot_token or not self.chat_id:
@@ -156,9 +159,14 @@ class Config:
drift_threshold=int(data["canary"].get("drift_threshold", 8)),
)
discord = DiscordCfg(webhook_url=data["discord"]["webhook_url"])
tg = data["telegram"]
_allowed = [str(c) for c in tg.get("allowed_chat_ids", [])] or [str(tg["chat_id"])]
telegram = TelegramCfg(
bot_token=data["telegram"]["bot_token"],
chat_id=str(data["telegram"]["chat_id"]),
bot_token=tg["bot_token"],
chat_id=str(tg["chat_id"]),
allowed_chat_ids=tuple(_allowed),
poll_timeout_s=int(tg.get("poll_timeout_s", 30)),
auto_poll_interval_s=int(tg.get("auto_poll_interval_s", 180)),
)
opts = data.get("options", {})
region = None

View File

@@ -28,6 +28,7 @@ class DetectionResult:
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:
@@ -60,7 +61,13 @@ class Detector:
self._debounce: deque[str | None] = deque(maxlen=cfg.debounce_depth)
self._rolling: deque[DetectionResult] = deque(maxlen=20)
def step(self, ts: float) -> DetectionResult:
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:
@@ -117,6 +124,7 @@ class Detector:
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