diff --git a/example_live_parser.py b/example_live_parser.py new file mode 100644 index 0000000..4564455 --- /dev/null +++ b/example_live_parser.py @@ -0,0 +1,41 @@ +import traceback +from time import sleep + +from hslog.live.parser import LiveLogParser + + +def main(): + """ + ---------------------------------------------------------------------- + LiveLogParser assumes that you"ve configured Power.log to be a symlink. + + In "SOME_PATH/Hearthstone/Logs" folder: + ln -s Power.log /tmp/hearthstone-redirected.log + + This will redirect all data coming into Power.log + so we can access it from a RAM disk. + ---------------------------------------------------------------------- + For better performance make /tmp of type tmpfs (or another location) + + In /etc/fstab add line: + tmpfs /tmp tmpfs nodev,nosuid,size=1G 0 0 + + This will create in-memory storage which is faster then SSD. + You need to restart the computer for this to take effect. + ---------------------------------------------------------------------- + """ + try: + file = "/tmp/hearthstone-redirected.log" + liveParser = LiveLogParser(file) + liveParser.start() + + while True: + sleep(1) + + except Exception as e: + print(traceback.format_exc()) + liveParser.stop() + + +if __name__ == "__main__": + main() diff --git a/hslog/live/__init__.py b/hslog/live/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/hslog/live/entities.py b/hslog/live/entities.py new file mode 100644 index 0000000..31d7590 --- /dev/null +++ b/hslog/live/entities.py @@ -0,0 +1,90 @@ +from hearthstone.entities import Card, Entity, Game, Player +from hearthstone.enums import GameTag + +from hslog.live.utils import terminal_output + + +class LiveEntity(Entity): + + def __init__(self, entity_id): + super(LiveEntity, self).__init__(entity_id) + self._game = None + + @property + def game(self): + return self._game + + @game.setter + def game(self, value): + # this happens when game calls register_entity and entity sets self.game + self._game = value + if value is not None: + terminal_output("ENTITY CREATED", self) + # push data to an end-point + pass + + def tag_change(self, tag, value): + if tag == GameTag.CONTROLLER and not self._initial_controller: + self._initial_controller = self.tags.get(GameTag.CONTROLLER, value) + self.tags[tag] = value + terminal_output("TAG UPDATED", self, tag, value) + # push data to an end-point + pass + + def update_callback(self, caller): + terminal_output("ENTITY UPDATED", self) + # push data to an end-point + pass + + +class LiveCard(Card, LiveEntity): + """ + Card is called on export from game + LiveCard replaces Card and inserts update_callback + The point is to become able to route update events towards an API end-point + """ + + def __init__(self, entity_id, card_id): + super(LiveCard, self).__init__(entity_id, card_id) + + """ + if card_id doesn"t change, there"s no need to pass it as the argument. + we can use self.card_id instead as it is set by Card class + """ + def reveal(self, card_id, tags): + self.revealed = True + self.card_id = card_id + if self.initial_card_id is None: + self.initial_card_id = card_id + self.tags.update(tags) + + # update notify + self.update_callback(self) + + def hide(self): + self.revealed = False + + # update notify + self.update_callback(self) + + """ same comment as for reveal """ + def change(self, card_id, tags): + if self.initial_card_id is None: + self.initial_card_id = card_id + self.card_id = card_id + self.tags.update(tags) + + # update notify + self.update_callback(self) + + +class LivePlayer(Player, LiveEntity): + + def __init__(self, packet_id, player_id, hi, lo, name=None): + super(LivePlayer, self).__init__(packet_id, player_id, hi, lo, name) + + +class LiveGame(Game, LiveEntity): + + def __init__(self, entity_id): + super(LiveGame, self).__init__(entity_id) diff --git a/hslog/live/export.py b/hslog/live/export.py new file mode 100644 index 0000000..13be7fd --- /dev/null +++ b/hslog/live/export.py @@ -0,0 +1,31 @@ +from hearthstone.enums import GameTag + +from hslog.export import EntityTreeExporter +from hslog.live.entities import LiveCard, LiveGame, LivePlayer + + +class LiveEntityTreeExporter(EntityTreeExporter): + """ + Inherits EntityTreeExporter to provide Live entities + """ + + game_class = LiveGame + player_class = LivePlayer + card_class = LiveCard + + def __init__(self, packet_tree): + super(LiveEntityTreeExporter, self).__init__(packet_tree) + + def handle_player(self, packet): + entity_id = int(packet.entity) + + if hasattr(self.packet_tree, "manager"): + # If we have a PlayerManager, first we mutate the CreateGame.Player packet. + # This will have to change if we"re ever able to immediately get the names. + player = self.packet_tree.manager.get_player_by_id(entity_id) + packet.name = player.name + entity = self.player_class(entity_id, packet.player_id, packet.hi, packet.lo, packet.name) + entity.tags = dict(packet.tags) + self.game.register_entity(entity) + entity.initial_hero_entity_id = entity.tags.get(GameTag.HERO_ENTITY, 0) + return entity diff --git a/hslog/live/packets.py b/hslog/live/packets.py new file mode 100644 index 0000000..d2226b0 --- /dev/null +++ b/hslog/live/packets.py @@ -0,0 +1,17 @@ +from hslog.live.export import LiveEntityTreeExporter +from hslog.packets import PacketTree + + +class LivePacketTree(PacketTree): + + def __init__(self, ts, parser): + self.parser = parser + self.liveExporter = LiveEntityTreeExporter(self) + super(LivePacketTree, self).__init__(ts) + + def live_export(self, packet): + """ + Triggers packet export which will run the proper handler for the packet. + This will also run update_callback for entity being updated by the packet. + """ + return self.liveExporter.export_packet(packet) diff --git a/hslog/live/parser.py b/hslog/live/parser.py new file mode 100644 index 0000000..ea2033a --- /dev/null +++ b/hslog/live/parser.py @@ -0,0 +1,154 @@ +import time +from collections import deque +from threading import Thread + +from hslog import packets, tokens +from hslog.exceptions import RegexParsingError +from hslog.live.packets import LivePacketTree +from hslog.live.player import LivePlayerManager +from hslog.parser import LogParser +from hslog.player import LazyPlayer +from hslog.utils import parse_tag + + +class LiveLogParser(LogParser): + """ + LiveLogParser provides live log translation into useful data. + + Lines are read and pushed into a deque by a separate thread. + Deque is emptied by parse_worker which replaces the read() + function of LogParser and it"s also in a separate thread. + + This approach is non-blocking and allows for live parsing + of incoming lines. + """ + + def __init__(self, filepath): + super(LiveLogParser, self).__init__() + self.running = False + self.filepath = filepath + self.lines_deque = deque([]) + + def new_packet_tree(self, ts): + """ + LivePacketTree is introduced here because it instantiates LiveEntityTreeExporter + and keeps track of the parser parent. It also contains a function that + utilizes the liveExporter instance across all the games. + + self.parser = parser + self.liveExporter = LiveEntityTreeExporter(self) + """ + self._packets = LivePacketTree(ts, self) + self._packets.spectator_mode = self.spectator_mode + self._packets.manager = LivePlayerManager() + self.current_block = self._packets + self.games.append(self._packets) + + def handle_entities_chosen(self, ts, data): + super(LiveLogParser, self).handle_entities_chosen(ts, data) + if data.startswith("id="): + sre = tokens.ENTITIES_CHOSEN_RE.match(data) + if not sre: + raise RegexParsingError(data) + player_name = sre.groups()[1] + + # pick up opponent name from GameState.DebugPrintEntitiesChosen() + m = self._packets.manager + m.complete_player_names(player_name, self._packets) + + def handle_game(self, ts, data): + if data.startswith("PlayerID="): + sre = tokens.GAME_PLAYER_META.match(data) + if not sre: + raise RegexParsingError(data) + player_id, player_name = sre.groups() + + # set initial name based on GameState.DebugPrintGame() + m = self._packets.manager + m.set_initial_player_name(player_id, player_name, self._packets) + + player_id = int(player_id) + else: + super(LiveLogParser, self).handle_game(ts, data) + + def tag_change(self, ts, e, tag, value, def_change): + entity_id = self.parse_entity_or_player(e) + tag, value = parse_tag(tag, value) + self._check_for_mulligan_hack(ts, tag, value) + + if isinstance(entity_id, LazyPlayer): + entity_id = self._packets.manager.register_player_name_on_tag_change( + entity_id, tag, value) + + has_change_def = def_change == tokens.DEF_CHANGE + packet = packets.TagChange(ts, entity_id, tag, value, has_change_def) + if entity_id: + self.register_packet(packet) + return packet + + def register_packet(self, packet, node=None): + """ + LogParser.register_packet override + + This uses the live_export functionality introduces by LivePacketTree + It also keeps track of which LivePacketTree is being used when there + are multiple in parser.games + + A better naming for a PacketTree/LivePacketTree would be HearthstoneGame? + Then parser.games would contain HearthstoneGame instances and would + be more obvious what the purpose is. + """ + + # make sure we"re registering packets to the current game + if not self._packets or self._packets != self.games[-1]: + self._packets = self.games[-1] + + if node is None: + node = self.current_block.packets + node.append(packet) + self._packets._packet_counter += 1 + packet.packet_id = self._packets._packet_counter + self._packets.live_export(packet) + + def file_worker(self): + """ + File reader thread. (Naive implementation) + Reads the log file continuously and appends to deque. + """ + + file = open(self.filepath, "r") + while self.running: + line = file.readline() + if line: + self.lines_deque.append(line) + else: + time.sleep(0.2) + + def parse_worker(self): + """ + If deque contains lines, this initiates parsing. + """ + while self.running: + if len(self.lines_deque): + line = self.lines_deque.popleft() + self.read_line(line) + else: + time.sleep(0.2) + + def start_file_worker(self): + file_thread = Thread(target=self.file_worker) + file_thread.setDaemon(True) + file_thread.start() + + def start_parse_worker(self): + parse_thread = Thread(target=self.parse_worker) + parse_thread.setDaemon(True) + parse_thread.start() + + def start(self): + self.running = True + self.start_file_worker() + self.start_parse_worker() + + def stop(self): + self.running = False diff --git a/hslog/live/player.py b/hslog/live/player.py new file mode 100644 index 0000000..5e4203f --- /dev/null +++ b/hslog/live/player.py @@ -0,0 +1,45 @@ +from hearthstone.enums import GameTag + +from hslog.player import PlayerManager + + +class LivePlayerManager(PlayerManager): + + def __init__(self): + super(LivePlayerManager, self).__init__() + + self.actual_player_names = set() + self.names_used = set() + self.name_assignment_done = False + + def register_player_name_on_tag_change(self, player, tag, value): + if tag not in [GameTag.ENTITY_ID, GameTag.LAST_CARD_PLAYED]: + return None + super(LivePlayerManager, self).register_player_name_on_tag_change(player, tag, value) + + def set_initial_player_name(self, player_id, player_name, current_game): + players = current_game.liveExporter.game.players + for p in players: + if p.player_id == int(player_id): + p.name = player_name + + def complete_player_names(self, player_name, current_game): + # populate names if they haven"t been used already + if player_name not in self.names_used: + self.actual_player_names.add(player_name) + + # if there are two names available we have enough to assign + if len(self.actual_player_names) == 2 and not self.name_assignment_done: + unnamed = None + for p in current_game.liveExporter.game.players: + if p.name in self.actual_player_names: + self.names_used.add(p.name) + self.actual_player_names.remove(p.name) + else: + unnamed = p + if len(self.actual_player_names): + other_player_name = self.actual_player_names.pop() + self.names_used.add(other_player_name) + unnamed.name = other_player_name + + self.name_assignment_done = True diff --git a/hslog/live/utils.py b/hslog/live/utils.py new file mode 100644 index 0000000..d3c15d8 --- /dev/null +++ b/hslog/live/utils.py @@ -0,0 +1,58 @@ +def align(text, size, char=" "): + """ + Format text to fit into a predefined amount of space + + Positive size aligns text to the left + Negative size aligns text to the right + """ + text = str(text).strip() + text_len = len(text) + if text_len > abs(size): + text = f"{text[:size-3]}..." + offset = "".join(char * (abs(size) - text_len)) + if size < 0: + return f"{offset}{text}" + else: + return f"{text}{offset}" + + +def color(): + def color_decorator(func): + colors = { + "LivePlayer": 93, + "red": 31, + "green": 32, + "LiveGame": 33, + "blue": 34, + "purple": 35, + "cyan": 36, + "grey": 37, + } + + def func_wrapper(msg_type, obj, *args): + class_name = obj.__class__.__name__ + color_key = class_name if class_name in colors else "green" + line = "\033[{}m{}\033[0m".format(colors[color_key], func(msg_type, obj, *args)) + print(line) + + return func_wrapper + + return color_decorator + + +@color() +def terminal_output(msg_type, obj, attr=None, value=None): + return "{} | {} | {} | {}".format( + align(msg_type, -20), + align(repr(obj), 120), + align(repr(attr), 40), + align(value, 30), + ) + + +def debug_player_names(player_manager): + print("{} | {} | {}".format( + align(player_manager.actual_player_names, 40), + align(player_manager.names_used, 40), + align(player_manager.name_assignment_done, 10), + ))