RLStatsAPI

https://manucabral.github.io/rlstatsapi/
Admin
A Python client library for reading live Rocket League Stats API events over a local TCP socket.
Tokens:22,182
Snippets:238
Trust Score:9.7
Update:4 months ago
Show doc for...
Tokens:
Raw
### CLI quick checks

Source: https://manucabral.github.io/rlstatsapi/quickstart

Provides examples of using the rlstatsapi command-line interface for quick checks.

```bash
rlstatsapi status
```

```bash
rlstatsapi listen --event GoalScored
```

--------------------------------

### Verify Installation

Source: https://manucabral.github.io/rlstatsapi/installation

Verify the installation by importing the package and printing its version.

```python
python -c "import rlstatsapi; print(rlstatsapi.__version__)"

```

--------------------------------

### Install from GitHub

Source: https://manucabral.github.io/rlstatsapi/installation

Install the rlstatsapi package directly from its GitHub repository using pip.

```bash
pip install git+https://github.com/manucabral/RocketLeagueStatsAPI.git

```

--------------------------------

### Install from PyPI

Source: https://manucabral.github.io/rlstatsapi/installation

Install the rlstatsapi package using pip from the Python Package Index.

```bash
pip install rlstatsapi

```

--------------------------------

### Basic listener

Source: https://manucabral.github.io/rlstatsapi/quickstart

This code snippet demonstrates how to set up a basic listener to print all incoming events from the local Rocket League exporter.

```python
import asyncio
from rlstatsapi import StatsClient


async def main() -> None:
    async with StatsClient() as client:
        client.on_any(lambda msg: print(msg.event, msg.data))
        await asyncio.Event().wait()


asyncio.run(main())
```

--------------------------------

### Run event_names_only.py

Source: https://manucabral.github.io/rlstatsapi/examples

This snippet shows how to run the `event_names_only.py` script from the repository root.

```bash
python examples/event_names_only.py

```

--------------------------------

### Filter specific events

Source: https://manucabral.github.io/rlstatsapi/quickstart

This snippet shows how to filter and listen for specific events like 'GoalScored' and 'MatchEnded', and how to register a single handler for multiple events.

```python
async with StatsClient() as client:
    async for message in client.events("GoalScored", "MatchEnded"):
        print(message.event, message.data)
```

```python
client.on_many(["MatchCreated", "MatchEnded"], lambda msg: print(msg.event))
```

--------------------------------

### Typed payload pattern (Pylance-friendly)

Source: https://manucabral.github.io/rlstatsapi/quickstart

Demonstrates a Pylance-friendly pattern for handling typed payloads, using `cast_event_data` to narrow down to event-specific payload types.

```python
from rlstatsapi.models import EventMessage
from rlstatsapi.types import GoalScoredPayload, cast_event_data


def on_goal(msg: EventMessage) -> None:
    data: GoalScoredPayload = cast_event_data("GoalScored", msg.data)
    scorer = data.get("Scorer", {})
    print(scorer.get("Name"))
```

--------------------------------

### connect()

Source: https://manucabral.github.io/rlstatsapi/api

Start the background reader task and begin connecting. No-op if already running.

```python
async def connect(self) -> None:
    """Start the background reader task and begin connecting. No-op if already running."""
    if self._reader_task and not self._reader_task.done():
        self._logger.debug("connect() ignored: reader already running")
        return

    self._stopping = False
    self._permanently_failed = False
    self._last_error = None
    self._reader_task = asyncio.create_task(self._run(), name="rlstatsapi-reader")
    self._logger.debug("reader task started")

```

--------------------------------

### Python Usage Example

Source: https://manucabral.github.io/rlstatsapi/events

Example of how to use the rlstatsapi library to handle a 'GoalScored' event.

```python
from rlstatsapi.models import EventMessage
from rlstatsapi.types import GoalScoredPayload, cast_event_data


def on_goal(msg: EventMessage) -> None:
    data: GoalScoredPayload = cast_event_data("GoalScored", msg.data)
    scorer = data.get("Scorer", {})
    print(f"{scorer.get('Name')}" scored!)
```

--------------------------------

### Detect match start and end

Source: https://manucabral.github.io/rlstatsapi/recipes

This snippet demonstrates how to listen for multiple events related to match lifecycle.

```python
import asyncio
from rlstatsapi import StatsClient


async def main() -> None:
    async with StatsClient() as client:
        client.on_many(
            ["MatchCreated", "RoundStarted", "MatchEnded", "MatchDestroyed"],
            lambda msg: print(msg.event),
        )
        await asyncio.Event().wait()

```

--------------------------------

### MatchInitialized Event Structure

Source: https://manucabral.github.io/rlstatsapi/events

This event is fired when the match finishes loading and is ready to start.

```json
{
  "event": "MatchInitialized",
  "data": {
    "MatchGuid": "A1B2C3D4E5F6"
  }
}
```

--------------------------------

### CountdownBegin Event Structure

Source: https://manucabral.github.io/rlstatsapi/events

This event is fired when the pre-kickoff countdown starts.

```json
{
  "event": "CountdownBegin",
  "data": {
    "MatchGuid": "A1B2C3D4E5F6"
  }
}
```

--------------------------------

### StatsClient Connect Method

Source: https://manucabral.github.io/rlstatsapi/api

Initiates the connection process to the Stats API and starts the background reader task.

```python
async def connect(self) -> None:
    """Start the background reader task and begin connecting. No-op if already running."""
    if self._reader_task and not self._reader_task.done():
        self._logger.debug("connect() ignored: reader already running")
        return

    self._stopping = False
    self._permanently_failed = False
    self._last_error = None
    self._reader_task = asyncio.create_task(self._run(), name="rlstatsapi-reader")
    self._logger.debug("reader task started")
```

--------------------------------

### StatsClient Lifecycle Callbacks

Source: https://manucabral.github.io/rlstatsapi/api

Example of how StatsClient executes connection lifecycle callbacks, awaiting asynchronous ones.

