|
# retoor <retoor@molodetz.nl>
|
|
|
|
import logging
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from devplacepy.config import BOT_DIR
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
MONITOR_DIR: Path = BOT_DIR / "monitor"
|
|
FRAME_SUFFIX = ".jpg"
|
|
ACTIVE_WINDOW_SECONDS = 90
|
|
|
|
|
|
@dataclass
|
|
class BotFrame:
|
|
slot: int
|
|
username: str = ""
|
|
persona: str = ""
|
|
action: str = ""
|
|
status: str = ""
|
|
url: str = ""
|
|
captured_at: float = 0.0
|
|
has_image: bool = False
|
|
|
|
def label(self) -> str:
|
|
return self.username or f"bot{self.slot}"
|
|
|
|
def age_seconds(self) -> float:
|
|
return max(0.0, time.time() - self.captured_at) if self.captured_at else 0.0
|
|
|
|
def is_active(self) -> bool:
|
|
return self.has_image and self.age_seconds() <= ACTIVE_WINDOW_SECONDS
|
|
|
|
def as_dict(self) -> dict:
|
|
return {
|
|
"slot": self.slot,
|
|
"username": self.username,
|
|
"persona": self.persona,
|
|
"action": self.action,
|
|
"status": self.status,
|
|
"url": self.url,
|
|
"label": self.label(),
|
|
"captured_at": int(self.captured_at) if self.captured_at else 0,
|
|
"age_seconds": int(self.age_seconds()),
|
|
"has_image": self.has_image,
|
|
"active": self.is_active(),
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class BotMonitor:
|
|
frames: dict[int, BotFrame] = field(default_factory=dict)
|
|
|
|
def frame_path(self, slot: int) -> Path:
|
|
return MONITOR_DIR / f"slot{slot}{FRAME_SUFFIX}"
|
|
|
|
def update_meta(
|
|
self,
|
|
slot: int,
|
|
*,
|
|
username: str = "",
|
|
persona: str = "",
|
|
action: str = "",
|
|
status: str = "",
|
|
url: str = "",
|
|
) -> None:
|
|
frame = self.frames.get(slot) or BotFrame(slot=slot)
|
|
if username:
|
|
frame.username = username
|
|
if persona:
|
|
frame.persona = persona
|
|
if action:
|
|
frame.action = action
|
|
if status:
|
|
frame.status = status
|
|
if url:
|
|
frame.url = url
|
|
self.frames[slot] = frame
|
|
|
|
def store_image(self, slot: int, image: bytes) -> None:
|
|
if not image:
|
|
return
|
|
frame = self.frames.get(slot) or BotFrame(slot=slot)
|
|
try:
|
|
MONITOR_DIR.mkdir(parents=True, exist_ok=True)
|
|
path = self.frame_path(slot)
|
|
tmp = path.with_suffix(path.suffix + ".tmp")
|
|
tmp.write_bytes(image)
|
|
tmp.replace(path)
|
|
except OSError as e:
|
|
logger.debug("monitor store_image slot %s failed: %s", slot, e)
|
|
return
|
|
frame.has_image = True
|
|
frame.captured_at = time.time()
|
|
self.frames[slot] = frame
|
|
|
|
def drop(self, slot: int) -> None:
|
|
self.frames.pop(slot, None)
|
|
try:
|
|
self.frame_path(slot).unlink(missing_ok=True)
|
|
except OSError as e:
|
|
logger.debug("monitor drop slot %s failed: %s", slot, e)
|
|
|
|
def read_image(self, slot: int) -> Optional[bytes]:
|
|
path = self.frame_path(slot)
|
|
if not path.exists():
|
|
return None
|
|
try:
|
|
return path.read_bytes()
|
|
except OSError as e:
|
|
logger.debug("monitor read_image slot %s failed: %s", slot, e)
|
|
return None
|
|
|
|
def snapshot(self) -> list[dict]:
|
|
return [self.frames[slot].as_dict() for slot in sorted(self.frames)]
|
|
|
|
|
|
monitor = BotMonitor()
|