Skip to content

Entities

Manages packet entity lifecycle (create/update/delete) and typed access to live networked field state.

See also: The Entity System, Entity State

gem.entities.EntityOp

Bases: IntFlag

Bitmask indicating what happened to an entity in a packet.

Source code in src/gem/entities.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
class EntityOp(enum.IntFlag):
    """Bitmask indicating what happened to an entity in a packet."""

    NONE = 0x00
    CREATED = 0x01
    UPDATED = 0x02
    DELETED = 0x04
    ENTERED = 0x08
    LEFT = 0x10

    # Convenience combinations
    CREATED_ENTERED = 0x01 | 0x08
    UPDATED_ENTERED = 0x02 | 0x08
    DELETED_LEFT = 0x04 | 0x10

    def has(self, other: EntityOp) -> bool:
        """Return True if this op includes *other*.

        Args:
            other: The flag to test for.

        Returns:
            True if the flag is set.
        """
        return bool(self & other)

has(other: EntityOp) -> bool

Return True if this op includes other.

Parameters:

Name Type Description Default
other EntityOp

The flag to test for.

 required

Returns:

Type Description
bool

True if the flag is set.
Source code in src/gem/entities.py 
65
66
67
68
69
70
71
72
73
74
def has(self, other: EntityOp) -> bool:
    """Return True if this op includes *other*.

    Args:
        other: The flag to test for.

    Returns:
        True if the flag is set.
    """
    return bool(self & other)

gem.entities.Entity

A live game entity with decoded field state.

The entity state is stored in _state (a FieldState tree for decoded replay fields) and optionally in a plain dict overlay for direct key-value access. The get() method queries the flat _state dict first (which tests may set directly), then falls back to FieldState path resolution.

Attributes:

Name Type Description
index

Entity slot index.
serial

Serial number for handle validation.
cls

Class metadata (has .name, .class_id, .serializer).
active

False while the entity is in the leave state.
Source code in src/gem/entities.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
class Entity:
    """A live game entity with decoded field state.

    The entity state is stored in ``_state`` (a FieldState tree for decoded
    replay fields) and optionally in a plain dict overlay for direct key-value
    access. The ``get()`` method queries the flat ``_state`` dict first (which
    tests may set directly), then falls back to FieldState path resolution.

    Attributes:
        index: Entity slot index.
        serial: Serial number for handle validation.
        cls: Class metadata (has .name, .class_id, .serializer).
        active: False while the entity is in the leave state.
    """

    __slots__ = (
        "index",
        "serial",
        "cls",
        "active",
        "_field_state",
        "_state",
        "_fp_cache",
        "_fp_noop",
    )

    def __init__(self, index: int, serial: int, cls: Any) -> None:
        self.index = index
        self.serial = serial
        self.cls = cls
        self.active = True
        self._field_state = FieldState()
        # Flat dict for direct key-value access (tests may write here directly)
        self._state: dict[str, Any] = {}
        self._fp_cache: dict[str, FieldPath] = {}
        self._fp_noop: set[str] = set()

    # ------------------------------------------------------------------
    # Field access
    # ------------------------------------------------------------------

    def get(self, name: str) -> Any:
        """Return the current value of *name*, or None if absent.

        Checks the flat ``_state`` dict first, then resolves through the
        FieldState tree using the serializer's field schema.

        Args:
            name: Dotted field name, e.g. ``"m_iHealth"``.

        Returns:
            The decoded value, or None.
        """
        # Fast path: direct dict (set by tests or flat lookups)
        if name in self._state:
            return self._state[name]

        # Slow path: resolve through serializer field paths
        serializer = getattr(self.cls, "serializer", None)
        if serializer is None:
            return None
        if name in self._fp_noop:
            return None
        if name in self._fp_cache:
            return self._field_state.get(self._fp_cache[name])

        fp = _find_field_path(serializer, name)
        if fp is None:
            self._fp_noop.add(name)
            return None

        self._fp_cache[name] = fp
        return self._field_state.get(fp)

    def exists(self, name: str) -> bool:
        """Return True if *name* has a value in the entity state.

        Args:
            name: Field name to check.
        """
        return self.get(name) is not None

    def get_int32(self, name: str) -> int | None:
        """Return the value as int32, or None if absent/wrong type.

        Args:
            name: Field name.

        Returns:
            Integer value, or None if the field is absent or not an int.
        """
        v = self.get(name)
        return v if isinstance(v, int) else None

    def get_uint32(self, name: str) -> int | None:
        """Return the value as uint32 (low 32 bits), or None if absent.

        Accepts both int and uint64 values (truncating to 32 bits if needed).

        Args:
            name: Field name.

        Returns:
            Integer value masked to 32 bits, or None if absent.
        """
        v = self.get(name)
        return (v & 0xFFFFFFFF) if isinstance(v, int) else None

    def get_uint64(self, name: str) -> int | None:
        """Return the value as uint64, or None if absent.

        Args:
            name: Field name.

        Returns:
            Integer value, or None if absent.
        """
        v = self.get(name)
        return v if isinstance(v, int) else None

    def get_float32(self, name: str) -> float | None:
        """Return the value as float32, or None if absent.

        Args:
            name: Field name.

        Returns:
            Float value, or None if absent or not numeric.
        """
        v = self.get(name)
        return float(v) if isinstance(v, (int, float)) else None

    def get_string(self, name: str) -> str | None:
        """Return the value as str, or None if absent.

        Args:
            name: Field name.

        Returns:
            String value, or None if absent or not a string.
        """
        v = self.get(name)
        return v if isinstance(v, str) else None

    def get_bool(self, name: str) -> bool | None:
        """Return the value as bool, or None if absent.

        Args:
            name: Field name.

        Returns:
            Boolean value, or None if absent or not a bool/int.
        """
        v = self.get(name)
        return bool(v) if isinstance(v, (bool, int)) else None

    def to_map(self) -> dict[str, Any]:
        """Return a snapshot of the flat _state dict.

        Returns:
            Dict of field name → value.
        """
        return dict(self._state)

    def get_class_name(self) -> str:
        """Return the entity class name."""
        return self.cls.name

    def get_class_id(self) -> int:
        """Return the entity class ID."""
        return self.cls.class_id

    def get_index(self) -> int:
        """Return the entity slot index."""
        return self.index

    def get_serial(self) -> int:
        """Return the entity serial number."""
        return self.serial

    def __repr__(self) -> str:
        return f"Entity({self.index}, {self.cls.name!r})"

