Combat Log¶

Normalizes combat-log data from both Source 1 and Source 2 ingestion paths into a unified entry shape.

`gem.combatlog.CombatLogProcessor` ¶

Parses and dispatches combat log entries.

Attributes:

Name	Type	Description
`_handlers`	`list[CombatLogHandler]`	Registered CombatLogHandler callables.

Source code in src/gem/combatlog.py

class CombatLogProcessor:
    """Parses and dispatches combat log entries.

    Attributes:
        _handlers: Registered CombatLogHandler callables.
    """

    def __init__(self) -> None:
        self._handlers: list[CombatLogHandler] = []

    def on_combat_log_entry(self, handler: CombatLogHandler) -> None:
        """Register a handler to receive decoded CombatLogEntry objects.

        Args:
            handler: Callable ``(CombatLogEntry) -> None``.
        """
        self._handlers.append(handler)

    def process_rune_pickup(self, player_slot: int, rune_type: int, tick: int = 0) -> None:
        """Emit a PICKUP_RUNE CombatLogEntry from a CDOTAUserMsg_ChatEvent.

        Args:
            player_slot: Player slot (0-9) from ChatEvent.playerid_1.
            rune_type: Rune type integer from ChatEvent.value.
            tick: Current game tick.
        """
        entry = CombatLogEntry(
            tick=tick,
            log_type="PICKUP_RUNE",
            value=player_slot,
            gold_reason=rune_type,
        )
        self._emit(entry)

    def _emit(self, entry: Any) -> None:
        """Dispatch an entry to all registered handlers.

        Args:
            entry: A CombatLogEntry (or any object, for testing).
        """
        for h in self._handlers:
            h(entry)

    def process_s1_event(self, game_event: Any, name_table: Any, tick: int = 0) -> None:
        """Parse a ``dota_combatlog`` S1 game event and emit a CombatLogEntry.

        Names are integer indices resolved via the CombatLogNames string table.

        Args:
            game_event: A ``GameEvent`` object with typed field accessors
                (``get_int32``, ``get_bool``).
            name_table: An object with an ``items`` dict mapping int index →
                ``(key_str, value_bytes)`` for name resolution.
            tick: Current game tick.
        """
        type_val, _ = game_event.get_int32(_S1_FIELD_TYPE)
        log_type = _LOG_TYPE_NAMES.get(type_val, "DAMAGE")

        attacker_idx, _ = game_event.get_int32(_S1_FIELD_ATTACKER)
        target_idx, _ = game_event.get_int32(_S1_FIELD_TARGET)
        inflictor_idx, _ = game_event.get_int32(_S1_FIELD_INFLICTOR)

        value, _ = game_event.get_int32(_S1_FIELD_VALUE)
        attacker_illusion, _ = game_event.get_bool(_S1_FIELD_ATTACKER_ILLUSION)
        target_illusion, _ = game_event.get_bool(_S1_FIELD_TARGET_ILLUSION)
        attacker_hero, _ = game_event.get_bool(_S1_FIELD_ATTACKER_HERO)
        target_hero, _ = game_event.get_bool(_S1_FIELD_TARGET_HERO)
        ability_level, _ = game_event.get_int32(_S1_FIELD_ABILITY_LEVEL)
        gold_reason, _ = game_event.get_int32(_S1_FIELD_GOLD_REASON)
        xp_reason, _ = game_event.get_int32(_S1_FIELD_XP_REASON)

        entry = CombatLogEntry(
            tick=tick,
            log_type=log_type,
            attacker_name=_resolve_name(name_table, attacker_idx),
            target_name=_resolve_name(name_table, target_idx),
            inflictor_name=_resolve_name(name_table, inflictor_idx),
            value=value,
            attacker_is_hero=attacker_hero,
            target_is_hero=target_hero,
            attacker_is_illusion=attacker_illusion,
            target_is_illusion=target_illusion,
            ability_level=ability_level,
            gold_reason=gold_reason,
            xp_reason=xp_reason,
        )
        self._emit(entry)

    def process_s2_bulk(self, msg: Any, name_table: Any, tick: int = 0) -> None:
        """Parse a CDOTAUserMsg_CombatLogBulkData and emit CombatLogEntry per entry.

        Args:
            msg: A ``CDOTAUserMsg_CombatLogBulkData`` protobuf message whose
                ``combat_entries`` field is a repeated ``CMsgDOTACombatLogEntry``.
            name_table: String table with ``items`` dict for name resolution.
            tick: Current game tick.
        """
        for entry_msg in msg.combat_entries:
            self.process_s2_entry(entry_msg, name_table, tick=tick)

    def process_s2_entry(self, msg: Any, name_table: Any, tick: int = 0) -> None:
        """Parse a CMsgDOTACombatLogEntry and emit a CombatLogEntry.

        Args:
            msg: A ``CMsgDOTACombatLogEntry``-like protobuf message with
                integer name indices and flag attributes.
            name_table: An object with an ``items`` dict mapping int index →
                ``(key_str, value_bytes)`` for name resolution, OR a legacy
                object with a ``get(index, default='')`` method.
            tick: Current game tick.
        """
        log_type = _LOG_TYPE_NAMES.get(msg.type, "DAMAGE")

        # Support both StringTable.items dict and legacy dict-like name_table
        if hasattr(name_table, "items") and isinstance(name_table.items, dict):
            attacker_name = _resolve_name(name_table, msg.attacker_name)
            target_name = _resolve_name(name_table, msg.target_name)
            inflictor_name = _resolve_name(name_table, msg.inflictor_name)
        else:
            attacker_name = name_table.get(msg.attacker_name, "")
            target_name = name_table.get(msg.target_name, "")
            inflictor_name = name_table.get(msg.inflictor_name, "")

        # For PURCHASE events, msg.value is a CombatLogNames index for the item name.
        # Reference: odota/Parse.java cle.getValueName() for DOTA_COMBATLOG_PURCHASE
        value_name = ""
        if log_type == "PURCHASE":
            if hasattr(name_table, "items") and isinstance(name_table.items, dict):
                value_name = _resolve_name(name_table, msg.value)
            elif hasattr(name_table, "get"):
                value_name = name_table.get(msg.value, "")

        # msg.value is proto uint32 but Dota encodes signed values (e.g. gold lost)
        # as two's complement. Reinterpret as signed int32.
        # Reference: clarity-examples/combatlog/Main.java — cle.getValue() < 0 check
        raw_value = msg.value
        value = raw_value if raw_value < 0x80000000 else raw_value - 0x100000000

        stun_duration = msg.stun_duration if msg.HasField("stun_duration") else 0.0
        damage_type = ""
        if log_type == "DAMAGE" and hasattr(msg, "damage_type"):
            damage_type = _DAMAGE_TYPE_NAMES.get(msg.damage_type, "")

        entry = CombatLogEntry(
            tick=tick,
            log_type=log_type,
            attacker_name=attacker_name,
            target_name=target_name,
            inflictor_name=inflictor_name,
            value=value,
            attacker_is_hero=msg.is_attacker_hero,
            target_is_hero=msg.is_target_hero,
            attacker_is_illusion=msg.is_attacker_illusion,
            target_is_illusion=msg.is_target_illusion,
            ability_level=msg.ability_level,
            gold_reason=msg.gold_reason,
            xp_reason=msg.xp_reason,
            value_name=value_name,
            damage_type=damage_type,
            stun_duration=stun_duration,
        )
        self._emit(entry)

