Compare commits

..

No commits in common. "master" and "v0.7.5" have entirely different histories.

7 changed files with 50 additions and 312 deletions

102
README.md
View file

@ -1,4 +1,4 @@
# Anti Prestige Tool v0.9.0 # Anti Prestige Tool v0.7.5
Queue-position reader for Arma Reforger on Linux Wayland. Queue-position reader for Arma Reforger on Linux Wayland.
@ -30,98 +30,30 @@ make
No sudo install step is required. No sudo install step is required.
## Standard Run ## Use
Watch the selected Reforger window and alert when the queue position reaches `5` or lower: Run:
```bash
uv run reforger_queue_read.py --watch --portal-window
```
On first run, your desktop should show a window-sharing picker. Select the Arma Reforger window. Later runs try to reuse that selection.
Read the queue once:
```bash
uv run reforger_queue_read.py --portal-window
```
Force the window picker again if the saved selection is stale or wrong:
```bash
uv run reforger_queue_read.py --portal-window --portal-reselect
```
Run once with crop and OCR details while tuning or checking accuracy:
```bash ```bash
uv run reforger_queue_read.py --portal-window --show-crop --debug uv run reforger_queue_read.py --portal-window --show-crop --debug
``` ```
The saved portal restore token lives here: On first run, your desktop should show a window-sharing picker. Select the Arma Reforger window. If the portal grants persistence, the tool stores a restore token here:
```text ```text
~/.local/state/anti-prestige-tool/portal-window-restore-token ~/.local/state/anti-prestige-tool/portal-window-restore-token
``` ```
## Watch Behavior Later runs try to reuse that token:
Default watch mode:
```text
poll interval: 15 seconds
alert threshold: 5, inclusive
alerts: once per newly detected position
notification urgency: critical
```
Each poll prints a timestamp, the detected queue number, and whether an alert fired:
```text
2026-05-01T14:44:05+01:00 number=5 alert=yes
2026-05-01T14:44:20+01:00 number=5 alert=no
```
If no digits are found, the watcher logs that and keeps running.
## Optional Flags
Common flags:
```text
--watch keep polling instead of reading once
--watch-interval SECONDS poll delay; default 15
--alert-threshold N alert when position is <= N; default 5
--portal-reselect ignore the saved portal selection and pick again
--show-crop print the resolved crop
--debug print OCR match details
--save-input PATH save the captured portal frame
--notify-command COMMAND notification command; default notify-send
```
Input and test flags:
```text
--image PATH read an existing screenshot
--dataset DIR read every image in a dataset directory
--expect-filenames compare dataset results to numeric filenames
--font FONT debug with synthetic font templates
--crop X,Y,W,H override the crop
--cropped treat input image as already cropped
```
## Examples
Alert at queue position `10` or lower:
```bash ```bash
uv run reforger_queue_read.py --watch --portal-window --alert-threshold 10 uv run reforger_queue_read.py --portal-window --show-crop --debug
``` ```
Poll every 5 seconds: Force a fresh picker if the saved selection is wrong or stale:
```bash ```bash
uv run reforger_queue_read.py --watch --portal-window --watch-interval 5 uv run reforger_queue_read.py --portal-window --portal-reselect --show-crop --debug
``` ```
Save the captured input image while debugging: Save the captured input image while debugging:
@ -130,18 +62,6 @@ Save the captured input image while debugging:
uv run reforger_queue_read.py --portal-window --save-input /tmp/apt-portal-window.png --show-crop --debug uv run reforger_queue_read.py --portal-window --save-input /tmp/apt-portal-window.png --show-crop --debug
``` ```
Test watcher alerting against a fixture without sending a real notification:
```bash
uv run reforger_queue_read.py --watch --image datasets/regression-test-set/3.png --watch-interval 1 --notify-command true
```
Read an existing screenshot:
```bash
uv run reforger_queue_read.py --image /path/to/screenshot.png --show-crop --debug
```
## Validation ## Validation
Run the regression set: Run the regression set:
@ -173,6 +93,12 @@ datasets/scootz-dataset/2160x1440.png 24
## Debug Options ## Debug Options
Read an existing screenshot:
```bash
uv run reforger_queue_read.py --image /path/to/screenshot.png --show-crop --debug
```
The default matcher uses bundled templates from: The default matcher uses bundled templates from:
```text ```text