```python
    async def _fire_simple(self, handlers: list[_SimpleHandler]) -> None:
        """Run connection lifecycle callbacks and await async ones when needed."""
        for handler in list(handlers):
            try:
                result = handler()
                if inspect.isawaitable(result):
                    await result
            except Exception as exc:
                self._logger.error("lifecycle handler error: %s", exc)
```

--------------------------------

### StatsClient Handler Logic

Source: https://manucabral.github.io/rlstatsapi/api

Example of how StatsClient handles incoming messages, including event routing, handler execution, and error handling for both synchronous and asynchronous handlers.

```python
        for handler in handlers:
            try:
                result = handler(message)
                if inspect.isawaitable(result):
                    if (
                        self.handler_timeout is not None
                        and inspect.iscoroutinefunction(handler)
                    ):
                        await asyncio.wait_for(result, timeout=self.handler_timeout)
                    else:
                        await result
            except Exception as exc:
                self._metrics.handler_errors += 1
                if self._error_handlers:
                    for error_handler in list(self._error_handlers):
                        try:
                            r = error_handler(message, exc, handler)
                            if inspect.isawaitable(r):
                                await r
                        except Exception as err_exc:
                            self._logger.error(
                                "error handler itself raised: %s", err_exc
                            )
                else:
                    self._logger.error("handler error for %s: %s", message.event, exc)
```

--------------------------------

### ReplayCreated Event Structure

Source: https://manucabral.github.io/rlstatsapi/events

Fired when an in-match replay starts (e.g. goal replay saved as replay).

```json
{
  "event": "ReplayCreated",
  "data": {
    "MatchGuid": "A1B2C3D4E5F6"
  }
}
```

--------------------------------

### on_podium_start

Source: https://manucabral.github.io/rlstatsapi/api

Typed helper for registering PodiumStart handlers.

```python
def on_podium_start(
    self,
    handler: Callable[
        [TypedEventMessage[PodiumStartPayload]], Awaitable[None] | None
    ],
) -> None:
    """Typed helper for registering PodiumStart handlers."""
    self._handlers_by_event["PodiumStart"].append(handler)
```

--------------------------------

### Client Initialization (`__init__`)

Source: https://manucabral.github.io/rlstatsapi/api

Configures the client with various parameters for connection, reconnection, queueing, and timeouts. The client does not connect until `connect()` is called or the async context manager is entered.

```python
def __init__(
    self,
    host: str = "127.0.0.1",
    port: int = 49123,
    reconnect: bool = True,
    reconnect_delay: float = 0.5,
    max_reconnect_delay: float = 30.0,
    max_reconnect_attempts: int | None = None,
    include_raw: bool = False,
    queue_size: int = 2048,
    overflow: Literal["block", "drop", "raise"] = "block",
    connect_timeout: float = 5.0,
    drain_on_disconnect: bool = False,
    handler_timeout: float | None = None,
) -> None:
    """Configure the client.

    Does not connect until ``connect()`` or async context manager entry.

    Args:
        host: Stats API host address.
        port: Stats API TCP port.
        reconnect: Whether to reconnect automatically on connection loss.
        reconnect_delay: Initial delay in seconds before the first reconnect attempt.
        max_reconnect_delay: Upper bound for exponential backoff delay.
        max_reconnect_attempts: Stop reconnecting after this many failures.
            None means unlimited.
        include_raw: Attach the original JSON string to each ``EventMessage.raw``.
        queue_size: Max events buffered in the internal queue.
        overflow: Queue-full behavior ``"block"`` waits, ``"drop"`` discards,
            ``"raise"`` kills the connection.
        connect_timeout: Seconds to wait for TCP handshake before raising.
        drain_on_disconnect: Clear the queue when the session ends.
        handler_timeout: Max seconds an async handler may run before being cancelled.
            Sync handlers are not affected. None disables the timeout.
    """
    if reconnect_delay <= 0:
        raise ValueError("reconnect_delay must be positive")
    self.host = host
    self.port = port
    self.reconnect = reconnect
    self.reconnect_delay = reconnect_delay
    self.max_reconnect_delay = max_reconnect_delay
    self.max_reconnect_attempts = max_reconnect_attempts
    self.include_raw = include_raw
    self.connect_timeout = connect_timeout
    self.overflow = overflow
    self.drain_on_disconnect = drain_on_disconnect
    self.handler_timeout = handler_timeout

    self._queue: asyncio.Queue[EventMessage] = asyncio.Queue(maxsize=queue_size)
    self._handlers_by_event: dict[str, list[_AnyCallable]] = defaultdict(list)
    self._handlers_any: list[_AnyCallable] = []
    self._on_connect_handlers: list[_SimpleHandler] = []
    self._on_disconnect_handlers: list[_SimpleHandler] = []
    self._error_handlers: list[_ErrorHandler] = []
    self._logger = logging.getLogger("rlstatsapi")

    self._reader: asyncio.StreamReader | None = None
    self._writer: asyncio.StreamWriter | None = None
    self._reader_task: asyncio.Task[None] | None = None
    self._stopping = False
    self._connection_state = ConnectionState.DISCONNECTED
    self._permanently_failed = False
    self._last_error: Exception | None = None
    self._metrics = ClientMetrics()
    self._state_tracker = MatchStateTracker()

```

--------------------------------

### on_goal_replay_start handler registration

Source: https://manucabral.github.io/rlstatsapi/api

Typed helper for registering GoalReplayStart handlers.

```python
def on_goal_replay_start(
    self,
    handler: Callable[
        [TypedEventMessage[GoalReplayStartPayload]], Awaitable[None] | None
    ],
) -> None:
    """Typed helper for registering GoalReplayStart handlers."""
    self._handlers_by_event["GoalReplayStart"].append(handler)
```

--------------------------------

### StatsClient Initialization

Source: https://manucabral.github.io/rlstatsapi/api

Initializes the StatsClient with various configuration options.

