new: add detection for line clear animation

nestris_ocr/config_class.py

@@ -58,6 +58,8 @@
     "calibration.pct.das.current_piece_das": [0.216, 0.211, 0.061, 0.034],
     "calibration.pct.das.instant_das": [0.154, 0.134, 0.061, 0.034],
 
+    "calibration.capture_line_clear": False,
+
     "network.host": "127.0.0.1",
     "network.port": 3338,
     "network.protocol": "LEGACY",

nestris_ocr/ocr_algo/line_clear.py

@@ -0,0 +1,110 @@
+import numpy as np
+
+PIC_PATTERNS = [
+    "XXXXXXXXXX",
+    "XXXXOOXXXX",
+    "XXXOOOOXXX",
+    "XXOOOOOOXX",
+    "XOOOOOOOOX",
+    "OOOOOOOOOO",
+]
+
+
+def pattern_optimizer(x):
+    return x == "X"
+
+
+def gen_numpy_patterns(patterns):
+    optimizer_func = np.vectorize(pattern_optimizer)
+    patterns = [optimizer_func(list(p)) for p in patterns]
+    return np.stack(patterns)
+
+
+PATTERNS = gen_numpy_patterns(PIC_PATTERNS)
+NO_MATCH = len(PATTERNS)
+
+
+class LineClearDetection:
+    def __init__(self):
+        self.prev_line_states = [NO_MATCH for i in range(20)]
+        self.clearing_lines = [False for i in range(20)]
+        self.prev_block_count = 0
+        self.lines_cleared = []
+
+    def reset(self):
+        self.prev_line_states = [NO_MATCH for i in range(20)]
+        self.prev_block_count = 0
+
+    def process(self, field):
+        field = np.where(field != 0, True, False)
+        line_states = [self.match_state(line) for line in field]
+        block_count = np.count_nonzero(field)
+        block_diff = self.prev_block_count - block_count
+
+        if block_diff > 0 and block_diff % 2 == 0:
+            self.lines_cleared = []
+            for i in range(len(self.clearing_lines)):
+                is_clear = self.is_line_clear_anim(
+                    line_states[i], self.prev_line_states[i]
+                )
+                self.clearing_lines[i] = is_clear
+                if is_clear:
+                    self.lines_cleared.append(i)
+
+        # end of function, setup "prev"
+        self.prev_line_states = line_states
+        self.prev_block_count = block_count
+        anim_state = 6
+        if len(self.lines_cleared) > 0:
+            anim_state = self.prev_line_states[self.lines_cleared[0]]
+        return self.lines_cleared, anim_state
+
+    # returns true if the line is in a line clear animation
+    def is_line_clear_anim(self, current, prev):
+        if current == NO_MATCH:
+            return False
+        if current == NO_MATCH - 1 and prev in [NO_MATCH - 1, NO_MATCH - 2]:
+            return False  # animation is over
+        if prev == NO_MATCH:
+            if current in [0, 1]:
+                return True
+        elif prev == NO_MATCH - 2:
+            if current == NO_MATCH - 1:  # prev guaranteed to be 0-5 now...
+                return True
+        elif 1 <= current - prev <= 2:
+            return True
+        return False
+
+    def match_state(self, line):
+        for i, pattern in enumerate(PATTERNS):
+            if np.array_equal(line, pattern):
+                return i
+        return NO_MATCH
+
+
+def gen_test_clear(y_values):
+    result = []
+    for i in range(24):
+        field = ["OOOOOOOOOO" for j in range(20)]
+        for y in y_values:
+            field[y] = PIC_PATTERNS[i // 4]
+        result.append(gen_numpy_patterns(field))
+    return result
+
+
+def test_clear():
+    # single
+    single = gen_test_clear([17])
+    double = gen_test_clear([15, 16])
+    triple = gen_test_clear([16, 17, 18])
+    tetris = gen_test_clear([16, 17, 18, 19])
+    return [single, double, triple, tetris]
+
+
+if __name__ == "__main__":
+    print(PATTERNS)
+    lcd = LineClearDetection()
+    sequences = test_clear()
+    for seq in sequences:
+        for frame in seq:
+            print(lcd.process(frame))

nestris_ocr/ocr_state/field_state.py

@@ -9,10 +9,6 @@ class FieldState(object):  # noqa: E302
     def __init__(self, data):
         self.data = data
 
-    # returns block count for field below row 18
-    def blockCountAdjusted(self):
-        return 0
-
     def __eq__(self, other):
         if isinstance(other, self.__class__):
             result = np.array_equal(self.data, other.data)
@@ -27,9 +23,6 @@ def __eq__(self, other):
     def piece_spawned(self, other):
         return False
 
-    def line_clear_animation(self, other):
-        return False
-
     def serialize(self):
         result = self.data
         if config["network.protocol"] in [

nestris_ocr/scan_strat/base_strategy.py

@@ -37,7 +37,7 @@ def __init__(self):
         self.cur_piece_das = None
         self.instant_das = None
         self.current_time = 0
-
+        self.line_clear_anim = None
         self.colors = Colors()
 
         # initialize palette
@@ -68,6 +68,9 @@ def to_dict(self):
             result["cur_piece"] = self.cur_piece
             result["cur_piece_das"] = self.cur_piece_das
             result["instant_das"] = self.instant_das
+        if config["calibration.capture_line_clear"]:
+            result["line_clear_anim"] = self.line_clear_anim
+
         return result
 
     def update(self, timestamp, frame):

nestris_ocr/scan_strat/naive_strategy.py

@@ -1,6 +1,7 @@
 from nestris_ocr.scan_strat.base_strategy import BaseStrategy, GameState
 from nestris_ocr.ocr_state.field_state import FieldState
 from nestris_ocr.config import config
+from nestris_ocr.ocr_algo.line_clear import LineClearDetection
 
 from nestris_ocr.scan_strat.scan_helpers import (
     scan_black_n_white,
@@ -28,6 +29,11 @@ def __init__(self, *args):
         self.tasks = self.setup_tasks()
         self.interpolate = config["calibration.color_interpolation"]
 
+    def on_newgame(self):
+        super().on_newgame()
+        if config["calibration.capture_line_clear"]:
+            self.line_clear_detect.reset()
+
     # sets up the tasks that we will be doing naively.
     def setup_tasks(self):
         tasks = []
@@ -47,6 +53,8 @@ def setup_tasks(self):
                 tasks.append(self.lookup_colors)
 
             tasks.append(self.scan_field)
+            if config["calibration.capture_line_clear"]:
+                self.line_clear_detect = LineClearDetection()
 
         if config["calibration.capture_preview"]:
             tasks.append(self.scan_preview)
@@ -119,6 +127,8 @@ def scan_level(self, img):
     def scan_field(self, img):
         result = scan_field(img, self.colors)
         self.field = FieldState(result)
+        if config["calibration.capture_line_clear"]:
+            self.line_clear_anim = self.line_clear_detect.process(self.field.data)
 
     def scan_preview(self, img):
         self.preview = scan_preview(img, self.colors)