`on_combat_log_entry(handler: CombatLogHandler) -> None` ¶

Register a handler to receive decoded CombatLogEntry objects.

Parameters:

Name	Type	Description	Default
`handler`	`CombatLogHandler`	Callable `(CombatLogEntry) -> None`.	required

Source code in src/gem/combatlog.py

def on_combat_log_entry(self, handler: CombatLogHandler) -> None:
    """Register a handler to receive decoded CombatLogEntry objects.

    Args:
        handler: Callable ``(CombatLogEntry) -> None``.
    """
    self._handlers.append(handler)

`process_rune_pickup(player_slot: int, rune_type: int, tick: int = 0) -> None` ¶

Emit a PICKUP_RUNE CombatLogEntry from a CDOTAUserMsg_ChatEvent.

Parameters:

Name	Type	Description	Default
`player_slot`	`int`	Player slot (0-9) from ChatEvent.playerid_1.	required
`rune_type`	`int`	Rune type integer from ChatEvent.value.	required
`tick`	`int`	Current game tick.	`0`

Source code in src/gem/combatlog.py

def process_rune_pickup(self, player_slot: int, rune_type: int, tick: int = 0) -> None:
    """Emit a PICKUP_RUNE CombatLogEntry from a CDOTAUserMsg_ChatEvent.

    Args:
        player_slot: Player slot (0-9) from ChatEvent.playerid_1.
        rune_type: Rune type integer from ChatEvent.value.
        tick: Current game tick.
    """
    entry = CombatLogEntry(
        tick=tick,
        log_type="PICKUP_RUNE",
        value=player_slot,
        gold_reason=rune_type,
    )
    self._emit(entry)

