Source code for tarrasque.binding

import collections

[docs]class StreamBinding(object): """ The StreamBinding class is Tarrasque's metaphor for the replay. Every Tarrasque entity class has a reference to an instance of this class, and when the tick of the instance changes, the data returned by those classes changes. This makes it easy to handle complex object graphs without explicitly needing to pass the Skadi demo object around. .. note:: Where methods on this class take absolute tick values (i.e. the ``start`` and ``end`` arguments to :meth:`iter_ticks`), special string arguments may be passed. These are: * ``"start"`` - The start of the replay * ``"draft"`` - The start of the draft * ``"pregame"`` - The end of the draft phase * ``"game"`` - The time when the game clock hits 0 * ``"postgame"`` - The time the ancient is destroyed * ``"end"`` - The last tick in the replay These values will not be 100% accurate, but should be good +-50 ticks """ @property
[docs] def user_messages(self): """ The user messages for the current tick. """ return self._user_messages
@property
[docs] def game_events(self): """ The game events in the current tick. """ from .gameevents import create_game_event events = [] for data in self._game_events: events.append(create_game_event(stream_binding=self, data=data)) return events # Just another layer of indirection # These are properties for autodoc reasons mostly
@property
[docs] def world(self): """ The Skadi wold object for the current tick. """ return self._stream.world
@property
[docs] def tick(self): """ The current tick. """ return self._stream.tick
@property
[docs] def demo(self): """ The Skadi demo object that the binding is reading from. """ return self._demo
@property
[docs] def modifiers(self): """ The Skadi modifiers object for the tick. """ return self._stream.modifiers
@property
[docs] def string_tables(self): """ The string_table provided by Skadi. """ return self._stream.string_tables
@property
[docs] def prologue(self): """ The prologue of the replay. """ return self._stream.prologue
def __init__(self, demo, start_tick=None, start_time=None): self._demo = demo self._user_messages = [] self._game_events = [] self._state_change_ticks = { "start": 0, "end": self.demo.file_info.playback_ticks - 2 } self.go_to_tick(self.demo.file_info.playback_ticks - 2) self._state_change_times = { "draft": self.info.draft_start_time, "pregame": self.info.pregame_start_time, "game": self.info.game_start_time, "postgame": self.info.game_end_time } # We're already here! if start_tick == "end": pass elif start_tick is not None: self.go_to_tick(start_tick) elif start_time is not None: self.go_to_time(start_time) else: self.go_to_state_change("game")
[docs] def iter_ticks(self, start=None, end=None, step=1): """ A generator that iterates through the demo's ticks and updates the :class:`StreamBinding` to that tick. Yields the current tick. The start parameter defines the tick to iterate from, and if not set, the current tick will be used instead. The end parameter defines the point to stop iterating; if not set, the iteration will continue until the end of the replay. The step parameter is the number of ticks to consume before yielding the tick; the default of one means that every tick will be yielded. Do not assume that the step is precise; the gap between two ticks will always be larger than the step, but usually not equal to it. """ if start is not None: self.go_to_tick(start) if end is None: end = self._state_change_ticks["end"] self._user_messages = [] self._game_events = [] last_tick = self.tick - step - 1 for _ in self._stream: if isinstance(end, str): if self.info.game_state == end: break elif self.tick >= end: break self._user_messages.extend(self._stream.user_messages) self._game_events.extend(self._stream.game_events) if self.tick - last_tick < step: continue else: last_tick = self.tick yield self.tick self._user_messages = [] self._game_events = []
[docs] def iter_full_ticks(self, start=None, end=None): """ A generator that iterates through the demo's 'full ticks'; sync points that occur once a minute. Should be _much_ faster than :method:`iter_ticks`. The ``start`` argument may take the same range of values as the ``start`` argument of :method:`iter_ticks`. The first full tick yielded will be the next full tick after the position obtained via `self.go_to_tick(start)`. The end tick may either be a tick value or a game state. The last full tick yielded will be the first full tick after the tick value/game state change. """ if start is not None: self.go_to_tick(start) for _ in self._stream.iterfullticks(): if end: if isinstance(end, basestring): if self.info.game_state == end: break else: if self.tick > end: break self._user_messages = [] self._game_events = [] yield self.tick
[docs] def go_to_tick(self, tick): """ Moves to the given tick, or the nearest tick after it. Returns the tick moved to. """ if isinstance(tick, str): return self.go_to_state_change(tick) if tick > self.demo.file_info.playback_ticks or tick < 0: raise IndexError("Tick {} out of range".format(tick)) self._stream = self.demo.stream(tick=tick) self._user_messages = (self._stream.user_messages or [])[:] self._game_events = (self._stream.game_events or [])[:] return self.tick
[docs] def go_to_time(self, time): """ Moves to the tick with the given game time. Could potentially overshoot, but not by too much. Will not undershoot. Returns the tick it has moved to. """ # Go for 31 tps as would rather exit FP loop earlier than # sooner. 1.5 for same reason FP_REGION = 1800 * 1.5 / 31 # If the time we're going to is behind us, recreate the stream if time < self.info.game_time: self._stream = self.demo.stream(tick=0) # If the time is more than 1.5 full packets ahead of us, iter full ticks if time > self.info.game_time + FP_REGION: for _ in self._stream.iterfullticks(): # If this full tick is within 1.5 full packets of the target, stop if self.info.game_time + FP_REGION > time and\ self.info.game_time < time: break else: # If we didn't stop via break, then it was EOF, so raise raise IndexError("Time {} out of range".format(time)) for _ in self._stream: if self.info.game_time > time: break else: raise IndexError("Time {} out of range".format(time)) self._user_messages = self._stream.user_messages[:] self._game_events = self._stream.game_events[:] return self.tick
[docs] def go_to_state_change(self, state): """ Moves to the time when the :attr:`GameInfo.game_state` changed to the given state. Valid values are equal to the possible values of :att:`~GameInfo.game_state`, along with ``"start"`` and ``"end"`` which signify the first and last tick in the replay, respectively. Returns the tick moved to. """ if state in self._state_change_ticks: return self.go_to_tick(self._state_change_ticks[state]) elif state in self._state_change_times: return self.go_to_time(self._state_change_times[state]) else: raise ValueError("Unsupported state {}".format(repr(state)))
def __iter__(self): return self.iter_ticks() @property
[docs] def players(self): """ A list of :class:`Player` objects, one for each player in the game. This excludes spectators and other non-hero-controlling players. """ from . import Player return sorted([p for p in Player.get_all(self) if p.index != None and p.team != "spectator"], key=lambda p:p.index)
@property
[docs] def info(self): """ The :class:`GameInfo` object for the replay. """ from .gameinfo import GameInfo info = GameInfo.get_all(self) assert len(info) == 1 return info[0]
@property
[docs] def creeps(self): """ The :class:`CreepManager` object for the replay. """ from .creeps import CreepManager return CreepManager(self)
@property
[docs] def buildings(self): """ The :class:`BuildingManager` object for the replay. """ from .buildings import BuildingManager return BuildingManager(self)
@staticmethod
[docs] def from_file(filename, *args, **kwargs): """ Loads the demo from the filename, and then initialises the :class:`StreamBinding` with it, along with any other passed arguments. """ import skadi.demo demo = skadi.demo.construct(filename) return StreamBinding(demo, *args, **kwargs)
Read the Docs v: latest
Versions
latest
Downloads
PDF
HTML
Epub
On Read the Docs
Project Home
Builds

Free document hosting provided by Read the Docs.