```python
class StatsClient:
    """Event client for Rocket League Stats API."""

    def __init__(
        self,
        host: str = "127.0.0.1",
        port: int = 49123,
        reconnect: bool = True,
        reconnect_delay: float = 0.5,
        max_reconnect_delay: float = 30.0,
        max_reconnect_attempts: int | None = None,
        include_raw: bool = False,
        queue_size: int = 2048,
        overflow: Literal["block", "drop", "raise"] = "block",
        connect_timeout: float = 5.0,
        drain_on_disconnect: bool = False,
        handler_timeout: float | None = None,
    ) -> None:
        """Configure the client.

        Does not connect until ``connect()`` or async context manager entry.

        Args:
            host: Stats API host address.
            port: Stats API TCP port.
            reconnect: Whether to reconnect automatically on connection loss.
            reconnect_delay: Initial delay in seconds before the first reconnect attempt.
            max_reconnect_delay: Upper bound for exponential backoff delay.
            max_reconnect_attempts: Stop reconnecting after this many failures.
                None means unlimited.
            include_raw: Attach the original JSON string to each ``EventMessage.raw``.
            queue_size: Max events buffered in the internal queue.
            overflow: Queue-full behavior ``"block"`` waits, ``"drop"`` discards,

```

--------------------------------

### Keep your client port in sync with the config file

Source: https://manucabral.github.io/rlstatsapi/recipes

This snippet shows how to configure the StatsClient and keep its port in sync with the configuration.

```python
from rlstatsapi import StatsClient, configure_stats_api

status = configure_stats_api(enabled=True, port=49123, packet_send_rate=30)
client = StatsClient(port=status.port or 49123)

```

--------------------------------

### on method implementation

Source: https://manucabral.github.io/rlstatsapi/api

The actual implementation of the on method in the client.py file.

```python
def on(self, event_name: str, handler: _AnyCallable | None = None) -> Any:
    """Register a handler for an event or return a decorator form of registration."""
    if handler is None:

        def decorator(h: _AnyCallable) -> _AnyCallable:
            @functools.wraps(h)
            def wrapper(*args: Any, **kwargs: Any) -> Any:
                return h(*args, **kwargs)

            self._handlers_by_event[event_name].append(wrapper)
            return wrapper

        return decorator
    self._handlers_by_event[event_name].append(handler)
    return None
```

--------------------------------

### GoalReplayStart Event Structure

Source: https://manucabral.github.io/rlstatsapi/events

This event is fired when the goal replay begins.

```json
{
  "event": "GoalReplayStart",
  "data": {
    "MatchGuid": "A1B2C3D4E5F6"
  }
}
```

--------------------------------

### StatsClient Initialization

Source: https://manucabral.github.io/rlstatsapi/api

Initializes the StatsClient with various configuration options for connecting to the Stats API.

```python
def __init__(
    self,
    host: str,
    port: int,
    queue_size: int = 1000,
    reconnect: bool = True,
    reconnect_delay: float = 5.0,
    max_reconnect_delay: float = 60.0,
    max_reconnect_attempts: int = 10,
    include_raw: bool = False,
    connect_timeout: float | None = 10.0,
    overflow: str = "drop",
    drain_on_disconnect: bool = True,
    handler_timeout: float | None = None,
):
    """
    Connects to the Stats API at the given host and port.

    Args:
        host: Hostname or IP address of the Stats API server.
        port: Port number of the Stats API server.
        queue_size: Max number of events to buffer.
        reconnect: Whether to attempt reconnecting on disconnect.
        reconnect_delay: Seconds to wait before attempting a reconnect.
        max_reconnect_delay: Max seconds to wait between reconnect attempts.
        max_reconnect_attempts: Max number of reconnect attempts before giving up.
        include_raw: Whether to include raw event data in messages.
        connect_timeout: Seconds to wait for TCP handshake before raising.
        overflow: Action to take when the queue is full ("drop" or "raise").
        drain_on_disconnect: Clear the queue when the session ends.
        handler_timeout: Max seconds an async handler may run before being cancelled.
            Sync handlers are not affected. None disables the timeout.
    """
    if reconnect_delay <= 0:
        raise ValueError("reconnect_delay must be positive")
    self.host = host
    self.port = port
    self.reconnect = reconnect
    self.reconnect_delay = reconnect_delay
    self.max_reconnect_delay = max_reconnect_delay
    self.max_reconnect_attempts = max_reconnect_attempts
    self.include_raw = include_raw
    self.connect_timeout = connect_timeout
    self.overflow = overflow
    self.drain_on_disconnect = drain_on_disconnect
    self.handler_timeout = handler_timeout

    self._queue: asyncio.Queue[EventMessage] = asyncio.Queue(maxsize=queue_size)
    self._handlers_by_event: dict[str, list[_AnyCallable]] = defaultdict(list)
    self._handlers_any: list[_AnyCallable] = []
    self._on_connect_handlers: list[_SimpleHandler] = []
    self._on_disconnect_handlers: list[_SimpleHandler] = []
    self._error_handlers: list[_ErrorHandler] = []
    self._logger = logging.getLogger("rlstatsapi")

    self._reader: asyncio.StreamReader | None = None
    self._writer: asyncio.StreamWriter | None = None
    self._reader_task: asyncio.Task[None] | None = None
    self._stopping = False
    self._connection_state = ConnectionState.DISCONNECTED
    self._permanently_failed = False
    self._last_error: Exception | None = None
    self._metrics = ClientMetrics()
    self._state_tracker = MatchStateTracker()

```

--------------------------------

### CLI Configuration

Source: https://manucabral.github.io/rlstatsapi/configuration

Manage rlstatsapi configuration using command-line interface commands.

```bash
rlstatsapi enable --port 49123 --rate 30
rlstatsapi disable
rlstatsapi status

```

--------------------------------

### PodiumStart Event Structure

Source: https://manucabral.github.io/rlstatsapi/events

Fired when the post-match podium sequence begins.