`process_s1_event(game_event: Any, name_table: Any, tick: int = 0) -> None` ¶

Parse a dota_combatlog S1 game event and emit a CombatLogEntry.

Names are integer indices resolved via the CombatLogNames string table.

Parameters:

Name	Type	Description	Default
`game_event`	`Any`	A `GameEvent` object with typed field accessors (`get_int32`, `get_bool`).	required
`name_table`	`Any`	An object with an `items` dict mapping int index → `(key_str, value_bytes)` for name resolution.	required
`tick`	`int`	Current game tick.	`0`

Source code in src/gem/combatlog.py

def process_s1_event(self, game_event: Any, name_table: Any, tick: int = 0) -> None:
    """Parse a ``dota_combatlog`` S1 game event and emit a CombatLogEntry.

    Names are integer indices resolved via the CombatLogNames string table.

    Args:
        game_event: A ``GameEvent`` object with typed field accessors
            (``get_int32``, ``get_bool``).
        name_table: An object with an ``items`` dict mapping int index →
            ``(key_str, value_bytes)`` for name resolution.
        tick: Current game tick.
    """
    type_val, _ = game_event.get_int32(_S1_FIELD_TYPE)
    log_type = _LOG_TYPE_NAMES.get(type_val, "DAMAGE")

    attacker_idx, _ = game_event.get_int32(_S1_FIELD_ATTACKER)
    target_idx, _ = game_event.get_int32(_S1_FIELD_TARGET)
    inflictor_idx, _ = game_event.get_int32(_S1_FIELD_INFLICTOR)

    value, _ = game_event.get_int32(_S1_FIELD_VALUE)
    attacker_illusion, _ = game_event.get_bool(_S1_FIELD_ATTACKER_ILLUSION)
    target_illusion, _ = game_event.get_bool(_S1_FIELD_TARGET_ILLUSION)
    attacker_hero, _ = game_event.get_bool(_S1_FIELD_ATTACKER_HERO)
    target_hero, _ = game_event.get_bool(_S1_FIELD_TARGET_HERO)
    ability_level, _ = game_event.get_int32(_S1_FIELD_ABILITY_LEVEL)
    gold_reason, _ = game_event.get_int32(_S1_FIELD_GOLD_REASON)
    xp_reason, _ = game_event.get_int32(_S1_FIELD_XP_REASON)

    entry = CombatLogEntry(
        tick=tick,
        log_type=log_type,
        attacker_name=_resolve_name(name_table, attacker_idx),
        target_name=_resolve_name(name_table, target_idx),
        inflictor_name=_resolve_name(name_table, inflictor_idx),
        value=value,
        attacker_is_hero=attacker_hero,
        target_is_hero=target_hero,
        attacker_is_illusion=attacker_illusion,
        target_is_illusion=target_illusion,
        ability_level=ability_level,
        gold_reason=gold_reason,
        xp_reason=xp_reason,
    )
    self._emit(entry)

`process_s2_bulk(msg: Any, name_table: Any, tick: int = 0) -> None` ¶

Parse a CDOTAUserMsg_CombatLogBulkData and emit CombatLogEntry per entry.

Parameters:

Name	Type	Description	Default
`msg`	`Any`	A `CDOTAUserMsg_CombatLogBulkData` protobuf message whose `combat_entries` field is a repeated `CMsgDOTACombatLogEntry`.	required
`name_table`	`Any`	String table with `items` dict for name resolution.	required
`tick`	`int`	Current game tick.	`0`

Source code in src/gem/combatlog.py

def process_s2_bulk(self, msg: Any, name_table: Any, tick: int = 0) -> None:
    """Parse a CDOTAUserMsg_CombatLogBulkData and emit CombatLogEntry per entry.

    Args:
        msg: A ``CDOTAUserMsg_CombatLogBulkData`` protobuf message whose
            ``combat_entries`` field is a repeated ``CMsgDOTACombatLogEntry``.
        name_table: String table with ``items`` dict for name resolution.
        tick: Current game tick.
    """
    for entry_msg in msg.combat_entries:
        self.process_s2_entry(entry_msg, name_table, tick=tick)

`process_s2_entry(msg: Any, name_table: Any, tick: int = 0) -> None` ¶

Parse a CMsgDOTACombatLogEntry and emit a CombatLogEntry.

Parameters:

Name	Type	Description	Default
`msg`	`Any`	A `CMsgDOTACombatLogEntry`-like protobuf message with integer name indices and flag attributes.	required
`name_table`	`Any`	An object with an `items` dict mapping int index → `(key_str, value_bytes)` for name resolution, OR a legacy object with a `get(index, default='')` method.	required
`tick`	`int`	Current game tick.	`0`

Source code in src/gem/combatlog.py

def process_s2_entry(self, msg: Any, name_table: Any, tick: int = 0) -> None:
    """Parse a CMsgDOTACombatLogEntry and emit a CombatLogEntry.

    Args:
        msg: A ``CMsgDOTACombatLogEntry``-like protobuf message with
            integer name indices and flag attributes.
        name_table: An object with an ``items`` dict mapping int index →
            ``(key_str, value_bytes)`` for name resolution, OR a legacy
            object with a ``get(index, default='')`` method.
        tick: Current game tick.
    """
    log_type = _LOG_TYPE_NAMES.get(msg.type, "DAMAGE")

    # Support both StringTable.items dict and legacy dict-like name_table
    if hasattr(name_table, "items") and isinstance(name_table.items, dict):
        attacker_name = _resolve_name(name_table, msg.attacker_name)
        target_name = _resolve_name(name_table, msg.target_name)
        inflictor_name = _resolve_name(name_table, msg.inflictor_name)
    else:
        attacker_name = name_table.get(msg.attacker_name, "")
        target_name = name_table.get(msg.target_name, "")
        inflictor_name = name_table.get(msg.inflictor_name, "")

    # For PURCHASE events, msg.value is a CombatLogNames index for the item name.
    # Reference: odota/Parse.java cle.getValueName() for DOTA_COMBATLOG_PURCHASE
    value_name = ""
    if log_type == "PURCHASE":
        if hasattr(name_table, "items") and isinstance(name_table.items, dict):
            value_name = _resolve_name(name_table, msg.value)
        elif hasattr(name_table, "get"):
            value_name = name_table.get(msg.value, "")

    # msg.value is proto uint32 but Dota encodes signed values (e.g. gold lost)
    # as two's complement. Reinterpret as signed int32.
    # Reference: clarity-examples/combatlog/Main.java — cle.getValue() < 0 check
    raw_value = msg.value
    value = raw_value if raw_value < 0x80000000 else raw_value - 0x100000000

    stun_duration = msg.stun_duration if msg.HasField("stun_duration") else 0.0
    damage_type = ""
    if log_type == "DAMAGE" and hasattr(msg, "damage_type"):
        damage_type = _DAMAGE_TYPE_NAMES.get(msg.damage_type, "")

    entry = CombatLogEntry(
        tick=tick,
        log_type=log_type,
        attacker_name=attacker_name,
        target_name=target_name,
        inflictor_name=inflictor_name,
        value=value,
        attacker_is_hero=msg.is_attacker_hero,
        target_is_hero=msg.is_target_hero,
        attacker_is_illusion=msg.is_attacker_illusion,
        target_is_illusion=msg.is_target_illusion,
        ability_level=msg.ability_level,
        gold_reason=msg.gold_reason,
        xp_reason=msg.xp_reason,
        value_name=value_name,
        damage_type=damage_type,
        stun_duration=stun_duration,
    )
    self._emit(entry)

