Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _register_choices(self, ts, id, player, tasklist, type, min, max):
id = int(id)
type = parse_enum(enums.ChoiceType, type)
min, max = int(min), int(max)
self._choice_packet = packets.Choices(ts, player, id, tasklist, type, min, max)
self.current_block.packets.append(self._choice_packet)
return self._choice_packet
def handle_send_choices(self, ts, data):
if data.startswith("id="):
sre = tokens.SEND_CHOICES_CHOICE_RE.match(data)
if not sre:
raise RegexParsingError(data)
id, type = sre.groups()
id = int(id)
type = parse_enum(enums.ChoiceType, type)
self._send_choice_packet = packets.SendChoices(ts, id, type)
self.current_block.packets.append(self._send_choice_packet)
return self._send_choice_packet
elif data.startswith("m_chosenEntities"):
sre = tokens.SEND_CHOICES_ENTITIES_RE.match(data)
if not sre:
raise RegexParsingError(data)
idx, entity = sre.groups()
id = self.parse_entity_or_player(entity)
if not id:
raise ParsingError("Missing chosen entity %r (%r)" % (id, entity))
if not self._send_choice_packet:
raise ParsingError("Chosen Entity outside of choice packet: %r" % (data))
self._send_choice_packet.choices.append(id)
return id
raise NotImplementedError("Unhandled send choice: %r" % (data))
def full_entity(self, ts, id, card_id):
id = int(id)
self._entity_packet = packets.FullEntity(ts, id, card_id)
self.current_block.packets.append(self._entity_packet)
if self._creating_game:
# First packet after create game should always be a FULL_ENTITY
self._creating_game = False
# It should always have ID 4
if id != 4:
raise ParsingError("Expected entity 4 after creating game, got %r" % (id))
# While we're at it, we check if we got an abnormal amount of players
player_count = len(self._game_packet.players)
if player_count != 2:
raise ParsingError("Expected exactly 2 players, got %r" % (player_count))
return self._entity_packet
def handle_send_option(self, ts, data):
if data.startswith("selectedOption="):
sre = tokens.SEND_OPTION_RE.match(data)
if not sre:
raise RegexParsingError(data)
option, suboption, target, position = sre.groups()
packet = packets.SendOption(ts, int(option), int(suboption), int(target), int(position))
self.current_block.packets.append(packet)
return packet
raise NotImplementedError("Unhandled send option: %r" % (data))
def register_player(self, ts, id, player_id, hi, lo):
id = int(id)
player_id = int(player_id)
hi = int(hi)
lo = int(lo)
lazy_player = self._packets.manager.new_player(id, player_id, is_ai=lo == 0)
self._entity_packet = packets.CreateGame.Player(ts, lazy_player, player_id, hi, lo)
self._game_packet.players.append(self._entity_packet)
return lazy_player
def handle_entities_chosen(self, ts, data):
if data.startswith("id="):
sre = tokens.ENTITIES_CHOSEN_RE.match(data)
if not sre:
raise RegexParsingError(data)
id, player, count = sre.groups()
id = int(id)
player = self.parse_entity_or_player(player)
self._chosen_packet_count = int(count)
self._chosen_packet = packets.ChosenEntities(ts, player, id)
self.current_block.packets.append(self._chosen_packet)
return self._chosen_packet
elif data.startswith("Entities["):
sre = tokens.ENTITIES_CHOSEN_ENTITIES_RE.match(data)
if not sre:
raise RegexParsingError(data)
idx, entity = sre.groups()
id = self.parse_entity_or_player(entity)
if not id:
raise ParsingError("Missing entity chosen %r (%r)" % (id, entity))
if not self._chosen_packet:
raise ParsingError("Entity Chosen outside of choice packet: %r" % (data))
self._chosen_packet.choices.append(id)
if len(self._chosen_packet.choices) > self._chosen_packet_count:
raise ParsingError("Too many choices (expected %r)" % (self._chosen_packet_count))
return id
def _check_for_mulligan_hack(self, ts, tag, value):
# Old game logs didn't handle asynchronous mulligans properly.
# If we're missing an ACTION_END packet after the mulligan SendChoices,
# we just close it out manually.
if tag == enums.GameTag.MULLIGAN_STATE and value == enums.Mulligan.DEALING:
assert self.current_block
if isinstance(self.current_block, packets.Block):
logging.warning("WARNING: Broken mulligan nesting. Working around...")
self.block_end(ts)
def get_dispatch_dict(self):
return {
packets.CreateGame: self.handle_create_game,
packets.CreateGame.Player: self.handle_player,
packets.Block: self.handle_block,
packets.FullEntity: self.handle_full_entity,
packets.HideEntity: self.handle_hide_entity,
packets.ShowEntity: self.handle_show_entity,
packets.ChangeEntity: self.handle_change_entity,
packets.TagChange: self.handle_tag_change,
packets.MetaData: self.handle_metadata,
packets.Choices: self.handle_choices,
packets.SendChoices: self.handle_send_choices,
packets.ChosenEntities: self.handle_chosen_entities,
packets.Options: self.handle_options,
packets.Option: self.handle_option,
packets.SendOption: self.handle_send_option,
}
def handle_options(self, ts, data):
if data.startswith("id="):
sre = tokens.OPTIONS_ENTITY_RE.match(data)
if not sre:
raise RegexParsingError(data)
id, = sre.groups()
id = int(id)
self._options_packet = packets.Options(ts, id)
self.current_block.packets.append(self._options_packet)
elif data.startswith("option "):
return self._parse_option_packet(ts, data)
elif data.startswith(("subOption ", "target ")):
return self._parse_suboption_packet(ts, data)
optype, id, type, entity, error, error_param = sre.groups()
if not sre:
raise RegexParsingError(data)
error, error_param = clean_option_errors(error, error_param)
else:
sre = tokens.OPTIONS_OPTION_RE.match(data)
if not sre:
raise RegexParsingError(data)
optype, id, type, entity = sre.groups()
error, error_param = None, None
id = int(id)
type = parse_enum(enums.OptionType, type)
if entity:
entity = self.parse_entity_or_player(entity)
self._option_packet = packets.Option(ts, entity, id, type, optype, error, error_param)
if not self._options_packet:
raise ParsingError("Option without a parent option group: %r" % (data))
self._options_packet.options.append(self._option_packet)
self._suboption_packet = None
return self._option_packet