feat: add mining collection automation

tests/test_mining.py

@@ -0,0 +1,74 @@
+import copy
+import unittest
+from pathlib import Path
+
+from PIL import Image
+
+from automation.engine import Matcher, Region, load_flow
+from automation.mining import MiningRunner
+
+ROOT = Path(__file__).resolve().parent.parent
+
+
+class MiningDesktop:
+    def __init__(self, images):
+        self.images = images
+        self.index = 0
+        self.clicks = []
+
+    def screenshot(self):
+        return self.images[min(self.index, len(self.images) - 1)]
+
+    def click(self, point, region, size):
+        self.clicks.append(point)
+        self.index += 1
+
+
+class MiningTests(unittest.TestCase):
+    def setUp(self):
+        self.flow = load_flow(ROOT / 'workflows/mining_claim.json')
+        self.flow['wait_after_collect_seconds'] = 0
+        self.matcher = Matcher(ROOT / 'assets')
+        self.region = Region(0, 0, 562, 1002)
+
+    def image(self, state):
+        image = Image.new('RGB', (562, 1002), (42, 58, 73))
+        for spec in self.flow['states'][state]:
+            template = Image.open(ROOT / 'assets' / spec['template']).convert('RGB')
+            x, y, _, _ = spec['search']
+            image.paste(template, (x + 5, y + 5))
+        return image
+
+    def runner(self, states):
+        desktop = MiningDesktop([self.image(state) for state in states])
+        return desktop, MiningRunner(desktop, self.matcher, copy.deepcopy(self.flow),
+                                     lambda _: self.region, log=lambda _: None, timeout=0.8)
+
+    def test_white_compass_is_selected_first(self):
+        desktop, runner = self.runner([
+            'home', 'territory', 'collect', 'reward', 'rat_house',
+            'white_compass', 'hire_dialog', 'rat_house', 'territory', 'home',
+        ])
+        self.assertEqual(runner.run(), 'completed')
+        expected = ('territory', 'rat_house', 'collect', 'dismiss_reward', 'hire_open',
+                    'select_white', 'hire_confirm', 'rat_back', 'territory_back')
+        self.assertEqual(desktop.clicks, [self.flow['clicks'][name] for name in expected])
+
+    def test_no_collect_skips_hiring_and_returns(self):
+        desktop, runner = self.runner(['home', 'territory', 'rat_house', 'territory', 'home'])
+        self.assertEqual(runner.run(), 'no_collection')
+        expected = ('territory', 'rat_house', 'rat_back', 'territory_back')
+        self.assertEqual(desktop.clicks, [self.flow['clicks'][name] for name in expected])
+
+    def test_no_compass_closes_dialog(self):
+        desktop, runner = self.runner([
+            'home', 'territory', 'collect', 'reward', 'rat_house',
+            'hire_dialog', 'rat_house', 'territory', 'home',
+        ])
+        self.assertEqual(runner.run(), 'completed')
+        self.assertIn(self.flow['clicks']['close_dialog'], desktop.clicks)
+        self.assertNotIn(self.flow['clicks']['hire_confirm'], desktop.clicks)
+
+
+if __name__ == '__main__':
+    unittest.main()