`gem.combatlog.CombatLogEntry` `dataclass` ¶

One decoded combat log entry.

Attributes:

Name	Type	Description
`tick`	`int`	Game tick at which the event occurred.
`log_type`	`str`	String label from COMBAT_LOG_TYPES.
`attacker_name`	`str`	Name of the attacker unit/hero.
`target_name`	`str`	Name of the target unit/hero.
`inflictor_name`	`str`	Ability or item that caused the event.
`value`	`int`	Numeric value (damage, heal amount, gold, xp, etc.).
`attacker_is_hero`	`bool`	True if the attacker is a hero.
`target_is_hero`	`bool`	True if the target is a hero.
`attacker_is_illusion`	`bool`	True if the attacker is an illusion.
`target_is_illusion`	`bool`	True if the target is an illusion.
`ability_level`	`int`	Ability level (for ability/item events).
`gold_reason`	`int`	Gold reason code (for GOLD events).
`xp_reason`	`int`	XP reason code (for XP events).
`value_name`	`str`	Resolved name for the value field (PURCHASE events: item name).
`damage_type`	`str`	Damage type label for DAMAGE events ("physical", "magical", "pure").
`stun_duration`	`float`	Duration of stun applied by this event in seconds (S2 only; 0.0 if none).

Source code in src/gem/combatlog.py

@dataclass
class CombatLogEntry:
    """One decoded combat log entry.

    Attributes:
        tick: Game tick at which the event occurred.
        log_type: String label from COMBAT_LOG_TYPES.
        attacker_name: Name of the attacker unit/hero.
        target_name: Name of the target unit/hero.
        inflictor_name: Ability or item that caused the event.
        value: Numeric value (damage, heal amount, gold, xp, etc.).
        attacker_is_hero: True if the attacker is a hero.
        target_is_hero: True if the target is a hero.
        attacker_is_illusion: True if the attacker is an illusion.
        target_is_illusion: True if the target is an illusion.
        ability_level: Ability level (for ability/item events).
        gold_reason: Gold reason code (for GOLD events).
        xp_reason: XP reason code (for XP events).
        value_name: Resolved name for the value field (PURCHASE events: item name).
        damage_type: Damage type label for DAMAGE events ("physical", "magical", "pure").
        stun_duration: Duration of stun applied by this event in seconds (S2 only; 0.0 if none).
    """

    tick: int
    log_type: str
    attacker_name: str = ""
    target_name: str = ""
    inflictor_name: str = ""
    value: int = 0
    attacker_is_hero: bool = False
    target_is_hero: bool = False
    attacker_is_illusion: bool = False
    target_is_illusion: bool = False
    ability_level: int = 0
    gold_reason: int = 0
    xp_reason: int = 0
    value_name: str = ""
    damage_type: str = ""
    stun_duration: float = 0.0

Combat Log¶

gem.combatlog.CombatLogProcessor ¶

on_combat_log_entry(handler: CombatLogHandler) -> None ¶

process_rune_pickup(player_slot: int, rune_type: int, tick: int = 0) -> None ¶

process_s1_event(game_event: Any, name_table: Any, tick: int = 0) -> None ¶

process_s2_bulk(msg: Any, name_table: Any, tick: int = 0) -> None ¶

process_s2_entry(msg: Any, name_table: Any, tick: int = 0) -> None ¶

gem.combatlog.CombatLogEntry dataclass ¶

`gem.combatlog.CombatLogProcessor` ¶

`on_combat_log_entry(handler: CombatLogHandler) -> None` ¶

`process_rune_pickup(player_slot: int, rune_type: int, tick: int = 0) -> None` ¶

`process_s1_event(game_event: Any, name_table: Any, tick: int = 0) -> None` ¶

`process_s2_bulk(msg: Any, name_table: Any, tick: int = 0) -> None` ¶

`process_s2_entry(msg: Any, name_table: Any, tick: int = 0) -> None` ¶

`gem.combatlog.CombatLogEntry` `dataclass` ¶