added async stone detection thread
This commit is contained in:
61
litris.py
61
litris.py
@@ -5,6 +5,7 @@ from pynput.keyboard import Key, Controller
|
||||
from field import Field
|
||||
from tetromino import Tetromino
|
||||
from optimizer import Optimizer
|
||||
from litris_stone_id_thread import NewStoneID
|
||||
|
||||
O_FULL = [[0, 0, 0, 0], [0, 1, 1, 0], [0, 1, 1, 0], [0, 0, 0, 0]]
|
||||
D_FULL = [[0, 0, 0, 0], [0, 1, 1, 0], [0, 0, 1, 0], [0, 0, 0, 0]]
|
||||
@@ -42,6 +43,8 @@ class Litris(GameBase):
|
||||
|
||||
self.needles = {1: cv.imread("litris/blue_needle.jpg", cv.IMREAD_UNCHANGED)}
|
||||
|
||||
self.stone_id_thread = NewStoneID()
|
||||
|
||||
|
||||
def reset_field(self):
|
||||
self.state = [[' ' for cols in range(Field.WIDTH)]
|
||||
@@ -62,26 +65,42 @@ class Litris(GameBase):
|
||||
self.stone_coordinates[e][i] = [(i * dim) + (i * i_spacing), (e * dim) + (e * e_spacing), dim, dim]
|
||||
|
||||
def assess_playfield_and_make_move(self):
|
||||
current_letter = self.new_stone_detection_and_identification()
|
||||
#current_letter = self.stone_id()
|
||||
#current_letter = 'D'
|
||||
print("current_letter: ", current_letter)
|
||||
if current_letter is None:
|
||||
cv.waitKey(50)
|
||||
return
|
||||
|
||||
current_tetromino = Tetromino.create(current_letter)
|
||||
opt = Optimizer.get_optimal_drop(self.field, current_tetromino)
|
||||
rotation = opt['tetromino_rotation']
|
||||
column = opt['tetromino_column']
|
||||
print("Rota:", rotation)
|
||||
print("column:", column)
|
||||
current_tetromino.rotate(rotation)
|
||||
offset_col = current_tetromino.get_offset_column(rotation)
|
||||
print("offset column:", offset_col)
|
||||
self.field.drop(current_tetromino, column)
|
||||
self.move_stone(column - offset_col, rotation)
|
||||
print(self.field)
|
||||
|
||||
while True:
|
||||
if self.stone_id_thread.get_pick_up_status() == False:
|
||||
if self.overlay.run_mode == 'stopped' or self.overlay.run_mode == 'paused':
|
||||
return
|
||||
cv.waitKey(50)
|
||||
continue
|
||||
|
||||
#current_letter = self.new_stone_detection_and_identification()
|
||||
#current_letter = self.stone_id()
|
||||
#current_letter = 'D'
|
||||
current_letter = self.stone_id_thread.get_actual_letter()
|
||||
#self.stone_id_thread.set_pick_up_status(False)
|
||||
print("current_letter: ", current_letter)
|
||||
#if current_letter is None:
|
||||
# cv.waitKey(50)
|
||||
# return
|
||||
|
||||
current_tetromino = Tetromino.create(current_letter)
|
||||
opt = Optimizer.get_optimal_drop(self.field, current_tetromino)
|
||||
rotation = opt['tetromino_rotation']
|
||||
column = opt['tetromino_column']
|
||||
print("Rota:", rotation)
|
||||
print("column:", column)
|
||||
current_tetromino.rotate(rotation)
|
||||
offset_col = current_tetromino.get_offset_column(rotation)
|
||||
print("offset column:", offset_col)
|
||||
self.field.drop(current_tetromino, column)
|
||||
#self.move_stone(column - offset_col, rotation)
|
||||
for i in range(1,5,1):
|
||||
self.keyboard.press(Key.down)
|
||||
self.keyboard.release(Key.down)
|
||||
print("direction pressed: drop down")
|
||||
cv.waitKey(50)
|
||||
self.stone_id_thread.set_pick_up_status(False)
|
||||
print(self.field)
|
||||
|
||||
def get_current_board_state(self):
|
||||
# get an updated image of the game
|
||||
@@ -151,7 +170,7 @@ class Litris(GameBase):
|
||||
|
||||
|
||||
def new_stone_detection_and_identification(self):
|
||||
stone_coords = np.zeros((2, 4), dtype=object)
|
||||
stone_coords = np.zeros((4, 4), dtype=object)
|
||||
fail_counter = 0
|
||||
while True:
|
||||
screenshot = self.capture_window.get_screenshot()
|
||||
|
||||
145
litris_stone_id_thread.py
Normal file
145
litris_stone_id_thread.py
Normal file
@@ -0,0 +1,145 @@
|
||||
import threading
|
||||
import cv2 as cv
|
||||
import numpy as np
|
||||
from window_capture import WindowCapture
|
||||
from vision import Vision
|
||||
from config_file import UserConfigs
|
||||
|
||||
O_FULL = [[0, 0, 0, 0], [0, 1, 1, 0], [0, 1, 1, 0], [0, 0, 0, 0]]
|
||||
D_FULL = [[0, 0, 0, 0], [0, 1, 1, 0], [0, 0, 1, 0], [0, 0, 0, 0]]
|
||||
L_FULL = [[0, 0, 0, 0], [0, 0, 1, 0], [1, 1, 1, 0], [0, 0, 0, 0]]
|
||||
J_FULL = [[0, 0, 0, 0], [1, 1, 1, 0], [0, 0, 1, 0], [0, 0, 0, 0]]
|
||||
I_FULL = [[0, 0, 0, 0], [1, 1, 1, 1], [0, 0, 0, 0], [0, 0, 0, 0]]
|
||||
C_FULL = [[0, 0, 0, 0], [0, 0, 1, 0], [0, 0, 0, 0], [0, 0, 0, 0]]
|
||||
B_FULL = [[0, 0, 0, 0], [0, 1, 1, 0], [0, 0, 0, 0], [0, 0, 0, 0]]
|
||||
A_FULL = [[0, 0, 0, 0], [0, 1, 1, 1], [0, 0, 0, 0], [0, 0, 0, 0]]
|
||||
S_FULL = [[0, 0, 0, 0], [0, 0, 1, 1], [0, 1, 1, 0], [0, 0, 0, 0]]
|
||||
Z_FULL = [[0, 0, 0, 0], [0, 1, 1, 0], [0, 0, 1, 1], [0, 0, 0, 0]]
|
||||
T_FULL = [[0, 0, 0, 0], [0, 1, 1, 1], [0, 0, 1, 0], [0, 0, 0, 0]]
|
||||
class NewStoneID(threading.Thread):
|
||||
|
||||
def __init__(self):
|
||||
threading.Thread.__init__(self)
|
||||
|
||||
self.config = UserConfigs()
|
||||
self.capture_window = WindowCapture(None, None, self.config)
|
||||
self.vision_stun = Vision()
|
||||
|
||||
self.stone_coordinates = np.zeros((4, 4), dtype=object)
|
||||
self.fill_data_coordinates()
|
||||
|
||||
self.needles = {1: cv.imread("litris/blue_needle.jpg", cv.IMREAD_UNCHANGED)}
|
||||
|
||||
self.run_mode = 'init'
|
||||
self.actual_letter = ""
|
||||
self.to_pick_up = False
|
||||
|
||||
self.start()
|
||||
|
||||
def run(self):
|
||||
while True:
|
||||
current_stone = self.new_stone_detection()
|
||||
if current_stone is None:
|
||||
cv.waitKey(50)
|
||||
continue
|
||||
else:
|
||||
current_letter = self.get_letter_for_stone(current_stone)
|
||||
if current_letter is None:
|
||||
continue
|
||||
else:
|
||||
self.actual_letter = current_letter
|
||||
self.to_pick_up = True
|
||||
while self.to_pick_up:
|
||||
cv.waitKey(10)
|
||||
self.actual_letter = ""
|
||||
|
||||
def get_actual_letter(self):
|
||||
return self.actual_letter
|
||||
def set_pick_up_status(self, status):
|
||||
self.to_pick_up = status
|
||||
def get_pick_up_status(self):
|
||||
return self.to_pick_up
|
||||
|
||||
def callback(self):
|
||||
pass
|
||||
|
||||
def destroy(self):
|
||||
self.destroy()
|
||||
|
||||
def get_run_mode(self):
|
||||
return self.run_mode
|
||||
|
||||
|
||||
def new_stone_detection(self):
|
||||
|
||||
screenshot = self.capture_window.get_screenshot()
|
||||
screenshot = screenshot[580:845, 1148:1412]
|
||||
|
||||
rectangles = self.vision_stun.find(screenshot, self.needles[1], 0.85, 16)
|
||||
if len(rectangles) == 0:
|
||||
return None
|
||||
|
||||
points = self.vision_stun.get_click_points(rectangles)
|
||||
|
||||
stone_coords = np.zeros((4, 4), dtype=object)
|
||||
for point in points:
|
||||
x, y = self.point_in_small_rect(point)
|
||||
if x is not None and y is not None:
|
||||
stone_coords[x][y] = 1
|
||||
# self.change_value(x, y, int(needle_key))
|
||||
# print(field.data_value_grid)
|
||||
#cv.circle(screenshot, points[0], 7, (0, 255, 0), -1)
|
||||
#output_image = self.vision_stun.draw_rectangles(screenshot, rectangles)
|
||||
#cv.imshow("output_image", output_image)
|
||||
#cv.waitKey(150)
|
||||
|
||||
return stone_coords
|
||||
|
||||
def get_letter_for_stone(self, stone):
|
||||
|
||||
if np.array_equal(stone, O_FULL):
|
||||
return "O"
|
||||
elif np.array_equal(stone, D_FULL):
|
||||
return "D"
|
||||
elif np.array_equal(stone, L_FULL):
|
||||
return "L"
|
||||
elif np.array_equal(stone, J_FULL):
|
||||
return "J"
|
||||
elif np.array_equal(stone, I_FULL):
|
||||
return "I"
|
||||
elif np.array_equal(stone, C_FULL):
|
||||
return "C"
|
||||
elif np.array_equal(stone, B_FULL):
|
||||
return "B"
|
||||
elif np.array_equal(stone, A_FULL):
|
||||
return "A"
|
||||
elif np.array_equal(stone, S_FULL):
|
||||
return "S"
|
||||
elif np.array_equal(stone, Z_FULL):
|
||||
return "Z"
|
||||
elif np.array_equal(stone, T_FULL):
|
||||
return "T"
|
||||
else:
|
||||
return None
|
||||
|
||||
def point_in_small_rect(self, point):
|
||||
for e in range(0, 4, 1):
|
||||
for i in range(0, 4, 1):
|
||||
x1, y1, w, h = self.stone_coordinates[e][i]
|
||||
x2, y2 = x1 + w, y1 + h
|
||||
x, y = point
|
||||
if x1 < x and x < x2:
|
||||
if y1 < y and y < y2:
|
||||
return e, i
|
||||
return None, None
|
||||
|
||||
def fill_data_coordinates(self):
|
||||
# 610 to 1950 = 1340 - 76 / 20 = 63
|
||||
# 40 to 1380 = 1340 - 76 / 20 = 63
|
||||
# spacing 19 * 4
|
||||
dim = 63
|
||||
e_spacing = 4
|
||||
i_spacing = 4
|
||||
for e in range(0, 4, 1):
|
||||
for i in range(0, 4, 1):
|
||||
self.stone_coordinates[e][i] = [(i * dim) + (i * i_spacing), (e * dim) + (e * e_spacing), dim, dim]
|
||||
Reference in New Issue
Block a user