get(name: str) -> Any

Return the current value of name, or None if absent.

Checks the flat _state dict first, then resolves through the FieldState tree using the serializer's field schema.

Parameters:

Name Type Description Default
name str

Dotted field name, e.g. "m_iHealth".

 required

Returns:

Type Description
Any

The decoded value, or None.
Source code in src/gem/entities.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def get(self, name: str) -> Any:
    """Return the current value of *name*, or None if absent.

    Checks the flat ``_state`` dict first, then resolves through the
    FieldState tree using the serializer's field schema.

    Args:
        name: Dotted field name, e.g. ``"m_iHealth"``.

    Returns:
        The decoded value, or None.
    """
    # Fast path: direct dict (set by tests or flat lookups)
    if name in self._state:
        return self._state[name]

    # Slow path: resolve through serializer field paths
    serializer = getattr(self.cls, "serializer", None)
    if serializer is None:
        return None
    if name in self._fp_noop:
        return None
    if name in self._fp_cache:
        return self._field_state.get(self._fp_cache[name])

    fp = _find_field_path(serializer, name)
    if fp is None:
        self._fp_noop.add(name)
        return None

    self._fp_cache[name] = fp
    return self._field_state.get(fp)

exists(name: str) -> bool

Return True if name has a value in the entity state.

Parameters:

Name Type Description Default
name str

Field name to check.

 required
Source code in src/gem/entities.py 
176
177
178
179
180
181
182
def exists(self, name: str) -> bool:
    """Return True if *name* has a value in the entity state.

    Args:
        name: Field name to check.
    """
    return self.get(name) is not None

get_int32(name: str) -> int | None

Return the value as int32, or None if absent/wrong type.

Parameters:

Name Type Description Default
name str

Field name.

 required

Returns:

Type Description
int | None