```json
{
  "event": "PodiumStart",
  "data": {
    "MatchGuid": "A1B2C3D4E5F6"
  }
}
```

--------------------------------

### on_round_started

Source: https://manucabral.github.io/rlstatsapi/api

Typed helper for registering RoundStarted handlers.

```python
def on_round_started(
    self,
    handler: Callable[
        [TypedEventMessage[RoundStartedPayload]], Awaitable[None] | None
    ],
) -> None:
    """Typed helper for registering RoundStarted handlers."""
    self._handlers_by_event["RoundStarted"].append(handler)
```

--------------------------------

### on_match_initialized

Source: https://manucabral.github.io/rlstatsapi/api

Typed helper for registering MatchInitialized handlers.

```python
def on_match_initialized(
    self,
    handler: Callable[
        [TypedEventMessage[MatchInitializedPayload]], Awaitable[None] | None
    ],
) -> None:
    """Typed helper for registering MatchInitialized handlers."""
    self._handlers_by_event["MatchInitialized"].append(handler)
```

--------------------------------

### Code Formatting and Linting Checks

Source: https://manucabral.github.io/rlstatsapi/contributing

Commands to run for code formatting and linting.

```bash
python -m black .
python -m pylint src examples tests

```

--------------------------------

### on(event_name, handler=None) method signatures

Source: https://manucabral.github.io/rlstatsapi/api

Various signatures for the on method, demonstrating event registration.

```python
on(event_name: Literal['MatchPaused']) -> Callable[[Callable[[TypedEventMessage[MatchPausedPayload]], Awaitable[None] | None]], Callable[[TypedEventMessage[MatchPausedPayload]], Awaitable[None] | None]]
```

```python
on(event_name: Literal['MatchUnpaused'], handler: Callable[[TypedEventMessage[MatchUnpausedPayload]], Awaitable[None] | None]) -> None
```

```python
on(event_name: Literal['MatchUnpaused']) -> Callable[[Callable[[TypedEventMessage[MatchUnpausedPayload]], Awaitable[None] | None]], Callable[[TypedEventMessage[MatchUnpausedPayload]], Awaitable[None] | None]]
```

```python
on(event_name: Literal['PodiumStart'], handler: Callable[[TypedEventMessage[PodiumStartPayload]], Awaitable[None] | None]) -> None
```

```python
on(event_name: Literal['PodiumStart']) -> Callable[[Callable[[TypedEventMessage[PodiumStartPayload]], Awaitable[None] | None]], Callable[[TypedEventMessage[PodiumStartPayload]], Awaitable[None] | None]]
```

```python
on(event_name: Literal['ReplayCreated'], handler: Callable[[TypedEventMessage[ReplayCreatedPayload]], Awaitable[None] | None]) -> None
```

```python
on(event_name: Literal['ReplayCreated']) -> Callable[[Callable[[TypedEventMessage[ReplayCreatedPayload]], Awaitable[None] | None]], Callable[[TypedEventMessage[ReplayCreatedPayload]], Awaitable[None] | None]]
```

```python
on(event_name: Literal['RoundStarted'], handler: Callable[[TypedEventMessage[RoundStartedPayload]], Awaitable[None] | None]) -> None
```

```python
on(event_name: Literal['RoundStarted']) -> Callable[[Callable[[TypedEventMessage[RoundStartedPayload]], Awaitable[None] | None]], Callable[[TypedEventMessage[RoundStartedPayload]], Awaitable[None] | None]]
```

```python
on(event_name: Literal['StatfeedEvent'], handler: Callable[[TypedEventMessage[StatfeedEventPayload]], Awaitable[None] | None]) -> None
```

```python
on(event_name: Literal['StatfeedEvent']) -> Callable[[Callable[[TypedEventMessage[StatfeedEventPayload]], Awaitable[None] | None]], Callable[[TypedEventMessage[StatfeedEventPayload]], Awaitable[None] | None]]
```

```python
on(event_name: str, handler: Handler) -> None
```

```python
on(event_name: str) -> Callable[[Handler], Handler]
```

--------------------------------

### StatsClient Event Registration Methods

Source: https://manucabral.github.io/rlstatsapi/api

Demonstrates various methods for registering event handlers with the StatsClient, including general event registration, decorator usage, handling multiple events, and specialized typed handlers for specific game events.

