mower-ng/mower/solvers/operation.py

175 lines
6.5 KiB
Python

from datetime import datetime, timedelta
from functools import cached_property
from typing import Optional
import cv2
from mower.solvers.navigation.utils import generate_name
from mower.utils import config
from mower.utils import typealias as tp
from mower.utils.datetime import get_server_weekday
from mower.utils.graph import SceneGraphSolver
from mower.utils.image import cropimg, diff_ratio, loadimg
from mower.utils.log import logger
from mower.utils.path import get_path
from mower.utils.recognize import Scene
from mower.utils.vector import va
class OperationSolver(SceneGraphSolver):
def run(self, stop_time: datetime):
logger.info("Start: 代理作战")
self.stop_time = stop_time - timedelta(minutes=5)
self.sanity_drain = False
self.timeout = False
self.drop_list: list[dict[str, int]] = []
super().run()
def number(self, scope: tp.Scope, height: Optional[int] = None):
rect_limits = [{"w": 5, "h": 5, "char": ""}]
return config.recog.num.number_int(
"secret_front", scope, height, rect_limits=rect_limits
)
def drop_animation(self) -> bool:
drop_scope = (100, 775), (1920, 945)
img1 = cropimg(config.recog.gray, drop_scope)
self.sleep()
img2 = cropimg(config.recog.gray, drop_scope)
return diff_ratio(img1, img2, ratio=0.03)
@cached_property
def drop_data(self) -> dict[str, tp.GrayImage]:
data = {}
for i in get_path("@install/ui/public/depot").iterdir():
if "信物" in i.stem:
continue
supplement = "罗德岛物资补给"
if i.stem.startswith(supplement) and i.stem != supplement:
continue
data[i.stem] = loadimg.__wrapped__(i, True, "BLACK")
return data
@cached_property
def drop_digits(self) -> list[tp.GrayImage]:
return [generate_name(str(i), font_size=28, style="dark") for i in range(10)]
def drop_recog(self):
drop_result = {}
for name, img in self.drop_data.items():
score, scope = config.recog.matcher.match(img)
if score < 0.7:
continue
logger.debug(f"{name=} {score=} {scope=}")
target = config.recog.gray.copy()
target = cropimg(target, scope)
target = cropimg(target, ((0, 125), (180, 155)))
values = []
for _ in range(10):
max_score, max_pos, max_digit = 0, None, None
for n in range(10):
result = cv2.matchTemplate(
target, self.drop_digits[n], cv2.TM_CCOEFF_NORMED
)
min_val, max_val, min_loc, max_loc = cv2.minMaxLoc(result)
if max_val > max_score:
max_score, max_pos, max_digit = max_val, max_loc, n
if max_score < 0.75:
break
logger.debug(f"{max_score=} {max_pos=} {max_digit=}")
values.append((max_pos[0], max_digit))
h, w = self.drop_digits[max_digit].shape
cv2.rectangle(target, max_pos, va(max_pos, (w, h)), (0,), -1)
logger.debug(f"{name=} {values=}")
values = sorted(values)
value = 0
for v in values:
value = value * 10 + v[1]
if value != 0:
drop_result[name] = value
logger.debug(f"{drop_result=}")
self.drop_list.append(drop_result)
def transition(self):
if (scene := self.scene()) == Scene.OPERATOR_BEFORE:
if datetime.now() > self.stop_time:
self.timeout = True
return True
if config.recog.gray[65][1333] < 200:
self.sleep()
return
if config.recog.gray[907][1600] < 127:
self.tap((1776, 908))
return
repeat = self.number(((1520, 890), (1545, 930)), 28)
if repeat > 1:
self.tap((1500, 910))
self.tap((1500, 801))
return
self.tap_element("ope_start", interval=2)
elif scene == Scene.OPERATOR_SELECT:
self.tap((1655, 781))
elif scene == Scene.OPERATOR_FINISH:
if self.drop_animation():
return
self.drop_recog()
self.tap((310, 330))
elif scene == Scene.OPERATOR_FAILED:
self.tap((310, 330))
elif scene == Scene.OPERATOR_ONGOING:
if self.find("ope_agency_fail"):
self.tap((121, 79))
else:
self.sleep(10)
elif scene == Scene.OPERATOR_GIVEUP:
self.tap_element("fight/give_up")
elif scene == Scene.OPERATOR_RECOVER_POTION:
use_medicine = False
# 先看设置是否吃药
if config.conf.maa_expiring_medicine:
if config.conf.exipring_medicine_on_weekend:
use_medicine = get_server_weekday() >= 5
else:
use_medicine = True
# 再看是否有药可吃
if use_medicine:
img = cropimg(config.recog.img, ((1015, 515), (1170, 560)))
img = cv2.cvtColor(img, cv2.COLOR_RGB2HSV)
img = cv2.inRange(img, (170, 0, 0), (174, 255, 255))
count = cv2.countNonZero(img)
logger.debug(f"{count=}")
use_medicine = count > 3000
if use_medicine:
logger.info("使用即将过期的理智药")
self.tap((1635, 865))
return
else:
self.sanity_drain = True
return True
elif scene == Scene.OPERATOR_RECOVER_ORIGINITE:
self.sanity_drain = True
return True
elif scene == Scene.OPERATOR_ELIMINATE:
if self.find("ope_agency_lock"):
logger.error("无法代理当期剿灭")
return True
if self.find("1800"):
logger.info("本周剿灭已完成")
return True
if pos := self.find("ope_elimi_agency"):
self.tap(pos)
return
self.tap_element("ope_start", interval=2)
elif scene == Scene.OPERATOR_ELIMINATE_AGENCY:
self.tap_element("ope_elimi_agency_confirm", interval=2)
elif scene == Scene.UPGRADE:
self.tap((960, 540))
elif scene in self.waiting_scene:
self.waiting_solver()
else:
return True