Integer value, or None if the field is absent or not an int.
Source code in src/gem/entities.py 
184
185
186
187
188
189
190
191
192
193
194
def get_int32(self, name: str) -> int | None:
    """Return the value as int32, or None if absent/wrong type.

    Args:
        name: Field name.

    Returns:
        Integer value, or None if the field is absent or not an int.
    """
    v = self.get(name)
    return v if isinstance(v, int) else None

get_uint32(name: str) -> int | None

Return the value as uint32 (low 32 bits), or None if absent.

Accepts both int and uint64 values (truncating to 32 bits if needed).

Parameters:

Name Type Description Default
name str

Field name.

 required

Returns:

Type Description
int | None

Integer value masked to 32 bits, or None if absent.
Source code in src/gem/entities.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
def get_uint32(self, name: str) -> int | None:
    """Return the value as uint32 (low 32 bits), or None if absent.

    Accepts both int and uint64 values (truncating to 32 bits if needed).

    Args:
        name: Field name.

    Returns:
        Integer value masked to 32 bits, or None if absent.
    """
    v = self.get(name)
    return (v & 0xFFFFFFFF) if isinstance(v, int) else None

get_uint64(name: str) -> int | None

Return the value as uint64, or None if absent.

Parameters:

Name Type Description Default
name str

Field name.

 required

Returns:

Type Description
int | None

Integer value, or None if absent.
Source code in src/gem/entities.py 
210
211
212
213
214
215
216
217
218
219
220
def get_uint64(self, name: str) -> int | None:
    """Return the value as uint64, or None if absent.

    Args:
        name: Field name.

    Returns:
        Integer value, or None if absent.
    """
    v = self.get(name)
    return v if isinstance(v, int) else None

get_float32(name: str) -> float | None

Return the value as float32, or None if absent.

Parameters:

Name Type Description Default
name str

Field name.

 required

Returns:

Type Description
float | None

Float value, or None if absent or not numeric.
Source code in src/gem/entities.py 
222
223
224
225
226
227
228
229
230
231
232
def get_float32(self, name: str) -> float | None:
    """Return the value as float32, or None if absent.

    Args:
        name: Field name.

    Returns:
        Float value, or None if absent or not numeric.
    """
    v = self.get(name)
    return float(v) if isinstance(v, (int, float)) else None

get_string(name: str) -> str | None

Return the value as str, or None if absent.

Parameters:

Name Type Description Default
name str

Field name.

 required

Returns:

Type Description
str | None

String value, or None if absent or not a string.
Source code in src/gem/entities.py 
234
235
236
237
238
239
240
241
242
243
244
def get_string(self, name: str) -> str | None:
    """Return the value as str, or None if absent.

    Args:
        name: Field name.

    Returns:
        String value, or None if absent or not a string.
    """
    v = self.get(name)
    return v if isinstance(v, str) else None

get_bool(name: str) -> bool | None

Return the value as bool, or None if absent.

Parameters:

Name Type Description Default
name str

Field name.

 required

Returns:

Type Description
bool | None

Boolean value, or None if absent or not a bool/int.
Source code in src/gem/entities.py 
246
247
248
249
250
251
252
253
254
255
256
def get_bool(self, name: str) -> bool | None:
    """Return the value as bool, or None if absent.

    Args:
        name: Field name.

    Returns:
        Boolean value, or None if absent or not a bool/int.
    """
    v = self.get(name)
    return bool(v) if isinstance(v, (bool, int)) else None

to_map() -> dict[str, Any]

Return a snapshot of the flat _state dict.

Returns:

Type Description
dict[str, Any]

Dict of field name → value.
Source code in src/gem/entities.py 
258
259
260
261
262
263
264
def to_map(self) -> dict[str, Any]:
    """Return a snapshot of the flat _state dict.

    Returns:
        Dict of field name → value.
    """
    return dict(self._state)

get_class_name() -> str

Return the entity class name.

Source code in src/gem/entities.py 
266
267
268
def get_class_name(self) -> str:
    """Return the entity class name."""
    return self.cls.name

get_class_id() -> int

Return the entity class ID.

Source code in src/gem/entities.py 
270
271
272
def get_class_id(self) -> int:
    """Return the entity class ID."""
    return self.cls.class_id

get_index() -> int

Return the entity slot index.

Source code in src/gem/entities.py 
274
275
276
def get_index(self) -> int:
    """Return the entity slot index."""
    return self.index