View file

@ -1,6 +1,6 @@
[project] [project]
name = "anti-prestige-tool" name = "anti-prestige-tool"
version = "0.9.0" version = "0.7.5"
description = "Arma Reforger queue-position reader for Linux Wayland" description = "Arma Reforger queue-position reader for Linux Wayland"
readme = "README.md" readme = "README.md"
requires-python = ">=3.10" requires-python = ">=3.10"

View file

@ -1,6 +1,6 @@
import argparse import argparse
import os import os
import sys import tempfile
from .config import ( from .config import (
DEFAULT_POINTSIZE, DEFAULT_POINTSIZE,
@ -9,15 +9,14 @@ from .config import (
DEFAULT_REFERENCE_SIZE, DEFAULT_REFERENCE_SIZE,
DEFAULT_TEMPLATE_SET, DEFAULT_TEMPLATE_SET,
) )
from .magick import run_bytes
from .ocr import ( from .ocr import (
read_queue_number, read_queue_number,
resolve_crop, resolve_crop,
templates_from_dataset, templates_from_dataset,
write_template_set, write_template_set,
) )
from .portal import PortalCaptureError from .portal import capture_portal_window
from .reader import read_image_once, read_portal_window_once
from .watcher import run_watch
def parse_crop(value): def parse_crop(value):
@ -105,39 +104,52 @@ def build_template_set(args):
def read_single_image(args): def read_single_image(args):
result = read_image_once(args, args.image) crop = resolve_crop(args.image, args.crop, args.reference_crop, args.reference_size, args.scale_mode, args.cropped)
if args.show_crop and result.crop: if args.show_crop and crop:
print(f"crop={result.crop[0]},{result.crop[1]},{result.crop[2]},{result.crop[3]}") print(f"crop={crop[0]},{crop[1]},{crop[2]},{crop[3]}")
if not result.number: number, details = read_queue_number(args.image, crop, args.template_set, args.font, args.pointsize, args.cropped)
if not number:
print("no digits found") print("no digits found")
return 2 return 2
print(result.number) print(number)
if args.debug: if args.debug:
for digit, bbox, area, hamming, iou in result.details: for digit, bbox, area, hamming, iou in details:
print(f"digit={digit} bbox={bbox} area={area} hamming={hamming:.3f} iou={iou:.3f}") print(f"digit={digit} bbox={bbox} area={area} hamming={hamming:.3f} iou={iou:.3f}")
return 0 return 0
def read_portal_window(args): def read_portal_window(args):
fd, image_path = tempfile.mkstemp(prefix="reforger-queue-", suffix=".png")
os.close(fd)
try: try:
result = read_portal_window_once(args) portal_info = capture_portal_window(
except PortalCaptureError as exc: image_path,
print(str(exc), file=sys.stderr) args.portal_restore_token,
return 1 args.portal_reselect,
args.portal_timeout,
)
if args.save_input:
run_bytes(["magick", image_path, "+repage", args.save_input])
crop = resolve_crop(image_path, args.crop, args.reference_crop, args.reference_size, args.scale_mode, args.cropped)
if args.show_crop and crop:
print(f"crop={crop[0]},{crop[1]},{crop[2]},{crop[3]}")
number, details = read_queue_number(image_path, crop, args.template_set, args.font, args.pointsize, args.cropped)
finally:
try:
os.unlink(image_path)
except OSError:
pass
if args.show_crop and result.crop: if not number:
print(f"crop={result.crop[0]},{result.crop[1]},{result.crop[2]},{result.crop[3]}")
if not result.number:
print("no digits found") print("no digits found")
return 2 return 2
print(result.number) print(number)
if args.debug: if args.debug:
portal_info = result.portal_info or {}
print(f"portal-window node_id={portal_info.get('node_id', '')} streams={len(portal_info.get('streams', []))}") print(f"portal-window node_id={portal_info.get('node_id', '')} streams={len(portal_info.get('streams', []))}")
if portal_info.get("restore_token") and args.portal_restore_token: if portal_info.get("restore_token") and args.portal_restore_token:
print(f"portal-restore-token {args.portal_restore_token}") print(f"portal-restore-token {args.portal_restore_token}")
for digit, bbox, area, hamming, iou in result.details: for digit, bbox, area, hamming, iou in details:
print(f"digit={digit} bbox={bbox} area={area} hamming={hamming:.3f} iou={iou:.3f}") print(f"digit={digit} bbox={bbox} area={area} hamming={hamming:.3f} iou={iou:.3f}")
return 0 return 0
@ -147,24 +159,6 @@ def main():
parser.add_argument("--image", help="Read from an existing image.") parser.add_argument("--image", help="Read from an existing image.")
parser.add_argument("--dataset", help="Read every image in a dataset directory.") parser.add_argument("--dataset", help="Read every image in a dataset directory.")
parser.add_argument("--portal-window", action="store_true", help="Capture a user-selected Wayland window.") parser.add_argument("--portal-window", action="store_true", help="Capture a user-selected Wayland window.")
parser.add_argument("--watch", action="store_true", help="Keep reading and notify when the queue position is low.")
parser.add_argument(
"--watch-interval",
type=float,
default=15.0,
help="Seconds between watch-mode polls.",
)
parser.add_argument(
"--alert-threshold",
type=int,
default=5,
help="In watch mode, notify for detected queue positions at or below this number.",
)
parser.add_argument(
"--notify-command",
default="notify-send",
help="Notification command used by watch mode.",
)
parser.add_argument( parser.add_argument(
"--portal-restore-token", "--portal-restore-token",
default=DEFAULT_PORTAL_RESTORE_TOKEN, default=DEFAULT_PORTAL_RESTORE_TOKEN,
@ -226,25 +220,15 @@ def main():
parser.error("--save-input can only be used with --portal-window") parser.error("--save-input can only be used with --portal-window")
if args.portal_timeout <= 0: if args.portal_timeout <= 0:
parser.error("--portal-timeout must be positive") parser.error("--portal-timeout must be positive")
if args.watch_interval <= 0:
parser.error("--watch-interval must be positive")
if args.alert_threshold <= 1:
parser.error("--alert-threshold must be greater than 1")
if args.watch and not args.notify_command.strip():
parser.error("--notify-command must not be empty")
selected_modes = [args.image, args.dataset, args.build_template_set, args.portal_window] selected_modes = [args.image, args.dataset, args.build_template_set, args.portal_window]
if sum(bool(value) for value in selected_modes) != 1: if sum(bool(value) for value in selected_modes) != 1:
parser.error("choose exactly one of --image, --dataset, --portal-window, or --build-template-set") parser.error("choose exactly one of --image, --dataset, --portal-window, or --build-template-set")
if args.watch and not (args.image or args.portal_window):
parser.error("--watch can only be used with --image or --portal-window")
if args.build_template_set: if args.build_template_set:
return build_template_set(args) return build_template_set(args)
if args.dataset: if args.dataset:
return run_dataset(args) return run_dataset(args)
if args.watch:
return run_watch(args)
if args.image: if args.image:
return read_single_image(args) return read_single_image(args)
return read_portal_window(args) return read_portal_window(args)

View file

@ -6,12 +6,6 @@ import sys
from .config import PROJECT_DIR from .config import PROJECT_DIR
class PortalCaptureError(RuntimeError):
def __init__(self, message, used_restore_token=False):
super().__init__(message)
self.used_restore_token = used_restore_token
def portal_helper_path(): def portal_helper_path():
return os.path.join(PROJECT_DIR, "portal_capture_frame") return os.path.join(PROJECT_DIR, "portal_capture_frame")
@ -65,14 +59,14 @@ def capture_portal_window(output, restore_token_path, reselect, timeout):
result = subprocess.run(command, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) result = subprocess.run(command, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
except subprocess.CalledProcessError as exc: except subprocess.CalledProcessError as exc:
message = exc.stderr.strip() if exc.stderr else str(exc) message = exc.stderr.strip() if exc.stderr else str(exc)
raise PortalCaptureError(message, used_restore_token=bool(restore_token)) from exc sys.exit(message)
capture_info = {} capture_info = {}
if result.stdout.strip(): if result.stdout.strip():
try: try:
capture_info = json.loads(result.stdout) capture_info = json.loads(result.stdout)
except json.JSONDecodeError as exc: except json.JSONDecodeError as exc:
raise PortalCaptureError(f"failed to parse portal capture metadata: {exc}: {result.stdout!r}") from exc sys.exit(f"failed to parse portal capture metadata: {exc}: {result.stdout!r}")
new_restore_token = str(capture_info.get("restore_token", "") or "") new_restore_token = str(capture_info.get("restore_token", "") or "")
if restore_token_path and new_restore_token: if restore_token_path and new_restore_token:

View file

@ -1,57 +0,0 @@
import os
import tempfile
from dataclasses import dataclass
from .magick import run_bytes
from .ocr import read_queue_number, resolve_crop
from .portal import capture_portal_window
@dataclass
class QueueReadResult:
number: str
details: list
crop: tuple | None = None
portal_info: dict | None = None
def read_image_once(args, image_path):
crop = resolve_crop(
image_path,
args.crop,
args.reference_crop,
args.reference_size,
args.scale_mode,
args.cropped,
)
number, details = read_queue_number(
image_path,
crop,
args.template_set,
args.font,
args.pointsize,
args.cropped,
)
return QueueReadResult(number=number, details=details, crop=crop)
def read_portal_window_once(args):
fd, image_path = tempfile.mkstemp(prefix="reforger-queue-", suffix=".png")
os.close(fd)
try:
portal_info = capture_portal_window(
image_path,
args.portal_restore_token,
args.portal_reselect,
args.portal_timeout,
)
if args.save_input:
run_bytes(["magick", image_path, "+repage", args.save_input])
result = read_image_once(args, image_path)
result.portal_info = portal_info
return result
finally:
try:
os.unlink(image_path)
except OSError:
pass

View file

@ -1,109 +0,0 @@
import shlex
import subprocess
import sys
import time
from datetime import datetime
from .portal import PortalCaptureError
from .reader import read_image_once, read_portal_window_once
APP_NAME = "Anti Prestige Tool"
NOTIFY_TITLE = "Arma Reforger Queue"
def notify_queue_position(notify_command, position):
command = [
*shlex.split(notify_command),
"-a",
APP_NAME,
"-u",
"critical",
NOTIFY_TITLE,
f"Queue position is {position}",
]
try:
result = subprocess.run(command, check=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
except FileNotFoundError:
return False, f"missing notification command: {command[0]}"
if result.returncode != 0:
message = result.stderr.strip() or result.stdout.strip() or f"exit status {result.returncode}"
return False, f"notification command failed: {message}"
return True, ""
def queue_position(number):
try:
position = int(number)
except ValueError:
return None
if position < 1:
return None
return position
def should_alert(number, threshold, notified_positions):
position = queue_position(number)
if position is None or position > threshold or position in notified_positions:
return None
return position
def timestamp():
return datetime.now().astimezone().isoformat(timespec="seconds")
def print_poll_status(number, alert_state, message=""):
if number:
line = f"{timestamp()} number={number} alert={alert_state}"
else:
line = f"{timestamp()} no digits found alert={alert_state}"
if message:
line = f"{line} {message}"
print(line, flush=True)
def read_once(args):
if args.image:
return read_image_once(args, args.image)
return read_portal_window_once(args)
def run_watch(args):
notified_positions = set()
try:
while True:
try:
result = read_once(args)
except PortalCaptureError as exc:
print(str(exc), file=sys.stderr)
if exc.used_restore_token:
print(
"Portal restore token appears stale; rerun with --portal-reselect.",
file=sys.stderr,
)
return 1
if args.portal_reselect:
args.portal_reselect = False
if not result.number:
print_poll_status("", "no")
else:
position = should_alert(result.number, args.alert_threshold, notified_positions)
if position is None:
print_poll_status(result.number, "no")
else:
fired, error = notify_queue_position(args.notify_command, position)
if fired:
notified_positions.add(position)
print_poll_status(result.number, "yes")
else:
print_poll_status(result.number, "failed", error)
time.sleep(args.watch_interval)
except KeyboardInterrupt:
print(f"{timestamp()} watch stopped", flush=True)
return 130

2
uv.lock generated
View file

@ -4,5 +4,5 @@ requires-python = ">=3.10"
[[package]] [[package]]
name = "anti-prestige-tool" name = "anti-prestige-tool"
version = "0.9.0" version = "0.7.5"
source = { virtual = "." } source = { virtual = "." }