```python
def on(self, event_name: str, handler: _AnyCallable | None = None) -> Any:
    """Register a handler for an event or return a decorator form of registration."""
    if handler is None:

        def decorator(h: _AnyCallable) -> _AnyCallable:
            @functools.wraps(h)
            def wrapper(*args: Any, **kwargs: Any) -> Any:
                return h(*args, **kwargs)

            self._handlers_by_event[event_name].append(wrapper)
            return wrapper

        return decorator
    self._handlers_by_event[event_name].append(handler)
    return None

def on_any(self, handler: Handler) -> None:
    """Register a handler that runs for every incoming event."""
    self._handlers_any.append(handler)

def on_many(self, event_names: Iterable[str], handler: Handler) -> None:
    """Register the same handler for several event names in one call."""
    for event_name in event_names:
        self._handlers_by_event[event_name].append(handler)

def off(self, event_name: str, handler: _AnyCallable) -> None:
    """Unregister one handler for a specific event if present."""
    handlers = self._handlers_by_event.get(event_name)
    if handlers:
        with contextlib.suppress(ValueError):
            handlers.remove(handler)

def off_any(self, handler: _AnyCallable) -> None:
    """Unregister a global handler registered with on_any."""
    with contextlib.suppress(ValueError):
        self._handlers_any.remove(handler)

def clear_queue(self) -> None:
    """Drop any queued events that have not been consumed yet."""
    while True:
        try:
            self._queue.get_nowait()
        except asyncio.QueueEmpty:
            break

def once(self, event_name: str, handler: _AnyCallable) -> None:
    """Register a handler that runs once and removes itself automatically."""

    async def wrapper(msg: EventMessage) -> None:
        self.off(event_name, wrapper)
        result = handler(msg)
        if inspect.isawaitable(result):
            await result

    self._handlers_by_event[event_name].append(wrapper)

def on_update_state(
    self,
    handler: Callable[
        [TypedEventMessage[UpdateStatePayload]], Awaitable[None] | None
    ],
) -> None:
    """Typed helper for registering UpdateState handlers."""
    self._handlers_by_event["UpdateState"].append(handler)

def on_ball_hit(
    self,
    handler: Callable[[TypedEventMessage[BallHitPayload]], Awaitable[None] | None],
) -> None:
    """Typed helper for registering BallHit handlers."""
    self._handlers_by_event["BallHit"].append(handler)

def on_clock_updated_seconds(
    self,
    handler: Callable[
        [TypedEventMessage[ClockUpdatedSecondsPayload]], Awaitable[None] | None
    ],
) -> None:
    """Typed helper for registering ClockUpdatedSeconds handlers."""
    self._handlers_by_event["ClockUpdatedSeconds"].append(handler)

def on_countdown_begin(
    self,
    handler: Callable[
        [TypedEventMessage[CountdownBeginPayload]], Awaitable[None] | None
    ],
) -> None:
    """Typed helper for registering CountdownBegin handlers."""
    self._handlers_by_event["CountdownBegin"].append(handler)

def on_crossbar_hit(
    self,
    handler: Callable[ [TypedEventMessage[CrossbarHitPayload]], Awaitable[None] | None ],
) -> None:
    """Typed helper for registering CrossbarHit handlers."""
    self._handlers_by_event["CrossbarHit"].append(handler)

def on_goal_replay_end(
    self,
    handler: Callable[ [TypedEventMessage[GoalReplayEndPayload]], Awaitable[None] | None ],
) -> None:
    """Typed helper for registering GoalReplayEnd handlers."""
    self._handlers_by_event["GoalReplayEnd"].append(handler)

def on_goal_replay_start(
    self,
    handler: Callable[ [TypedEventMessage[GoalReplayStartPayload]], Awaitable[None] | None ],
) -> None:
    """Typed helper for registering GoalReplayStart handlers."""
    self._handlers_by_event["GoalReplayStart"].append(handler)

def on_goal_replay_will_end(
    self,
    handler: Callable[ [TypedEventMessage[GoalReplayWillEndPayload]], Awaitable[None] | None ],
) -> None:
    """Typed helper for registering GoalReplayWillEnd handlers."""
    self._handlers_by_event["GoalReplayWillEnd"].append(handler)

def on_goal_scored(
    self,
    handler: Callable[ [TypedEventMessage[GoalScoredPayload]], Awaitable[None] | None ],
) -> None:
    """Typed helper for registering GoalScored handlers."""
    self._handlers_by_event["GoalScored"].append(handler)

def on_match_created(

```

--------------------------------

### Write all changed payloads to a file

Source: https://manucabral.github.io/rlstatsapi/recipes

This snippet shows how to run a script that writes all changed payloads to a file.

```bash
python examples/all_events_to_txt.py

```

--------------------------------

### Log goals only

Source: https://manucabral.github.io/rlstatsapi/recipes

This snippet shows how to listen for and log 'GoalScored' events.

```python
import asyncio
from rlstatsapi import StatsClient
from rlstatsapi.types import GoalScoredPayload, cast_event_data


def on_goal(msg) -> None:
    data: GoalScoredPayload = cast_event_data("GoalScored", msg.data)
    scorer = data.get("Scorer", {})
    print("Goal by:", scorer.get("Name"))


async def main() -> None:
    async with StatsClient() as client:
        client.on("GoalScored", on_goal)
        await asyncio.Event().wait()

```

--------------------------------

### __aenter__ Method

Source: https://manucabral.github.io/rlstatsapi/api

Connects the client and returns self for use as an asynchronous context manager.

```python
async def __aenter__(self) -> StatsClient:
    """Connect and return self for use as an async context manager."""
    await self.connect()
    return self
```

--------------------------------

### on_countdown_begin handler registration

Source: https://manucabral.github.io/rlstatsapi/api

Typed helper for registering CountdownBegin handlers.

```python
def on_countdown_begin(
    self,
    handler: Callable[
        [TypedEventMessage[CountdownBeginPayload]], Awaitable[None] | None
    ],
) -> None:
    """Typed helper for registering CountdownBegin handlers."""
    self._handlers_by_event["CountdownBegin"].append(handler)
```

--------------------------------

### on_any handler registration

Source: https://manucabral.github.io/rlstatsapi/api

Register a handler that runs for every incoming event.

```python
def on_any(self, handler: Handler) -> None:
    """Register a handler that runs for every incoming event."""
    self._handlers_any.append(handler)
```

--------------------------------

### Pylance-friendly pattern

Source: https://manucabral.github.io/rlstatsapi/typing

Demonstrates using `cast_event_data` for Pylance-friendly type hinting and IDE autocomplete when handling events.

```python
from rlstatsapi.models import EventMessage
from rlstatsapi.types import GoalScoredPayload, cast_event_data


def on_goal(msg: EventMessage) -> None:
    data: GoalScoredPayload = cast_event_data("GoalScored", msg.data)
    scorer = data.get("Scorer", {})
    print(scorer.get("Name"))

```

--------------------------------

### on_match_created

Source: https://manucabral.github.io/rlstatsapi/api

Typed helper for registering MatchCreated handlers.

```python
def on_match_created(
    self,
    handler: Callable[
        [TypedEventMessage[MatchCreatedPayload]], Awaitable[None] | None
    ],
) -> None:
    """Typed helper for registering MatchCreated handlers."""
    self._handlers_by_event["MatchCreated"].append(handler)
```

--------------------------------

### on_replay_created

Source: https://manucabral.github.io/rlstatsapi/api

Typed helper for registering ReplayCreated handlers.

```python
def on_replay_created(
    self,
    handler: Callable[
        [TypedEventMessage[ReplayCreatedPayload]], Awaitable[None] | None
    ],
) -> None:
    """Typed helper for registering ReplayCreated handlers."""
    self._handlers_by_event["ReplayCreated"].append(handler)
```