get_serial() -> int

Return the entity serial number.

Source code in src/gem/entities.py 
278
279
280
def get_serial(self) -> int:
    """Return the entity serial number."""
    return self.serial

gem.entities.EntityManager

Manages entity lifecycle across a replay stream.

Processes CSVCMsg_ServerInfo, CDemoClassInfo, CSVCMsg_CreateStringTable / CSVCMsg_UpdateStringTable side effects, and CSVCMsg_PacketEntities.

Attributes:

Name Type Description
entities list[Entity | None]

Sparse list indexed by entity slot (None = empty slot).
classes_by_id dict[int, ClassInfo]

class_id → ClassInfo.
classes_by_name dict[str, ClassInfo]

class_name → ClassInfo.
class_baselines dict[int, bytes]

class_id → baseline bytes.
serializers

Output of parse_send_tables.
string_tables

StringTables container (shared with caller).
game_build int

Server build number extracted from ServerInfo.
class_id_size int

Number of bits used to encode class IDs.
Source code in src/gem/entities.py 
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
class EntityManager:
    """Manages entity lifecycle across a replay stream.

    Processes ``CSVCMsg_ServerInfo``, ``CDemoClassInfo``,
    ``CSVCMsg_CreateStringTable`` / ``CSVCMsg_UpdateStringTable`` side effects,
    and ``CSVCMsg_PacketEntities``.

    Attributes:
        entities: Sparse list indexed by entity slot (None = empty slot).
        classes_by_id: class_id → ClassInfo.
        classes_by_name: class_name → ClassInfo.
        class_baselines: class_id → baseline bytes.
        serializers: Output of ``parse_send_tables``.
        string_tables: StringTables container (shared with caller).
        game_build: Server build number extracted from ServerInfo.
        class_id_size: Number of bits used to encode class IDs.
    """

    def __init__(self, serializers: dict[str, Serializer], string_tables: StringTables) -> None:
        self.serializers = serializers
        self.string_tables = string_tables
        self.entities: list[Entity | None] = []
        self.classes_by_id: dict[int, ClassInfo] = {}
        self.classes_by_name: dict[str, ClassInfo] = {}
        self.class_baselines: dict[int, bytes] = {}
        self.game_build: int = 0
        self.class_id_size: int = 0
        self._class_info_ready: bool = False
        self._full_packets: int = 0
        self.tracker = EntityTracker()

    def on_entity(self, handler: EntityHandler) -> None:
        """Register an entity event handler.

        Args:
            handler: Callable ``(Entity, EntityOp) -> None``.
        """
        self.tracker.on_entity(handler)

    # ------------------------------------------------------------------
    # Handlers
    # ------------------------------------------------------------------

    def on_server_info(self, msg: object) -> None:
        """Extract classIdSize and game build from CSVCMsg_ServerInfo.

        Args:
            msg: A ``CSVCMsg_ServerInfo`` protobuf message.
        """
        max_classes: int = msg.max_classes  # type: ignore[attr-defined]
        self.class_id_size = int(math.log2(max_classes)) + 1
        # Entity slot indices are drawn from the full Source 2 entity pool
        # (up to 2^14 = 16384 slots), not bounded by the number of class types.
        # Using max_classes here causes IndexError when a slot index > max_classes
        # is assigned in on_packet_entities.
        self.entities = [None] * (1 << 14)

        game_dir: str = msg.game_dir  # type: ignore[attr-defined]
        m = _GAME_BUILD_RE.search(game_dir)
        if m:
            self.game_build = int(m.group(1))

    def on_class_info(self, msg: object) -> None:
        """Build class maps from CDemoClassInfo.

        Args:
            msg: A ``CDemoClassInfo`` protobuf message.
        """
        for c in msg.classes:  # type: ignore[attr-defined]
            class_id: int = c.class_id
            net_name: str = c.network_name
            ser = self.serializers.get(net_name)
            ci = ClassInfo(class_id=class_id, name=net_name, serializer=ser)
            self.classes_by_id[class_id] = ci
            self.classes_by_name[net_name] = ci

        self._class_info_ready = True
        self._update_baselines()

    def on_baseline_updated(self) -> None:
        """Call after instancebaseline string table is created or updated."""
        self._update_baselines()

    def on_packet_entities(self, msg: object) -> list[tuple[Entity, EntityOp]]:
        """Decode a CSVCMsg_PacketEntities message.

        Args:
            msg: A ``CSVCMsg_PacketEntities`` protobuf message.

        Returns:
            List of (Entity, EntityOp) tuples in the order they were processed.
        """
        is_delta: bool = msg.legacy_is_delta  # type: ignore[attr-defined]
        updates: int = msg.updated_entries  # type: ignore[attr-defined]
        entity_data: bytes = msg.entity_data  # type: ignore[attr-defined]

        if not is_delta:
            if self._full_packets > 0:
                return []
            self._full_packets += 1

        r = BitReader(entity_data)
        index = -1
        results: list[tuple[Entity, EntityOp]] = []

        for _ in range(updates):
            index += r.read_ubit_var() + 1
            cmd = r.read_bits(2)

            if cmd & 0x01 == 0:
                if cmd & 0x02:
                    # Create entity
                    class_id = r.read_bits(self.class_id_size)
                    serial = r.read_bits(17)
                    r.read_varuint32()  # unknown / padding

                    ci = self.classes_by_id.get(class_id)
                    if ci is None:
                        raise RuntimeError(f"unknown class id {class_id}")

                    entity = Entity(index=index, serial=serial, cls=ci)
                    self.entities[index] = entity

                    # Apply baseline first, then delta
                    baseline = self.class_baselines.get(class_id)
                    if baseline and ci.serializer is not None:
                        read_fields(BitReader(baseline), ci.serializer, entity._field_state)
                    if ci.serializer is not None:
                        read_fields(r, ci.serializer, entity._field_state)

                    op = EntityOp.CREATED | EntityOp.ENTERED
                else:
                    # Update entity
                    _e = self.entities[index]
                    if _e is None:
                        raise RuntimeError(f"update on missing entity {index}")
                    entity = _e
                    op = EntityOp.UPDATED
                    if not entity.active:
                        entity.active = True
                        op |= EntityOp.ENTERED
                    if entity.cls.serializer is not None:
                        read_fields(r, entity.cls.serializer, entity._field_state)
            else:
                # Leave / delete
                _e = self.entities[index]
                if _e is None:
                    raise RuntimeError(f"leave on missing entity {index}")
                entity = _e
                op = EntityOp.LEFT
                if cmd & 0x02:
                    op |= EntityOp.DELETED
                    self.entities[index] = None
                else:
                    entity.active = False

            self.tracker._dispatch(entity, op)
            results.append((entity, op))

        return results

    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------

    def _update_baselines(self) -> None:
        if not self._class_info_ready:
            return
        table = self.string_tables.get_by_name("instancebaseline")
        if table is None:
            return
        for _idx, (key, value) in table.items.items():
            try:
                class_id = int(key)
            except (ValueError, AttributeError):
                continue
            self.class_baselines[class_id] = value

    # ------------------------------------------------------------------
    # Query helpers
    # ------------------------------------------------------------------

    def find(self, index: int) -> Entity | None:
        """Return the entity at the given slot index, or None.

        Args:
            index: Entity slot index.
        """
        if 0 <= index < len(self.entities):
            return self.entities[index]
        return None

    def find_by_handle(self, handle: int) -> Entity | None:
        """Return the entity for a Source 2 entity handle, or None.

        Args:
            handle: 32-bit entity handle (index in low 14 bits, serial in high bits).
        """
        idx = handle & _HANDLE_MASK
        serial = handle >> _INDEX_BITS
        entity = self.find(idx)
        if entity is not None and entity.serial == serial:
            return entity
        return None

    def filter(self, predicate: Any) -> list[Entity]:
        """Return all entities matching a predicate.

        Args:
            predicate: Callable ``(Entity) -> bool``.

        Returns:
            List of matching Entity objects.
        """
        return [e for e in self.entities if e is not None and predicate(e)]

    def find_by_class_name(self, class_name: str) -> Entity | None:
        """Return the first active entity whose class name matches, or None.

        Args:
            class_name: Entity class name, e.g. ``"CDOTAGamerulesProxy"``.
        """
        for e in self.entities:
            if e is not None and e.active and e.get_class_name() == class_name:
                return e
        return None

    def all_active(self) -> list[Entity]:
        """Return all currently active entities."""
        return [e for e in self.entities if e is not None and e.active]

