109 lines
3.1 KiB
Python
109 lines
3.1 KiB
Python
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
|