publish eve_orehold_watcher.py
This commit is contained in:
parent
4056724ff6
commit
5a7a01083e
1 changed files with 77 additions and 5 deletions
|
|
@ -176,17 +176,68 @@ def grab_region(region):
|
||||||
return Image.frombytes("RGB", raw.size, raw.bgra, "raw", "BGRX")
|
return Image.frombytes("RGB", raw.size, raw.bgra, "raw", "BGRX")
|
||||||
|
|
||||||
|
|
||||||
|
def _detect_readout(cp, img, mon_left=0, mon_top=0, save=True):
|
||||||
|
"""Find the ore-hold 'cur / cap m3' readout inside a screenshot. Returns the
|
||||||
|
absolute-screen region [l,t,w,h] (and saves it) or None. Split out from the
|
||||||
|
screen-grab so it can be unit-tested against a rendered image."""
|
||||||
|
import pytesseract
|
||||||
|
data = pytesseract.image_to_data(img, output_type=pytesseract.Output.DICT)
|
||||||
|
lines = {}
|
||||||
|
for i in range(len(data["text"])):
|
||||||
|
if not data["text"][i].strip():
|
||||||
|
continue
|
||||||
|
key = (data["block_num"][i], data["par_num"][i], data["line_num"][i])
|
||||||
|
lines.setdefault(key, []).append(i)
|
||||||
|
best = None # (cap, [l,t,w,h])
|
||||||
|
for idxs in lines.values():
|
||||||
|
joined = " ".join(data["text"][i] for i in idxs)
|
||||||
|
parsed = parse_orehold_text(joined)
|
||||||
|
if not parsed:
|
||||||
|
continue
|
||||||
|
cur, cap = parsed
|
||||||
|
if not (500 <= cap <= 2_000_000): # plausible ore/gas hold capacity
|
||||||
|
continue
|
||||||
|
xs = [data["left"][i] for i in idxs]
|
||||||
|
ys = [data["top"][i] for i in idxs]
|
||||||
|
rights = [data["left"][i] + data["width"][i] for i in idxs]
|
||||||
|
bots = [data["top"][i] + data["height"][i] for i in idxs]
|
||||||
|
pad = 8
|
||||||
|
region = [mon_left + min(xs) - pad, mon_top + min(ys) - pad,
|
||||||
|
(max(rights) - min(xs)) + 2 * pad, (max(bots) - min(ys)) + 2 * pad]
|
||||||
|
if best is None or cap > best[0]: # ore hold cap is the big number
|
||||||
|
best = (cap, region)
|
||||||
|
if best:
|
||||||
|
l, t, w, h = best[1]
|
||||||
|
if save:
|
||||||
|
save_region(cp, l, t, w, h)
|
||||||
|
print(f"[ocr] auto-detected ore-hold readout -> region={l},{t},{w},{h} "
|
||||||
|
f"(cap~{best[0]:,})")
|
||||||
|
return best[1]
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def auto_region(cp):
|
||||||
|
"""Scan the whole screen for the ore-hold readout and save its location —
|
||||||
|
replaces the manual --snip. Just have the in-game Ore Hold window open."""
|
||||||
|
import mss
|
||||||
|
from PIL import Image
|
||||||
|
try:
|
||||||
|
with mss.mss() as sct:
|
||||||
|
mon = sct.monitors[0] # full virtual desktop (all screens)
|
||||||
|
raw = sct.grab(mon)
|
||||||
|
img = Image.frombytes("RGB", raw.size, raw.bgra, "raw", "BGRX")
|
||||||
|
except Exception as e:
|
||||||
|
print(f"[ocr] auto-detect error: {e}")
|
||||||
|
return None
|
||||||
|
return _detect_readout(cp, img, mon["left"], mon["top"])
|
||||||
|
|
||||||
|
|
||||||
def run_ocr(cp):
|
def run_ocr(cp):
|
||||||
import pytesseract
|
import pytesseract
|
||||||
tcmd = cp.get("ocr", "tesseract_cmd", fallback="").strip()
|
tcmd = cp.get("ocr", "tesseract_cmd", fallback="").strip()
|
||||||
if tcmd:
|
if tcmd:
|
||||||
pytesseract.pytesseract.tesseract_cmd = tcmd
|
pytesseract.pytesseract.tesseract_cmd = tcmd
|
||||||
|
|
||||||
region_s = cp.get("ocr", "region", fallback="").strip()
|
|
||||||
if not region_s:
|
|
||||||
sys.exit("No OCR region set. Run: python eve_orehold_watcher.py --snip")
|
|
||||||
region = [int(x) for x in region_s.split(",")]
|
|
||||||
|
|
||||||
poll = cp.getint("watcher", "poll_secs", fallback=10)
|
poll = cp.getint("watcher", "poll_secs", fallback=10)
|
||||||
alert_pct = cp.getfloat("watcher", "alert_pct", fallback=95.0)
|
alert_pct = cp.getfloat("watcher", "alert_pct", fallback=95.0)
|
||||||
reset_pct = cp.getfloat("watcher", "reset_pct", fallback=50.0)
|
reset_pct = cp.getfloat("watcher", "reset_pct", fallback=50.0)
|
||||||
|
|
@ -194,6 +245,18 @@ def run_ocr(cp):
|
||||||
# stall = hold not growing -> lasers/drones stopped (depleted rock, idle drones)
|
# stall = hold not growing -> lasers/drones stopped (depleted rock, idle drones)
|
||||||
stall_secs = cp.getint("watcher", "stall_secs", fallback=150)
|
stall_secs = cp.getint("watcher", "stall_secs", fallback=150)
|
||||||
|
|
||||||
|
region_s = cp.get("ocr", "region", fallback="").strip()
|
||||||
|
if region_s:
|
||||||
|
region = [int(x) for x in region_s.split(",")]
|
||||||
|
else:
|
||||||
|
print("[ocr] no saved region — auto-detecting the Ore Hold readout on screen...")
|
||||||
|
region = auto_region(cp)
|
||||||
|
while region is None:
|
||||||
|
print(f"[ocr] not visible yet — open your in-game Ore Hold window. "
|
||||||
|
f"Retrying in {max(poll, 15)}s...")
|
||||||
|
time.sleep(max(poll, 15))
|
||||||
|
region = auto_region(cp)
|
||||||
|
|
||||||
print(f"[ocr] watching region={region} every {poll}s; "
|
print(f"[ocr] watching region={region} every {poll}s; "
|
||||||
f"alert>={alert_pct}% reset<{reset_pct}% stall>{stall_secs}s")
|
f"alert>={alert_pct}% reset<{reset_pct}% stall>{stall_secs}s")
|
||||||
armed = True
|
armed = True
|
||||||
|
|
@ -201,12 +264,14 @@ def run_ocr(cp):
|
||||||
last_cur = -1
|
last_cur = -1
|
||||||
last_grow = time.time()
|
last_grow = time.time()
|
||||||
last_stall_alert = 0.0
|
last_stall_alert = 0.0
|
||||||
|
misses = 0
|
||||||
while True:
|
while True:
|
||||||
try:
|
try:
|
||||||
img = grab_region(region)
|
img = grab_region(region)
|
||||||
text = pytesseract.image_to_string(img, config="--psm 7")
|
text = pytesseract.image_to_string(img, config="--psm 7")
|
||||||
parsed = parse_orehold_text(text)
|
parsed = parse_orehold_text(text)
|
||||||
if parsed:
|
if parsed:
|
||||||
|
misses = 0
|
||||||
cur, cap = parsed
|
cur, cap = parsed
|
||||||
pct = 100.0 * cur / cap
|
pct = 100.0 * cur / cap
|
||||||
print(f"[ocr] {cur}/{cap} m3 ({pct:.1f}%) armed={armed}")
|
print(f"[ocr] {cur}/{cap} m3 ({pct:.1f}%) armed={armed}")
|
||||||
|
|
@ -232,7 +297,14 @@ def run_ocr(cp):
|
||||||
armed = False
|
armed = False
|
||||||
last_alert = time.time()
|
last_alert = time.time()
|
||||||
else:
|
else:
|
||||||
|
misses += 1
|
||||||
print(f"[ocr] no reading (text={text!r})")
|
print(f"[ocr] no reading (text={text!r})")
|
||||||
|
# window moved / closed? after ~2 min of misses, re-find it on screen
|
||||||
|
if misses % 12 == 0:
|
||||||
|
print("[ocr] readout lost — re-scanning screen for the Ore Hold...")
|
||||||
|
new = auto_region(cp)
|
||||||
|
if new:
|
||||||
|
region = new
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
print(f"[ocr] error: {e}")
|
print(f"[ocr] error: {e}")
|
||||||
time.sleep(poll)
|
time.sleep(poll)
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue