Дурак Копайлот – Повышаем винрейт с помощью компьютерного зрения

import time
import cv2
import numpy as np
import state
from PIL import ImageGrab

from templates import (
    START,
    END,
    TRUMPS,
    FINAL_STEPS,
    STEPS,
    YOUR_TURN,
)

from helpers import log, update_cards_its_my_now, reset_game

from steps import (
    detect_draw,
    detect_trump,
    check_final_step,
    check_game_end,
    check_steps
)

# Статусы которые управляют состоянием игры, либо ждем старта, либо концовки
WAIT_START = 0
WAIT_END = 1

the_game_state = WAIT_START # При запуске скрипта начинаем ждать начало игры

final_step_done = False # Флаг для обнаружения финального хода
steps_locked_until = 0 # Полезная что бы блокировать паттерны конца хода (такие как Done) иначе может поймать пару раз на нескольких скринах

previous_screen = None # Нужна для анализа

def get_previous_screen(): # Можно передавать аргументами, но можно и так сделать что бы со степс дергать вроде не очень красиво, ну можно это в стейт запихнуть
    return previous_screen

# Бесконечный цикл - Основная функция
while True:

    # Получаем скриншот экрана
    screen = np.array(
        ImageGrab.grab(all_screens=True)
    )

    # Меняем формат на cv2 асептебел
    screen = cv2.cvtColor(
        screen,
        cv2.COLOR_RGB2BGR
    )


    # Игра еще не начата? Тогда идем сюда и проверяем на паттерн начала игры
    if the_game_state == WAIT_START:
        t_screen = screen[240:291, 550:599] # делаем вырезку со скрина что бы оно не перебило другие переменные а то может
        result = cv2.matchTemplate(
            t_screen,
            START,
            cv2.TM_CCOEFF_NORMED
        )

        _, score, _, _ = cv2.minMaxLoc(result)

        if score > 0.9: # Если нашли меняем статус
            log('Игра началась')
            the_game_state = WAIT_END

    # Сюда идем если игра уже идет
    else:
        if state.game['trump'] is None: # Если козырь не определен идем сюда
            trump = detect_trump(screen) # Определили

            if trump: # Попали сюда если опеределили, если нет то попробуем снова на следующем скриншоте (возможно не закочилась раздача), хотя почти всегда будет 1 попытка
                state.game['trump'] = trump # Изменили состояние козыря

                log(f'Козырь определен: {trump}') # Прологировали

                time.sleep(1) # Техническая задержка, на скриншоте с козырем может еще не закончится раздача 

                t_screen = np.array( # Сделали скриншот
                    ImageGrab.grab(all_screens=True)
                )

                t_screen = cv2.cvtColor( # Форматнули
                    t_screen,
                    cv2.COLOR_RGB2BGR
                )

                found_cards = detect_draw(t_screen) # Определили начальные карты

                log(f'Найдены карты: {found_cards}') # Прологировали

                update_cards_its_my_now(found_cards) # Обновили стейт

        if time.perf_counter() >= steps_locked_until: # Проверка нету ли блокировки на проверку стола
            if check_steps(screen, get_previous_screen()): # Проверяем скриншот на Done etc. паттерны
                steps_locked_until = time.perf_counter() + 3 # Если получили обратно True делаем блокировку на 3 секунды проверки стола

        if not final_step_done: # Проверка а не финальный ли это ход
            if check_final_step(screen): # Если так меняем статус переменной
                final_step_done = True

        if check_game_end(screen): # Проверка а не закончилась ли игра вообще
            log('Игра закончилась')

            reset_game()
            final_step_done = False

            log('Состояние игры сброшено')

            the_game_state = WAIT_START

    previous_screen = screen.copy() # Делаем копию для анализа стола (на определенных моментах нужно)
    time.sleep(1) # Делаем скриншоты раз в 1 секунду

import cv2

TRUMPS = {
    'C': cv2.imread('templates/trumps/clubs.png'),
    'D': cv2.imread('templates/trumps/diamonds.png'),
    'H': cv2.imread('templates/trumps/hearts.png'),
    'S': cv2.imread('templates/trumps/spades.png'),
}

FINAL_STEPS = {
    'C': cv2.imread('templates/final_step/clubs.png'),
    'D': cv2.imread('templates/final_step/diamonds.png'),
    'H': cv2.imread('templates/final_step/hearts.png'),
    'S': cv2.imread('templates/final_step/spades.png'),
}

STEPS = {
    'done': cv2.imread('templates/steps/done.png'),
    'pass_up': cv2.imread('templates/steps/pass_up.png'),
    'pass_down': cv2.imread('templates/steps/pass_down.png'),
    'i_take_up': cv2.imread('templates/steps/i_take_up.png'),
    'i_take_down': cv2.imread('templates/steps/i_take_down.png'),
}


START = cv2.imread('templates/start.png')
END = cv2.imread('templates/end.png')
YOUR_TURN = cv2.imread('templates/steps/your_turn.png')

import time
import state
import os
from datetime import datetime
from PIL import Image
import cv2

import requests

# Логирует с отметкой по времени
def log(text):
    with open('log.txt', 'a', encoding='utf-8') as f:
        f.write(
            f'{time.strftime("%Y-%m-%d %H:%M:%S")} {text}\n'
        )

# Обновляет найденные карты на статус Me
def update_cards_its_my_now(found_cards):
    for card in found_cards:
        old_status = state.cards[card]

        if old_status != state.ME:
            state.cards[card] = state.ME

            log(
                f'Карта изменена: {card}'
                f'{old_status} -> {state.ME}'
            )

# Обновляет найденные карты на статус Enemy
def update_cards_its_enemy_now(found_cards):
    for card in found_cards:
        old_status = state.cards[card]

        if old_status != state.ENEMY:
            state.cards[card] = state.ENEMY

            log(
                f'Карта изменена: {card}'
                f'{old_status} -> {state.ENEMY}'
            )

# Обновляет найденные карты на статус DISCARD
def update_cards_its_discard_now(found_cards):
    for card in found_cards:
        old_status = state.cards[card]

        if old_status != state.DISCARD:
            state.cards[card] = state.DISCARD

            log(
                f'Карта изменена: {card}'
                f'{old_status} -> {state.DISCARD}'
            )


# Сбрасывает стейт
def reset_game():
    state.game['trump'] = None
    state.game['attacker'] = None
    state.game['defender'] = None
    state.game['turn'] = 0

    for card in state.cards:
        state.cards[card] = state.UNKNOWN


def save_previous_screen(screen):
    if screen is not None:
        os.makedirs("screenshots", exist_ok=True)
        filename = f"screenshots/{datetime.now().strftime('%Y%m%d_%H%M%S')}.png"

        img = Image.fromarray(cv2.cvtColor(screen, cv2.COLOR_BGR2RGB))

        img.save(filename, "PNG")


def send_final_cards_to_telegram(final_cards):
    text = f"Финальный шаг\nКарты: {', '.join(final_cards)}"

    url = f"https://api.telegram.org/bot/sendMessage"

    requests.post(
        url,
        data={
            "chat_id": ,
            "text": text
        }
    )

import os
import time
import cv2
import threading
import state
from concurrent.futures import ThreadPoolExecutor
from PIL import ImageGrab
import numpy as np
from helpers import log, update_cards_its_my_now, update_cards_its_enemy_now, update_cards_its_discard_now, save_previous_screen, send_final_cards_to_telegram


from templates import (
    TRUMPS,
    FINAL_STEPS,
    STEPS,
    END,
)

# Анализ Стола
def detect_table(previous_screen):
    start = time.perf_counter() # Отчитываем время старта для анализа

    previous_screen = previous_screen[147:616, 543:1054] # Кусок который идет на анализ
    templates = [] # Сюда пойдет (['JH', 'TH'], image) / (['AS'], image)

    for file in os.listdir("templates/table"): # Перебираем шаблоны
        filename = os.path.splitext(file)[0] # Убираем расширение png

        if "_" in filename: # Если в названии есть _ значит это пара
            cards = filename.split("_") # JH_TH становится ['JH', 'TH']
        else:
            cards = [filename] # ['JH']

        image = cv2.imread(f"templates/table/{file}")

        templates.append((cards, image))

    found_pairs = [] # Сюда попадают найденные пары
    found_singles = set() # Сюда соло карты

    lock = threading.Lock() # Синхронизация потоков
    stop_event = threading.Event() # Дропаем когда найденно достаточное кол-во элементов

    def check(template_data): # Проверяем шаблоны один за другим
        if stop_event.is_set():
            return None

        cards, template = template_data # (['JH', 'TH'], image) переходит в cards = ['JH', 'TH'] + template = image

        # Если это пара и одна из ее карт уже найдена одиночкой пару вообще не проверяем
        if len(cards) == 2:
            with lock:
                if cards[0] in found_singles or cards[1] in found_singles:
                    return None

        _, score, _, _ = cv2.minMaxLoc(
            cv2.matchTemplate(
                previous_screen,
                template,
                cv2.TM_CCORR_NORMED
            )
        )

        if score <= 0.976: # Адекватный параметр
            return None

        with lock: # Если нашли попадаем сюда
            if len(cards) == 2: # Проверяем пара это или нет
                # Если одну из карт пары уже нашли одиночкой эту пару пропускаем (по сути дубль проверка)
                if cards[0] in found_singles or cards[1] in found_singles:
                    return

                pair = tuple(cards) # ['JH', 'TH'] -> ('JH', 'TH')

                if pair not in found_pairs:
                    found_pairs.append(pair) # Добавляем найденную пару

                if len(found_pairs) + len(found_singles) >= 6: # считается количество элементов стола, а не физических карт
                    stop_event.set() # Дропаем на 6

                return

            card = cards[0] # если это одиночная карта

            if card not in found_singles: # Добавляем одиночную карту
                found_singles.add(card)

            if len(found_pairs) + len(found_singles) >= 6: # считается количество элементов стола, а не физических карт
                stop_event.set() # Дропаем на 6

    # Разделяем шаблоны на пары и одиночные
    pair_templates = [item for item in templates if len(item[0]) == 2]
    single_templates = [item for item in templates if len(item[0]) == 1]

    # Сначала ищем одиночки
    with ThreadPoolExecutor(max_workers=4) as executor:
        for result in executor.map(check, single_templates):
            pass

    # это легаси проверка которая была когда проверяли сначала пары
    if len(found_pairs) + len(found_singles) < 6:
        stop_event.clear()

        with ThreadPoolExecutor(max_workers=4) as executor:
            for result in executor.map(check, pair_templates):
                pass

    found = set() # финальный набор

    for pair in found_pairs: # ('JH', 'TH'), - >  'JH', 'TH',
        found.update(pair)

    found.update(found_singles) # Добавляем одиночные

    # Однажды оно вернуло {'JH.', 'KH'}, почему, не понятно, но в остальном нормально было, убрали точку  {'JH.', 'KH'} станет {'JH', 'KH'}
    found = {card[:2] if len(card) == 3 and card.endswith(".") else card for card in found}

    if found:
        log(f'Найдены карты на столе: {found}')

    log(f'Время выполнения detect_table: {time.perf_counter() - start:.4f} сек')

    return sorted(found)