--------------------------------

### StatsClient Context Manager

Source: https://manucabral.github.io/rlstatsapi/api

Demonstrates using StatsClient as an asynchronous context manager for automatic connection and disconnection.

```python
async def __aenter__(self) -> StatsClient:
    """Connect and return self for use as an async context manager."""
    await self.connect()
    return self

async def __aexit__(self, *_: object) -> None:
    """Disconnect when leaving the async context manager."""
    await self.disconnect()
```

--------------------------------

### candidate_stats_api_paths()

Source: https://manucabral.github.io/rlstatsapi/api

Return likely user config locations for TAStatsAPI.ini on Windows.

```python
def candidate_stats_api_paths() -> list[Path]:
    """Return likely user config locations for ``TAStatsAPI.ini`` on Windows."""
    roots: list[Path] = []
    for env_name in ("USERPROFILE", "HOME"):
        value = os.getenv(env_name)
        if value:
            root = Path(value).expanduser().resolve()
            if root.exists():
                roots.append(root)

    candidates: list[Path] = []
    for root in roots:
        candidates.append(
            root / "Documents" / _USER_CONFIG_RELATIVE_DIR / DEFAULT_STATS_API_FILENAME
        )
        candidates.append(
            root / "Documentos" / _USER_CONFIG_RELATIVE_DIR / DEFAULT_STATS_API_FILENAME
        )
        candidates.append(
            root
            / "OneDrive"
            / "Documents"
            / _USER_CONFIG_RELATIVE_DIR
            / DEFAULT_STATS_API_FILENAME
        )
        candidates.append(
            root
            / "OneDrive"
            / "Documentos"
            / _USER_CONFIG_RELATIVE_DIR
            / DEFAULT_STATS_API_FILENAME
        )

    unique: list[Path] = []
    seen: set[str] = set()
    for path in candidates:
        key = str(path).casefold()
        if key not in seen:
            seen.add(key)
            unique.append(path)
    return unique
```

--------------------------------

### Manual INI Configuration

Source: https://manucabral.github.io/rlstatsapi/configuration

Manually configure rlstatsapi by editing the TAStatsAPI.ini file. This snippet shows the recommended minimum settings.

```ini
[TAGame.MatchStatsExporter_TA]
Port=49123
PacketSendRate=30

```

--------------------------------

### ClientMetrics.reset()

Source: https://manucabral.github.io/rlstatsapi/api

Resets all counters and updates the 'started_at' timestamp to the current time.

```python
def reset(self) -> None:
    """Reset all counters and update started_at to now."""
    self.received_events = 0
    self.queued_events = 0
    self.dropped_events = 0
    self.reconnect_count = 0
    self.handler_errors = 0
    self.connection_failures = 0
    self.started_at = datetime.now()
```

--------------------------------

### Python Configuration

Source: https://manucabral.github.io/rlstatsapi/configuration

Configure rlstatsapi automatically using Python by calling the configure_stats_api function with desired parameters.

```python
from rlstatsapi import configure_stats_api

configure_stats_api(enabled=True, port=49123, packet_send_rate=30)

```

--------------------------------

### once

Source: https://manucabral.github.io/rlstatsapi/api

Register a handler that runs once and removes itself automatically.

```python
def once(self, event_name: str, handler: _AnyCallable) -> None:
    """Register a handler that runs once and removes itself automatically."""

    async def wrapper(msg: EventMessage) -> None:
        self.off(event_name, wrapper)
        result = handler(msg)
        if inspect.isawaitable(result):
            await result

    self._handlers_by_event[event_name].append(wrapper)
```

--------------------------------

### on_goal_replay_end handler registration

Source: https://manucabral.github.io/rlstatsapi/api

Typed helper for registering GoalReplayEnd handlers.

```python
def on_goal_replay_end(
    self,
    handler: Callable[
        [TypedEventMessage[GoalReplayEndPayload]], Awaitable[None] | None
    ],
) -> None:
    """Typed helper for registering GoalReplayEnd handlers."""
    self._handlers_by_event["GoalReplayEnd"].append(handler)
```

--------------------------------

### on(event_name, handler=None)

Source: https://manucabral.github.io/rlstatsapi/api

Register a handler for a specific event.

```python
on(event_name: Literal['UpdateState'], handler: Callable[[TypedEventMessage[UpdateStatePayload]], Awaitable[None] | None]) -> None
```

```python
on(event_name: Literal['UpdateState']) -> Callable[[Callable[[TypedEventMessage[UpdateStatePayload]], Awaitable[None] | None]], Callable[[TypedEventMessage[UpdateStatePayload]], Awaitable[None] | None]]
```

```python
on(event_name: Literal['BallHit'], handler: Callable[[TypedEventMessage[BallHitPayload]], Awaitable[None] | None]) -> None
```

```python
on(event_name: Literal['BallHit']) -> Callable[[Callable[[TypedEventMessage[BallHitPayload]], Awaitable[None] | None]], Callable[[TypedEventMessage[BallHitPayload]], Awaitable[None] | None]]
```

```python
on(event_name: Literal['ClockUpdatedSeconds'], handler: Callable[[TypedEventMessage[ClockUpdatedSecondsPayload]], Awaitable[None] | None]) -> None
```

```python
on(event_name: Literal['ClockUpdatedSeconds']) -> Callable[[Callable[[TypedEventMessage[ClockUpdatedSecondsPayload]], Awaitable[None] | None]], Callable[[TypedEventMessage[ClockUpdatedSecondsPayload]], Awaitable[None] | None]]
```

```python
on(event_name: Literal['CountdownBegin'], handler: Callable[[TypedEventMessage[CountdownBeginPayload]], Awaitable[None] | None]) -> None
```

