В части 2 я привел способ обнаружения соперника в локальной сети. Как только двое игроков нашли друг друга, они могут начать игру, обмениваясь UDP пакетами напрямую, т.е. посылая их непосредственно по IP адресу и нужному порту.
Да, номера портов вы выбираете сами. Старайтесь выбирать порты так, чтобы они не перекрывались с существующими сервисами (список популярных портов). Порты 0-1023 системные, их вы не займете просто так без особых прав. Берите любые порты из диапазона 1024-49151, лучше ближе к концу, так меньше вероятности конфликтов.
Если для протокола обнаружения нам нужен был только один порт 37020, то для обмена сообщениями между клиентами я возьму 2 порта: 37020 и 37021. Почему два? Потому что мы с вами отлаживаем код обычно на одной машине. А двум программам слушать один и тот же порт на одной машине не получится, даже переиспользуя порт. Сообщения просто будут приходить только на один из клиентов во всех случаях, а второй не получит ничего. Следовательно делатем так: один клиент слушает 37020 и отправляет сообщения на 37021, а другой – наоборот слушает 37021 и отправляет сообщения на 37020. Такая схема будет работать и на одной машине, и на разных машинах.
Примечание. Эту проблему, вроде бы, можно решить путем мультикаст-групп, но это выходит за рамки статьи: нам гораздо проще просто взять два разных порта.
Итак, костяк нашей игры будет лежать в модуле main.py:
from render import ConsoleRenderer
from net_game import DurakNetGame
from discovery_protocol import DiscoveryProtocol
from util import rand_id
PORT_NO = 37020
PORT_NO_AUX = 37021
def main():
my_pid = rand_id() # создадим себе ID случайно
discovery = DiscoveryProtocol(my_pid, port_no=PORT_NO)
print('Сканирую локальную сеть...')
(remote_addr, _port), remote_pid = discovery.run()
del discovery
renderer = ConsoleRenderer()
game = DurakNetGame(renderer, my_pid, remote_pid, remote_addr, [PORT_NO, PORT_NO_AUX])
game.start()
if __name__ == '__main__':
main()
Класс DiscoveryProtocol был описан в части 2, а класс ConsoleRenderer аж в первой части. Мы же сейчас обратимся к классу DurakNetGame, код которого я поместил в файл net_game.py.
Класс требует ID локального игрока, а также только что найденные ID удаленного игрока и его IP адрес в сети. Еще ему нужна пара номеров портов для коммуникации.
Так как у каждого есть свой ID и это просто числа, то мы выбираем того, кто ходит первый просто сравнивая эти числа: у кого меньше, тот и первый, а другой про себя решит точно также, что он второй. Аналогично идет выбор портов. Если я первый (мой ID меньше твоего), то слушаю, допустим, порт 37020. Другой клиент также совершенно однозначно решит, слушать порт 37021, сравнив свой ID с моим и поняв, что он больше. Вот такой элементарный алгоритм консенсуса. Так инициализируется класс сетевой игры:
class DurakNetGame:
def __init__(self, renderer: GameRenderer, my_id, remote_id, remote_addr, ports):
self._renderer = renderer
self._game = DurakSerialized() # геймплей
self._my_id = int(my_id)
self._remote_id = int(remote_id)
self._remote_addr = remote_addr
# проверка на адекватность ID
assert self._my_id != 0 and self._remote_id != 0 and self._my_id != self._remote_id
# кто ходит первый выбираем просто сравнивая ID (они же случайные)!
me_first = self._my_id < self._remote_id
# мой индекс 0 если я первый, и 1 иначе. у соперника наоборот
self._my_index = 0 if me_first else 1
# две сетевых примочки на разны портах
network1 = Networking(port_no=ports[0])
network2 = Networking(port_no=ports[1])
# кто слушает какой порт выбираем также на базе сравнения ID как чисел
self._receiver = network1 if me_first else network2
self._receiver.bind("")
self._sender = network2 if me_first else network1
Запуск игры (метод start) инициализирует новую игру (если я первый ходящий), и отсылает ее другому игроку, чтобы синхронизировать их состояния.
def _new_game(self):
self._game = DurakSerialized() # игрок с индексом 0 создает игру!
self._send_game_state() # и отсылает ее сопернику
self._renderer.render_game(self._game, self._my_index)
Сериализация состояния игры происходит в классе DurakSerialized – наследнике класса Durak. Она очень простая, почти без ухищрений: просто собираем в словарь разные аспекты игры. Код сериализации находится в файле serialization.py. Пример состояния игры после сериализации:
{'trump': '♦', 'attacker_index': 0, 'deck': [('J', '♠'), ('10', '♦'), ('8', '♣'), ('10', '♠'), ('A', '♦'), ('6', '♣'), ('7', '♣'), ('6', '♠'), ('Q', '♥'), ('J', '♣'), ('K', '♣'), ('6', '♦'), ('9', '♣'), ('6', '♥'), ('7', '♦'), ('9', '♠'), ('A', '♣'), ('7', '♥'), ('10', '♣'), ('A', '♠'), ('J', '♦'), ('8', '♥'), ('A', '♥'), ('7', '♠')], 'winner': None, 'field': [(('8', '♠'), None), (('8', '♦'), ('Q', '♦'))], 'players': [{'index': 0, 'cards': [('9', '♥'), ('J', '♥'), ('Q', '♠'), ('Q', '♣')]}, {'index': 1, 'cards': [('9', '♦'), ('10', '♥'), ('K', '♠'), ('K', '♥'), ('K', '♦')]}]}
Возвращаясь к запуску сетевой игры, он подразумевает запуск двух вещей: потока чтения и цикла-обработчика ввода пользователя.
def start(self):
print(f'Мой ID #{self._my_id}, мой индекс {self._my_index}')
print(f'Удаленный адрес {self._remote_addr}')
self._renderer.help()
if self._my_index == 0:
# игрок с индексом 0 создает игру!
self._new_game()
self._receiver.run_reader_thread(self._on_remote_message)
self._game_loop()
Первая часть – отдельный поток, который слушает порт на прием. На этот порт приходят сообщения от клиента соперника, содержащие закодированное состояние игры, если оно изменится в результате действий и ходов соперника. Ведь игра в дурака – синхронная: пока я отбиваюсь, ты можешь подкидывать мне еще карт того же достоинства, что уже лежат на столе. В networking.py определен метод запуска потока чтения, который вызывает стороннюю функцию-обработчик callback.
def run_reader_thread(self, callback):
"""
Запускает отдельный поток, чтобы получать данные из сокета
:param callback: функция, которая вызывается, если получены данные
"""
def reader_job():
while True:
data, _ = self.recv_json()
if data:
callback(data)
# daemon=True, чтобы не зависал, если выйдет основной поток
threading.Thread(target=reader_job, daemon=True).start()
Обработка сообщений от соперника идет в DurakNetGame и включают в себя обновление состояния игры локально и его отрисовку в терминале. Соперник еще может прислать сообщение, что он покидает игру, тогда мы тоже выйдем из программы.
def _on_remote_message(self, data):
action = data['action']
if action == 'state':
self._game = DurakSerialized(data['state']) # обновить остояние
print('Пришел ход от соперника!')
self._renderer.render_game(self._game, self._my_index)
elif action == 'quit':
print('Соперник вышел!')
exit(0)
Вторая ключевая вещь игры – это бесконечный цикл, который читает пользовательский ввод с клавиатуры и обрабатывает его. Если я походил, отбился или еще что-то сделал, тогда я обновляю состояние игры у себя и отсылаю его копию другому игроку, чтобы тот отобразил изменения у себя на экране.
def _game_loop(self):
while True:
q = input('Ваш ход (q = выход)? ')
parts = q.lower().split(' ')
if not parts:
continue
command = parts[0]
good_move = False # флаг, удачный ли был ход после ввода команды
g = self._game
try:
if command == 'f':
good_move = self._handle_finish()
elif command == 'a' and g.attacker_index == self._my_index:
good_move = self._handle_attack(parts)
elif command == 'd' and g.attacker_index != self._my_index:
good_move = self._handle_defence(parts)
elif command == 'q':
print('Вы вышли из игры!')
self._send_quit()
break
else:
print('Неизвестная команда.')
except IndexError:
print('ОШИБКА! Неверный выбор карты')
except ValueError:
print('Введите число через пробел после команды')
if good_move:
# если ход удачный, пошлем копию состояния игры
self._send_game_state()
self._renderer.render_game(g, self._my_index)
# проверка на окончание игры (если есть победитель, в g.winner - этог индекс)
if g.winner is not None:
outcome = 'Вы победили!' if g.winner == self._my_index else 'Вы проиграли!'
print(f'Игра окончена! {outcome}')
break
Как и в случае локальной игры из первой части, мы читает строку с клавиатуры, где первый символ команда, а второй через пробел – опциональный аргумент. Команды будут такие:
- f – завершить ход (сказав «Бито!» или забрав карты, зависит от того, кто вызвал команду)
- a 2 – атакующий ход (аргумент – номер карты, с которой ходить или подбросить)
- d 1 – защитный ход (аргумент – номер карты, которой отбиться). Если выбор карты для покрытия неоднозначен, то спросим пользователя о том, какую именно карту следует отбить.
- q – выход из игры, уведомив соперника, чтобы тот тоже завершил цикл и вышел из программы.
def _game_loop(self):
while True:
q = input('Ваш ход (q = выход)? ')
parts = q.lower().split(' ')
if not parts:
continue
command = parts[0]
good_move = False # флаг, удачный ли был ход после ввода команды
g = self._game
try:
if command == 'f':
good_move = self._handle_finish()
elif command == 'a' and g.attacker_index == self._my_index:
good_move = self._handle_attack(parts)
elif command == 'd' and g.attacker_index != self._my_index:
good_move = self._handle_defence(parts)
elif command == 'q':
print('Вы вышли из игры!')
self._send_quit()
break
else:
print('Неизвестная команда.')
except IndexError:
print('ОШИБКА! Неверный выбор карты')
except ValueError:
print('Введите число через пробел после команды')
if good_move:
# если ход удачный, пошлем копию состояния игры
self._send_game_state()
self._renderer.render_game(g, self._my_index)
# проверка на окончание игры (если есть победитель, в g.winner - этог индекс)
if g.winner is not None:
outcome = 'Вы победили!' if g.winner == self._my_index else 'Вы проиграли!'
print(f'Игра окончена! {outcome}')
break
Если в результате ввода состоялся удачный ход (по правилам игры, не было исключений), то состояние игры локально обновиться. А мы будем обязаны уведомить об этом другую сторону. Код отправки сообщений:
def _send_game_state(self):
self._sender.send_json({
'action': 'state',
'state': self._game.serialized()
}, self._remote_addr)
def _send_quit(self):
self._sender.send_json({
'action': 'quit'
}, self._remote_addr)
Если в игре будет обнаружен победитель, то оба игрока выходят из цикла. Переходим к обработчикам каждой команды.
Атака. Атаковать, как понятно, можно только в свой ход. Возвращает True, если атака успешна, иными словами, если игрок сходил с одной из правильных карт (нельзя подкинуть десятку, если на столе только, например, семерки, валеты и дамы).
def _handle_attack(self, parts):
g = self._game
index = int(parts[1]) - 1
card = g.current_player.cards[index]
if not g.attack(card):
print('ОШИБКА! Вы не можете ходить этой картой!')
return False
else:
return True
Защита возможна только в ход соперника. Защищающийся выбирает карту, игра ищет варианты, которыми он может отбиться именно этой картой. Если вариантов несколько, будет спрошено, что он конкретно имеет в виду. Возвращает True, если успешно отбился одной картой, но это еще не значит, что ход завершен. Вот код метода:
def _handle_defence(self, parts):
g = self._game
index = int(parts[1]) - 1
new_card = g.opponent_player.cards[index]
if g.field:
variants = g.defend_variants(new_card)
print(f'variants {variants} - {new_card}')
if len(variants) == 1:
def_index = variants[0]
elif len(variants) >= 2:
max_pos = len(g.field)
def_index = int(input(f'Какую позицию отбить {new_card} (1-{max_pos})? ')) - 1
else:
print('Вам придется взять карты!')
return False
old_card = list(g.field.keys())[def_index]
if not g.defend(old_card, new_card):
print('ОШИБКА! Нельзя так отбиться!')
else:
return True
else:
print('Пока нечего отбивать!')
return False
Завершение хода. Если защищающийся игрок не отбил все карты, то завершение хода вынудит его забрать со стола все карты себе в руку. Ход не переходит. Если же он успешно отбился, то завершение хода – это «Бито!», ход перейдет, а все карты со стола уйдут из игры в стопку «бито».
def _handle_finish(self, my_turn):
g = self._game
if g.field:
if my_turn and g.any_unbeaten_cards:
print('Не можете вынудить соперника взять карты!')
return False
elif not my_turn and not g.any_unbeaten_cards:
print('Только атакующий может сказать "Бито!"')
return False
else:
r = g.finish_turn()
if r == g.GAME_OVER:
r_str = 'игра окончена!'
elif r == g.TOOK_CARDS:
r_str = 'взяли карты.'
else:
r_str = 'бито.'
print(f'Ход завершен: {r_str}')
return True
else:
print('Пока ход не сделал, чтобы его завершить!')
return False
Вот и весь код игры! Теперь можно запустить пару клиентов игры на одном или разных компьютерах и играть в дурака!
Напомню, что весь код в репозитории на Github. Ставьте звездочки, мне будет очень приятно.
Надеюсь, вам понравилось. Может быть прикрутим к игре графический интерфейс и нормальное управление?
Предыдущие части:
Специально для канала @pyway. Подписывайтесь на мой канал в Телеграм @pyway 👈