# Добор + он же анализ начально полученных карт
def detect_draw(screen):
    start = time.perf_counter() # Запускаем таймер для понимании за сколько отработала функция

    screen = screen[577:829, 539:1065]
    templates = [] # Сюда идут все шаблоны

    for file in os.listdir("templates/start"):
        name = file[:2] # Первые два символа имени файла это название карты, типа 9S
        image = cv2.imread(f"templates/start/{file}")

        templates.append((name, image)) # название карты + шаблон

    found = set() # сюда идут найденные
    lock = threading.Lock() # это нужно для синхронизации потоков
    stop_event = threading.Event() # это нужно что бы дропнуть функцию после того как нашли 6

    # Функция для проверки одной штуки шаблон
    def check(template):
        if stop_event.is_set(): # Если нашли уже 6 больше не надо
            return None

        name, template = template # Название карты + шаблон

        with lock: # не была ли эта карта уже найдена другим потоком (на другой позиции)
            if name in found:
                return None

        _, score, _, _ = cv2.minMaxLoc( # сравнимаем шаблон + скриншот
            cv2.matchTemplate(
                screen,
                template,
                cv2.TM_CCORR_NORMED
            )
        )
        # Проверка совпадения. Изначально было на 0.993 но так слишком жестко получалось, на 0.991 вроде очень хорошо отрабатывает
        if score > 0.991:
            with lock:
                found.add(name) # Добавляем найденную карту

                if len(found) >= 6: # Если нашли 6 карт дропаем
                    stop_event.set()

            return name

        return None

    # Запускаем проверку в 4 потоках для всех шаблонов
    with ThreadPoolExecutor(max_workers=4) as executor:
        for name in executor.map(check, templates):
            pass

    log(f'Найдены карты: {found}')
    log(f'Время выполнения detect_draw: 'f'{time.perf_counter() - start:.4f} сек')
    return sorted(found)


# Берет 4 возможных козыря и делает сверку - ВСЕ
def detect_trump(screen):
    screen = screen[301:455, 540:641]
    for suit, template in TRUMPS.items():
        result = cv2.matchTemplate(
            screen,
            template,
            cv2.TM_CCORR_NORMED
        )

        _, score, _, _ = cv2.minMaxLoc(result)

        if score > 0.99:
            return suit

    return None

def check_final_step(screen):
    trump = state.game['trump'] # узнаем козырь

    if trump is None:
        return False

    template = FINAL_STEPS[trump] # нужен что бы сократить количество сверок с 4 до 1

    screen = screen[280:406, 544:652]
    result = cv2.matchTemplate(
        screen,
        template,
        cv2.TM_CCORR_NORMED
    )

    _, score, _, _ = cv2.minMaxLoc(result)

    if score > 0.99:
        log(f'{trump} - финальный шаг')
        final_cards = [
            card
            for card, status in state.cards.items()
            if status in (state.UNKNOWN, state.ENEMY)
        ]
        log(f'Карты на финальном шаге: {final_cards}')
        send_final_cards_to_telegram(final_cards)
        return True

    return False



# Проверяем а не закончилась ли игра
def check_game_end(screen):
    screen = screen[9:868, 720:875]
    result = cv2.matchTemplate(
        screen,
        END,
        cv2.TM_CCOEFF_NORMED
    )

    _, score, _, _ = cv2.minMaxLoc(result)

    return score > 0.9


def check_steps(screen, previous_screen):

    screen = screen[13:868, 739:866]
    # Проверка DONE универсальна (сразу и на вверх и на низ)
    result = cv2.matchTemplate(
        screen,
        STEPS['done'],
        cv2.TM_CCORR_NORMED
    )

    _, score, _, _ = cv2.minMaxLoc(result)

    if score > 0.99:
        log('Done найден')
        log('Нужно проверить I take')

        result = cv2.matchTemplate(
            screen,
            STEPS['i_take_up'],
            cv2.TM_CCORR_NORMED
        )

        _, score_up, _, _ = cv2.minMaxLoc(result)

        if score_up > 0.99:
            log('I take найден вверху')
            log('Проверили стол')
            save_previous_screen(previous_screen)
            founds_cards_on_table = detect_table(previous_screen)
            log(f'Найдены карты: {founds_cards_on_table}')
            log('Найденные карты статус Enemy')
            update_cards_its_enemy_now(founds_cards_on_table)
            log('Проверка добора')
            time.sleep(1)
            pure_screen = np.array(ImageGrab.grab(all_screens=True)) # Тут делаем скрин, нужен более свежий, тот что летает на анализе он немного не готов
            pure_screen = cv2.cvtColor(pure_screen,cv2.COLOR_RGB2BGR)
            found_cards = detect_draw(pure_screen)
            log(f'Найдены карты: {found_cards}')
            update_cards_its_my_now(found_cards)
            return True

        # I TAKE DOWN
        result = cv2.matchTemplate(
            screen,
            STEPS['i_take_down'],
            cv2.TM_CCORR_NORMED
        )

        _, score_down, _, _ = cv2.minMaxLoc(result)

        if score_down > 0.99:
            log('I take найден внизу')
            log('Проверили стол')
            save_previous_screen(previous_screen)
            founds_cards_on_table = detect_table(previous_screen)
            log(f'Найдены карты: {founds_cards_on_table}')
            log('Найденные карты статус Me')
            update_cards_its_my_now(founds_cards_on_table)
            log('Проверка добора')
            time.sleep(1)
            pure_screen = np.array(ImageGrab.grab(all_screens=True)) # Тут делаем скрин, нужен более свежий, тот что летает на анализе он немного не готов
            pure_screen = cv2.cvtColor(pure_screen,cv2.COLOR_RGB2BGR)
            found_cards = detect_draw(pure_screen)
            log(f'Найдены карты: {found_cards}')
            update_cards_its_my_now(found_cards)
            return True

        log('I take не найден')
        log('Проверили стол')
        save_previous_screen(previous_screen)
        founds_cards_on_table = detect_table(previous_screen)
        log(f'Найдены карты: {founds_cards_on_table}')
        log('Найденные карты вышли')
        update_cards_its_discard_now(founds_cards_on_table)
        log('Проверка добора')
        time.sleep(1)
        pure_screen = np.array(ImageGrab.grab(all_screens=True)) # Тут делаем скрин, нужен более свежий, тот что летает на анализе он немного не готов
        pure_screen = cv2.cvtColor(pure_screen,cv2.COLOR_RGB2BGR)
        found_cards = detect_draw(pure_screen)
        log(f'Найдены карты: {found_cards}')
        update_cards_its_my_now(found_cards)
        return True

    # PASS UP
    result = cv2.matchTemplate(
        screen,
        STEPS['pass_up'],
        cv2.TM_CCORR_NORMED
    )

    _, score_up, _, _ = cv2.minMaxLoc(result)

    if score_up > 0.99:
        log('Опонент пассует - у нас I take')
        log('Проверяем стол')
        save_previous_screen(previous_screen)
        founds_cards_on_table = detect_table(previous_screen)
        log(f'Найдены карты: {founds_cards_on_table}')
        log('Найденные карты статус Me')
        update_cards_its_my_now(founds_cards_on_table)
        return True

    # PASS DOWN
    result = cv2.matchTemplate(
        screen,
        STEPS['pass_down'],
        cv2.TM_CCORR_NORMED
    )

    _, score_down, _, _ = cv2.minMaxLoc(result)

    if score_down > 0.99:
        log('Мы пассуем - у Enemy I take')
        log('Проверяем стол')
        save_previous_screen(previous_screen)
        founds_cards_on_table = detect_table(previous_screen)
        log(f'Найдены карты: {founds_cards_on_table}')
        log('Найденные карты статус Enemy')
        update_cards_its_enemy_now(founds_cards_on_table)
        return True

    return False