```python
on(event_name: Literal['CountdownBegin']) -> Callable[[Callable[[TypedEventMessage[CountdownBeginPayload]], Awaitable[None] | None]], Callable[[TypedEventMessage[CountdownBeginPayload]], Awaitable[None] | None]]
```

```python
on(event_name: Literal['CrossbarHit'], handler: Callable[[TypedEventMessage[CrossbarHitPayload]], Awaitable[None] | None]) -> None
```

```python
on(event_name: Literal['CrossbarHit']) -> Callable[[Callable[[TypedEventMessage[CrossbarHitPayload]], Awaitable[None] | None]], Callable[[TypedEventMessage[CrossbarHitPayload]], Awaitable[None] | None]]
```

```python
on(event_name: Literal['GoalReplayEnd'], handler: Callable[[TypedEventMessage[GoalReplayEndPayload]], Awaitable[None] | None]) -> None
```

```python
on(event_name: Literal['GoalReplayEnd']) -> Callable[[Callable[[TypedEventMessage[GoalReplayEndPayload]], Awaitable[None] | None]], Callable[[TypedEventMessage[GoalReplayEndPayload]], Awaitable[None] | None]]
```

```python
on(event_name: Literal['GoalReplayStart'], handler: Callable[[TypedEventMessage[GoalReplayStartPayload]], Awaitable[None] | None]) -> None
```

```python
on(event_name: Literal['GoalReplayStart']) -> Callable[[Callable[[TypedEventMessage[GoalReplayStartPayload]], Awaitable[None] | None]], Callable[[TypedEventMessage[GoalReplayStartPayload]], Awaitable[None] | None]]
```

```python
on(event_name: Literal['GoalReplayWillEnd'], handler: Callable[[TypedEventMessage[GoalReplayWillEndPayload]], Awaitable[None] | None]) -> None
```

```python
on(event_name: Literal['GoalReplayWillEnd']) -> Callable[[Callable[[TypedEventMessage[GoalReplayWillEndPayload]], Awaitable[None] | None]], Callable[[TypedEventMessage[GoalReplayWillEndPayload]], Awaitable[None] | None]]
```

```python
on(event_name: Literal['GoalScored'], handler: Callable[[TypedEventMessage[GoalScoredPayload]], Awaitable[None] | None]) -> None
```

```python
on(event_name: Literal['GoalScored']) -> Callable[[Callable[[TypedEventMessage[GoalScoredPayload]], Awaitable[None] | None]], Callable[[TypedEventMessage[GoalScoredPayload]], Awaitable[None] | None]]
```

```python
on(event_name: Literal['MatchCreated'], handler: Callable[[TypedEventMessage[MatchCreatedPayload]], Awaitable[None] | None]) -> None
```

```python
on(event_name: Literal['MatchCreated']) -> Callable[[Callable[[TypedEventMessage[MatchCreatedPayload]], Awaitable[None] | None]], Callable[[TypedEventMessage[MatchCreatedPayload]], Awaitable[None] | None]]
```

```python
on(event_name: Literal['MatchInitialized'], handler: Callable[[TypedEventMessage[MatchInitializedPayload]], Awaitable[None] | None]) -> None
```

```python
on(event_name: Literal['MatchInitialized']) -> Callable[[Callable[[TypedEventMessage[MatchInitializedPayload]], Awaitable[None] | None]], Callable[[TypedEventMessage[MatchInitializedPayload]], Awaitable[None] | None]]
```

```python
on(event_name: Literal['MatchDestroyed'], handler: Callable[[TypedEventMessage[MatchDestroyedPayload]], Awaitable[None] | None]) -> None
```

```python
on(event_name: Literal['MatchDestroyed']) -> Callable[[Callable[[TypedEventMessage[MatchDestroyedPayload]], Awaitable[None] | None]], Callable[[TypedEventMessage[MatchDestroyedPayload]], Awaitable[None] | None]]
```

```python
on(event_name: Literal['MatchEnded'], handler: Callable[[TypedEventMessage[MatchEndedPayload]], Awaitable[None] | None]) -> None
```

```python
on(event_name: Literal['MatchEnded']) -> Callable[[Callable[[TypedEventMessage[MatchEndedPayload]], Awaitable[None] | None]], Callable[[TypedEventMessage[MatchEndedPayload]], Awaitable[None] | None]]
```

```python
on(event_name: Literal['MatchPaused'], handler: Callable[[TypedEventMessage[MatchPausedPayload]], Awaitable[None] | None]) -> None
```

--------------------------------

### on_goal_replay_will_end handler registration

Source: https://manucabral.github.io/rlstatsapi/api

Typed helper for registering GoalReplayWillEnd handlers.

```python
def on_goal_replay_will_end(
    self,
    handler: Callable[
        [TypedEventMessage[GoalReplayWillEndPayload]], Awaitable[None] | None
    ],
) -> None:
    """Typed helper for registering GoalReplayWillEnd handlers."""
    self._handlers_by_event["GoalReplayWillEnd"].append(handler)
```

--------------------------------

### MatchStateTracker Player Snapshot

Source: https://manucabral.github.io/rlstatsapi/api

Parses player data from a game dictionary to create PlayerSnapshot objects.

```python
boost=p.get("Boost") or 0,
speed=float(p.get("Speed") or 0.0),
is_demolished=bool(p.get("bDemolished", False)),
)
)
snapshot.players = result
```

--------------------------------

### StatsClient Methods

Source: https://manucabral.github.io/rlstatsapi/api

This snippet shows the methods available on the StatsClient class for registering event handlers and callbacks for disconnections and errors.

```python
def on_disconnect(self, handler: _SimpleHandler) -> None:
    """Register a callback fired when the active TCP session is closed."""
    self._on_disconnect_handlers.append(handler)

def on_handler_error(self, handler: _ErrorHandler) -> None:
    """Register a callback for exceptions raised inside event handlers."""
    self._error_handlers.append(handler)
```