on_entity(handler: EntityHandler) -> None

Register an entity event handler.

Parameters:

Name Type Description Default
handler EntityHandler

Callable (Entity, EntityOp) -> None.

 required
Source code in src/gem/entities.py 
413
414
415
416
417
418
419
def on_entity(self, handler: EntityHandler) -> None:
    """Register an entity event handler.

    Args:
        handler: Callable ``(Entity, EntityOp) -> None``.
    """
    self.tracker.on_entity(handler)

on_server_info(msg: object) -> None

Extract classIdSize and game build from CSVCMsg_ServerInfo.

Parameters:

Name Type Description Default
msg object

A CSVCMsg_ServerInfo protobuf message.

 required
Source code in src/gem/entities.py 
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
def on_server_info(self, msg: object) -> None:
    """Extract classIdSize and game build from CSVCMsg_ServerInfo.

    Args:
        msg: A ``CSVCMsg_ServerInfo`` protobuf message.
    """
    max_classes: int = msg.max_classes  # type: ignore[attr-defined]
    self.class_id_size = int(math.log2(max_classes)) + 1
    # Entity slot indices are drawn from the full Source 2 entity pool
    # (up to 2^14 = 16384 slots), not bounded by the number of class types.
    # Using max_classes here causes IndexError when a slot index > max_classes
    # is assigned in on_packet_entities.
    self.entities = [None] * (1 << 14)

    game_dir: str = msg.game_dir  # type: ignore[attr-defined]
    m = _GAME_BUILD_RE.search(game_dir)
    if m:
        self.game_build = int(m.group(1))

on_class_info(msg: object) -> None

Build class maps from CDemoClassInfo.

Parameters:

Name Type Description Default
msg object

A CDemoClassInfo protobuf message.

 required
Source code in src/gem/entities.py 
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
def on_class_info(self, msg: object) -> None:
    """Build class maps from CDemoClassInfo.

    Args:
        msg: A ``CDemoClassInfo`` protobuf message.
    """
    for c in msg.classes:  # type: ignore[attr-defined]
        class_id: int = c.class_id
        net_name: str = c.network_name
        ser = self.serializers.get(net_name)
        ci = ClassInfo(class_id=class_id, name=net_name, serializer=ser)
        self.classes_by_id[class_id] = ci
        self.classes_by_name[net_name] = ci

    self._class_info_ready = True
    self._update_baselines()

on_baseline_updated() -> None

Call after instancebaseline string table is created or updated.

Source code in src/gem/entities.py 
461
462
463
def on_baseline_updated(self) -> None:
    """Call after instancebaseline string table is created or updated."""
    self._update_baselines()

on_packet_entities(msg: object) -> list[tuple[Entity, EntityOp]]

Decode a CSVCMsg_PacketEntities message.

Parameters:

Name Type Description Default
msg object

A CSVCMsg_PacketEntities protobuf message.

 required

Returns:

Type Description
list[tuple[Entity, EntityOp]]

