feat: add sweep and task reward automation

tests/test_sweep.py

@@ -0,0 +1,70 @@
+import copy
+import unittest
+from pathlib import Path
+
+from PIL import Image
+
+from automation.engine import Matcher, Region, WorkflowError, load_flow
+from automation.sweep import SweepRunner
+
+ROOT = Path(__file__).resolve().parent.parent
+
+
+class SweepDesktop:
+    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, hold=0.2):
+        self.clicks.append((point, hold))
+        self.index += 1
+
+
+class SweepTests(unittest.TestCase):
+    def setUp(self):
+        self.flow = load_flow(ROOT / 'workflows/sweep_claim.json')
+        self.flow['interval_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 = SweepDesktop([self.image(state) for state in states])
+        flow = copy.deepcopy(self.flow)
+        return desktop, SweepRunner(desktop, self.matcher, flow,
+                                    lambda _: self.region, log=lambda _: None, timeout=0.8)
+
+    def test_five_sweeps_dismisses_each_result_and_returns_home(self):
+        states = ['home', 'sweep_available']
+        for _ in range(5):
+            states.extend(('reward', 'sweep_available'))
+        states.append('home')
+        desktop, runner = self.runner(states)
+        self.assertEqual(runner.run(5), 'completed')
+        expected = [(self.flow['clicks']['open_pig'], 0.2)]
+        for _ in range(5):
+            expected.extend(((self.flow['clicks']['sweep'], 0.8),
+                             (self.flow['clicks']['dismiss_reward'], 0.2)))
+        expected.append((self.flow['clicks']['close'], 0.2))
+        self.assertEqual(desktop.clicks, expected)
+
+    def test_invalid_count_stops_before_clicking(self):
+        desktop, runner = self.runner(['home'])
+        with self.assertRaises(WorkflowError):
+            runner.run(0)
+        self.assertEqual(desktop.clicks, [])
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_tasks.py

@@ -0,0 +1,86 @@
+import unittest
+from pathlib import Path
+
+from PIL import Image
+
+from automation.engine import Matcher, Region, load_flow
+from automation.tasks import TaskRunner
+
+ROOT = Path(__file__).resolve().parent.parent
+
+
+class TaskDesktop:
+    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, hold=0.2):
+        self.clicks.append((point, hold))
+        self.index += 1
+
+
+class TaskTests(unittest.TestCase):
+    def setUp(self):
+        self.flow = load_flow(ROOT / 'workflows/task_claim.json')
+        self.matcher = Matcher(ROOT / 'assets')
+        self.region = Region(0, 0, 562, 1002)
+
+    def image(self, state, claim=False, dot=None):
+        image = Image.new('RGB', (562, 1002), (42, 58, 73))
+        if state in self.flow['states']:
+            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))
+        if claim:
+            spec = self.flow['claim_button']
+            template = Image.open(ROOT / 'assets' / spec['template']).convert('RGB')
+            x, y, _, _ = spec['search']
+            image.paste(template, (x + 5, y + 5))
+        if dot == 'home':
+            spec = self.flow['task_dot'][0]
+        elif dot:
+            spec = self.flow['tab_dots'][dot]
+        else:
+            spec = None
+        if spec:
+            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, images):
+        desktop = TaskDesktop(images)
+        return desktop, TaskRunner(desktop, self.matcher, self.flow,
+                                   lambda _: self.region, log=lambda _: None, timeout=0.8)
+
+    def test_claims_current_and_dotted_other_panel(self):
+        images = [
+            self.image('home', dot='home'),
+            self.image('daily', claim=True),
+            self.image('reward'),
+            self.image('daily', dot='weekly'),
+            self.image('weekly', claim=True),
+            self.image('reward'),
+            self.image('weekly'),
+            self.image('home'),
+        ]
+        desktop, runner = self.runner(images)
+        self.assertEqual(runner.run(), 'claimed')
+        self.assertEqual(len(desktop.clicks), 7)
+        self.assertEqual(desktop.clicks[0][0], self.flow['clicks']['open_tasks'])
+        self.assertEqual(desktop.clicks[3][0], self.flow['clicks']['open_weekly'])
+        self.assertEqual(desktop.clicks[-1][0], self.flow['clicks']['close'])
+
+    def test_without_home_dot_does_not_open_tasks(self):
+        desktop, runner = self.runner([self.image('home')])
+        self.assertEqual(runner.run(), 'no_task_dot')
+        self.assertEqual(desktop.clicks, [])
+
+
+if __name__ == '__main__':
+    unittest.main()