```python
@overload
def on(
    self,
    event_name: Literal["UpdateState"],
    handler: Callable[
        [TypedEventMessage[UpdateStatePayload]], Awaitable[None] | None
    ],
) -> None: ...
@overload
def on(self, event_name: Literal["UpdateState"]) -> Callable[
    [Callable[[TypedEventMessage[UpdateStatePayload]], Awaitable[None] | None]],
    Callable[[TypedEventMessage[UpdateStatePayload]], Awaitable[None] | None],
]: ...

@overload
def on(
    self,
    event_name: Literal["BallHit"],
    handler: Callable[[TypedEventMessage[BallHitPayload]], Awaitable[None] | None],
) -> None: ...
@overload
def on(self, event_name: Literal["BallHit"]) -> Callable[
    [Callable[[TypedEventMessage[BallHitPayload]], Awaitable[None] | None]],
    Callable[[TypedEventMessage[BallHitPayload]], Awaitable[None] | None],
]: ...

@overload
def on(
    self,
    event_name: Literal["ClockUpdatedSeconds"],
    handler: Callable[
        [TypedEventMessage[ClockUpdatedSecondsPayload]], Awaitable[None] | None
    ],
) -> None: ...
@overload
def on(self, event_name: Literal["ClockUpdatedSeconds"]) -> Callable[
    [
        Callable[
            [TypedEventMessage[ClockUpdatedSecondsPayload]], Awaitable[None] | None
        ]
    ],
    Callable[
        [TypedEventMessage[ClockUpdatedSecondsPayload]], Awaitable[None] | None
    ],
]: ...

@overload
def on(
    self,
    event_name: Literal["CountdownBegin"],
    handler: Callable[
        [TypedEventMessage[CountdownBeginPayload]], Awaitable[None] | None
    ],
) -> None: ...
@overload
def on(self, event_name: Literal["CountdownBegin"]) -> Callable[
    [Callable[[TypedEventMessage[CountdownBeginPayload]], Awaitable[None] | None]],
    Callable[[TypedEventMessage[CountdownBeginPayload]], Awaitable[None] | None],
]: ...

@overload
def on(
    self,
    event_name: Literal["CrossbarHit"],
    handler: Callable[
        [TypedEventMessage[CrossbarHitPayload]], Awaitable[None] | None
    ],
) -> None: ...
@overload
def on(self, event_name: Literal["CrossbarHit"]) -> Callable[
    [Callable[[TypedEventMessage[CrossbarHitPayload]], Awaitable[None] | None]],
    Callable[[TypedEventMessage[CrossbarHitPayload]], Awaitable[None] | None],
]: ...

@overload
def on(
    self,
    event_name: Literal["GoalReplayEnd"],
    handler: Callable[
        [TypedEventMessage[GoalReplayEndPayload]], Awaitable[None] | None
    ],
) -> None: ...
@overload
def on(self, event_name: Literal["GoalReplayEnd"]) -> Callable[
    [Callable[[TypedEventMessage[GoalReplayEndPayload]], Awaitable[None] | None]],
    Callable[[TypedEventMessage[GoalReplayEndPayload]], Awaitable[None] | None],
]: ...

@overload
def on(
    self,
    event_name: Literal["GoalReplayStart"],
    handler: Callable[
        [TypedEventMessage[GoalReplayStartPayload]], Awaitable[None] | None
    ],
) -> None: ...
@overload
def on(self, event_name: Literal["GoalReplayStart"]) -> Callable[
    [Callable[[TypedEventMessage[GoalReplayStartPayload]], Awaitable[None] | None]],
    Callable[[TypedEventMessage[GoalReplayStartPayload]], Awaitable[None] | None],
]: ...

@overload
def on(
    self,
    event_name: Literal["GoalReplayWillEnd"],
    handler: Callable[
        [TypedEventMessage[GoalReplayWillEndPayload]], Awaitable[None] | None
    ],
) -> None: ...
@overload
def on(self, event_name: Literal["GoalReplayWillEnd"]) -> Callable[
    [
        Callable[
            [TypedEventMessage[GoalReplayWillEndPayload]], Awaitable[None] | None
        ]
    ],
    Callable[[TypedEventMessage[GoalReplayWillEndPayload]], Awaitable[None] | None],
]: ...

@overload
def on(
    self,
    event_name: Literal["GoalScored"],
    handler: Callable[
        [TypedEventMessage[GoalScoredPayload]], Awaitable[None] | None
    ],
) -> None: ...
@overload
def on(self, event_name: Literal["GoalScored"]) -> Callable[
    [Callable[[TypedEventMessage[GoalScoredPayload]], Awaitable[None] | None]],
    Callable[[TypedEventMessage[GoalScoredPayload]], Awaitable[None] | None],
]: ...

@overload
def on(
    self,
    event_name: Literal["MatchCreated"],

```

--------------------------------

### configure_stats_api()

Source: https://manucabral.github.io/rlstatsapi/api

Set PacketSendRate and Port together in one write operation.

```python
def configure_stats_api(
    enabled: bool,
    port: int = DEFAULT_STATS_API_PORT,
    packet_send_rate: int = DEFAULT_PACKET_SEND_RATE,
    path: str | Path | None = None,
) -> StatsAPIConfigStatus:
    """Set ``PacketSendRate`` and ``Port`` together in one write operation."""
    _validate_port(port)
    if enabled:
        _validate_packet_send_rate(packet_send_rate)

    config_path = _require_config_path(path)
    text = config_path.read_text(encoding="utf-8")
    lines = text.splitlines()
    lines = _set_or_append_key(
        lines,
        "PacketSendRate",
        "0" if not enabled else str(packet_send_rate),
    )
    lines = _set_or_append_key(lines, "Port", str(port))
    _write_lines(config_path, lines, text)
    return _status_for_path(config_path)
```

--------------------------------

### MatchCreated Event Structure

Source: https://manucabral.github.io/rlstatsapi/events

This event is fired when a match session is created (before initialization).

```json
{
  "event": "MatchCreated",
  "data": {
    "MatchGuid": "A1B2C3D4E5F6"
  }
}
```
Tokens:
Raw