List of (Entity, EntityOp) tuples in the order they were processed.
Source code in src/gem/entities.py 
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
def on_packet_entities(self, msg: object) -> list[tuple[Entity, EntityOp]]:
    """Decode a CSVCMsg_PacketEntities message.

    Args:
        msg: A ``CSVCMsg_PacketEntities`` protobuf message.

    Returns:
        List of (Entity, EntityOp) tuples in the order they were processed.
    """
    is_delta: bool = msg.legacy_is_delta  # type: ignore[attr-defined]
    updates: int = msg.updated_entries  # type: ignore[attr-defined]
    entity_data: bytes = msg.entity_data  # type: ignore[attr-defined]

    if not is_delta:
        if self._full_packets > 0:
            return []
        self._full_packets += 1

    r = BitReader(entity_data)
    index = -1
    results: list[tuple[Entity, EntityOp]] = []

    for _ in range(updates):
        index += r.read_ubit_var() + 1
        cmd = r.read_bits(2)

        if cmd & 0x01 == 0:
            if cmd & 0x02:
                # Create entity
                class_id = r.read_bits(self.class_id_size)
                serial = r.read_bits(17)
                r.read_varuint32()  # unknown / padding

                ci = self.classes_by_id.get(class_id)
                if ci is None:
                    raise RuntimeError(f"unknown class id {class_id}")

                entity = Entity(index=index, serial=serial, cls=ci)
                self.entities[index] = entity

                # Apply baseline first, then delta
                baseline = self.class_baselines.get(class_id)
                if baseline and ci.serializer is not None:
                    read_fields(BitReader(baseline), ci.serializer, entity._field_state)
                if ci.serializer is not None:
                    read_fields(r, ci.serializer, entity._field_state)

                op = EntityOp.CREATED | EntityOp.ENTERED
            else:
                # Update entity
                _e = self.entities[index]
                if _e is None:
                    raise RuntimeError(f"update on missing entity {index}")
                entity = _e
                op = EntityOp.UPDATED
                if not entity.active:
                    entity.active = True
                    op |= EntityOp.ENTERED
                if entity.cls.serializer is not None:
                    read_fields(r, entity.cls.serializer, entity._field_state)
        else:
            # Leave / delete
            _e = self.entities[index]
            if _e is None:
                raise RuntimeError(f"leave on missing entity {index}")
            entity = _e
            op = EntityOp.LEFT
            if cmd & 0x02:
                op |= EntityOp.DELETED
                self.entities[index] = None
            else:
                entity.active = False

        self.tracker._dispatch(entity, op)
        results.append((entity, op))

    return results

find(index: int) -> Entity | None

Return the entity at the given slot index, or None.

Parameters:

Name Type Description Default
index int

Entity slot index.

 required
Source code in src/gem/entities.py 
564
565
566
567
568
569
570
571
572
def find(self, index: int) -> Entity | None:
    """Return the entity at the given slot index, or None.

    Args:
        index: Entity slot index.
    """
    if 0 <= index < len(self.entities):
        return self.entities[index]
    return None

find_by_handle(handle: int) -> Entity | None

Return the entity for a Source 2 entity handle, or None.

Parameters:

Name Type Description Default
handle int

32-bit entity handle (index in low 14 bits, serial in high bits).

 required
Source code in src/gem/entities.py 
574
575
576
577
578
579
580
581
582
583
584
585
def find_by_handle(self, handle: int) -> Entity | None:
    """Return the entity for a Source 2 entity handle, or None.

    Args:
        handle: 32-bit entity handle (index in low 14 bits, serial in high bits).
    """
    idx = handle & _HANDLE_MASK
    serial = handle >> _INDEX_BITS
    entity = self.find(idx)
    if entity is not None and entity.serial == serial:
        return entity
    return None

filter(predicate: Any) -> list[Entity]

Return all entities matching a predicate.

Parameters:

Name Type Description Default
predicate Any

Callable (Entity) -> bool.

 required

Returns:

Type Description
list[Entity]

List of matching Entity objects.
Source code in src/gem/entities.py 
587
588
589
590
591
592
593
594
595
596
def filter(self, predicate: Any) -> list[Entity]:
    """Return all entities matching a predicate.

    Args:
        predicate: Callable ``(Entity) -> bool``.

    Returns:
        List of matching Entity objects.
    """
    return [e for e in self.entities if e is not None and predicate(e)]

find_by_class_name(class_name: str) -> Entity | None

Return the first active entity whose class name matches, or None.

Parameters:

Name Type Description Default
class_name str

Entity class name, e.g. "CDOTAGamerulesProxy".

 required
Source code in src/gem/entities.py 
598
599
600
601
602
603
604
605
606
607
def find_by_class_name(self, class_name: str) -> Entity | None:
    """Return the first active entity whose class name matches, or None.

    Args:
        class_name: Entity class name, e.g. ``"CDOTAGamerulesProxy"``.
    """
    for e in self.entities:
        if e is not None and e.active and e.get_class_name() == class_name:
            return e
    return None

all_active() -> list[Entity]

Return all currently active entities.

Source code in src/gem/entities.py 
609
610
611
def all_active(self) -> list[Entity]:
    """Return all currently active entities."""
    return [e for e in self.entities if e is not None and e.active]