Event System Reference

Event definitions and utilities for the WebSocket event stream.

This module contains the custom EventClient that adds support for the real-time endpoint, defines the available event types, and provides a trigger system used to respond to in-game events.

Event Types

class auraxium.event.Event(*, event_name, timestamp, world_id)

An event returned via the ESS websocket connection.

event_name: str

The raw event name linked to this type. Generally identical to the name of the class.

timestamp: int

The UTC timestamp of the event. May be used to infer latency to the event streaming endpoint.

world_id: int

ID of the World whose event streaming endpoint broadcast the event.

age() → float

The age of the event in seconds.

class auraxium.event.AchievementAdded(*, event_name, timestamp, world_id, character_id, achievement_id, zone_id)

Bases: Event, CharacterEvent

A character has earned a new achievement.

Achievements are either weapon medals or service ribbons.

character_id: int

ID of the Character that earned the achievement.

achievement_id: int

ID of the Achievement that was earned.

zone_id: int

The current Zone of the character.

timestamp: int

The UTC timestamp of the event. May be used to infer latency to the event streaming endpoint.

world_id: int

ID of the World whose event streaming endpoint broadcast the event.

class auraxium.event.BattleRankUp(*, event_name, timestamp, world_id, battle_rank, character_id, zone_id)

Bases: Event, CharacterEvent

A character has earned a new battle rank.

Note that this may not reflect the characters actual new battle rank as they may be have joined the A.S.P.

battle_rank: int

The new battle rank of the character.

character_id: int

ID of the Character that ranked up.

zone_id: int

The current Zone of the character.

timestamp: int

The UTC timestamp of the event. May be used to infer latency to the event streaming endpoint.

world_id: int

ID of the World whose event streaming endpoint broadcast the event.

class auraxium.event.ContinentLock(*, event_name, timestamp, world_id, zone_id, triggering_faction, previous_faction, vs_population, nc_population, tr_population, metagame_event_id)

Bases: Event, WorldEvent

A continent has been locked.

zone_id: int

ID of the Zone that was locked.

triggering_faction: int

The faction that triggered the meltdown alert.

previous_faction: int

The faction that has previously held this continent.

vs_population: float

Population percentage for VS.

nc_population: float

Population percentage for NC.

tr_population: float

Population percentage for TR.

metagame_event_id: int

The ID of the MetagameEvent that caused the continent to lock.

timestamp: int

The UTC timestamp of the event. May be used to infer latency to the event streaming endpoint.

world_id: int

ID of the World whose event streaming endpoint broadcast the event.

class auraxium.event.Death(*, event_name, timestamp, world_id, attacker_character_id, attacker_fire_mode_id, attacker_loadout_id, attacker_vehicle_id, attacker_weapon_id, attacker_team_id, character_id, character_loadout_id, is_critical=None, is_headshot, team_id, vehicle_id=None, zone_id)

Bases: Event, CharacterEvent

A character has been killed.

If the attacker and victim ID are identical, the character has killed themselves (e.g. with explosives).

An attacker ID of 0 indicates that the player has died to non-player sources like fall damage, or spawn room pain fields.

attacker_character_id: int

The ID of the killing Character.

attacker_fire_mode_id: int

ID of the FireMode used.

attacker_loadout_id: int

ID of the Loadout of the attacker.

attacker_vehicle_id: int

ID of the Vehicle of the attacker.

attacker_weapon_id: int

ID of the Item used by the attacker.

Important

The reference above is not an error, this field reports the item ID of the weapon, not the weapon ID.

attacker_team_id: int

ID of the team of the attacker.

character_id: int

ID of the Character that was killed.

character_loadout_id: int

ID of the Loadout of the victim.

is_critical: bool | None

Whether the killing blow dealt critical damage.

Note

This value is always false.

is_headshot: bool

Whether the killing blow was dealt via headshot.

team_id: int

Team ID of the victim.

vehicle_id: int | None

The type of Vehicle the victim was in at the time of death, if any.

zone_id: int

The current Zone of the character.

timestamp: int

The UTC timestamp of the event. May be used to infer latency to the event streaming endpoint.

world_id: int

ID of the World whose event streaming endpoint broadcast the event.

class auraxium.event.FacilityControl(*, event_name, timestamp, world_id, duration_held, facility_id, new_faction_id, old_faction_id, outfit_id, zone_id)

Bases: Event, WorldEvent

A facility has switched factions.

This is generally due to hostile takeover, but is also dispatched when a coninent is locked or unlocked server-side (e.g. due to an alert ending).

duration_held: int

The amount of time the base was in the old faction’s ownership in seconds.

facility_id: int

The facility ID of the base.

new_faction_id: int

ID of the new Faction.

old_faction_id: int

ID of the old Faction.

outfit_id: int

ID of the Outfit that was awarded the base capture.

zone_id: int

The Zone of the captured base.

timestamp: int

The UTC timestamp of the event. May be used to infer latency to the event streaming endpoint.

world_id: int

ID of the World whose event streaming endpoint broadcast the event.

class auraxium.event.GainExperience(*, event_name, timestamp, world_id, amount, character_id, experience_id, loadout_id, other_id, zone_id)

Bases: Event, CharacterEvent

A character has gained a tick of experience.

amount: int

The amount of experience gained.

character_id: int

ID of the Character that earned experience.

experience_id: int

The source of the experience gain.

Note

Not all types of experience gain have a cooresponding Experience entry.

loadout_id: int

The current Loadout of the character.

other_id: int

The ID of another entity involved in the experience acquisition. For heals, this would be the healed ally, for spots the enemy vehicle or player spottet, etc.

zone_id: int

The current Zone of the character.

timestamp: int

The UTC timestamp of the event. May be used to infer latency to the event streaming endpoint.

world_id: int

ID of the World whose event streaming endpoint broadcast the event.

classmethod filter_experience(id_: int) → str

Factory for custom, experience ID specific events.

This method is used to generate custom event names that allow only subscribing to a single type of experience gain. The returned string can be passed to a auraxium.event.Trigger.

Parameters:

id (int) – The experience ID to subscribe to.

Returns:

A custom event name for the given experience type.

class auraxium.event.ItemAdded(*, event_name, timestamp, world_id, character_id, context, item_count, item_id, zone_id)

Bases: Event, CharacterEvent

A character has been granted an item.

This includes internal flags and invisible items used to control outfit resources and war assets.

character_id: int

ID of the character that has been awarded an item.

context: str

The reason or mechanic that led to the item being awarded. Notably, this includes outfit resource use.

item_count: str

The number of items that were added.

item_id: int

The Item that was added.

zone_id: int

The current Zone of the character.

timestamp: int

The UTC timestamp of the event. May be used to infer latency to the event streaming endpoint.

world_id: int

ID of the World whose event streaming endpoint broadcast the event.

class auraxium.event.MetagameEvent(*, event_name, timestamp, world_id, experience_bonus, faction_nc, faction_tr, faction_vs, instance_id, metagame_event_id, metagame_event_state, metagame_event_state_name, zone_id=-1)

Bases: Event, WorldEvent

A metagame event (i.e. alert) has changed state.

experience_bonus: int

The experience bonus applied for the duration of the event (a value of 25 denotes a 25% experience bonus for all participants)

faction_nc: float

The current event score of the NC.

faction_tr: float

The current event score of the TR.

faction_vs: float

The current event score of the NC.

metagame_event_id: int

ID of the MetagameEvent that changed state.

metagame_event_state: int

The new MetagameEventState of the event.

zone_id: int

The Zone the event is taking place in.

timestamp: int

The UTC timestamp of the event. May be used to infer latency to the event streaming endpoint.

world_id: int

ID of the World whose event streaming endpoint broadcast the event.

class auraxium.event.PlayerFacilityCapture(*, event_name, timestamp, world_id, character_id, facility_id, outfit_id, zone_id)

Bases: Event, CharacterEvent

A player has participated in capturing a facility.

character_id: int

The ID of the Character that participated in the capture.

facility_id: int

ID of the facility that was captured.

outfit_id: int

The Outfit that was awarded the facility.

zone_id: int

The current Zone of the character.

timestamp: int

The UTC timestamp of the event. May be used to infer latency to the event streaming endpoint.

world_id: int

ID of the World whose event streaming endpoint broadcast the event.

class auraxium.event.PlayerFacilityDefend(*, event_name, timestamp, world_id, character_id, facility_id, outfit_id, zone_id)

Bases: Event, CharacterEvent

A player has participated in defending a facility.

character_id: int

The ID of the Character that participated in the defence.

facility_id: int

ID of the facility that was defended.

outfit_id: int

The Outfit that currently owns the facility.

zone_id: int

The current Zone of the character.

timestamp: int

The UTC timestamp of the event. May be used to infer latency to the event streaming endpoint.

world_id: int

ID of the World whose event streaming endpoint broadcast the event.

class auraxium.event.PlayerLogin(*, event_name, timestamp, world_id, character_id)

Bases: Event, CharacterEvent, WorldEvent

A player has logged into the game.

character_id: int

The Character that logged in.

timestamp: int

The UTC timestamp of the event. May be used to infer latency to the event streaming endpoint.

world_id: int

ID of the World whose event streaming endpoint broadcast the event.

class auraxium.event.PlayerLogout(*, event_name, timestamp, world_id, character_id)

Bases: Event, CharacterEvent, WorldEvent

A player has logged out.

character_id: int

The Character that logged off.

timestamp: int

The UTC timestamp of the event. May be used to infer latency to the event streaming endpoint.

world_id: int

ID of the World whose event streaming endpoint broadcast the event.

class auraxium.event.SkillAdded(*, event_name, timestamp, world_id, character_id, skill_id, zone_id)

Bases: Event, CharacterEvent

A player has unlocked a skill (i.e. certification or ASP).

character_id: int

The Character that gained a new skill.

skill_id: int

The Skill the character unlocked.

zone_id: int

The current Zone of the character.

timestamp: int

The UTC timestamp of the event. May be used to infer latency to the event streaming endpoint.

world_id: int

ID of the World whose event streaming endpoint broadcast the event.

class auraxium.event.VehicleDestroy(*, event_name, timestamp, world_id, attacker_character_id, attacker_loadout_id, attacker_vehicle_id, attacker_weapon_id, attacker_team_id, character_id, facility_id, faction_id, team_id, vehicle_id, zone_id)

Bases: Event, CharacterEvent

A player’s vehicle has been destroyed.

attacker_character_id: int

The ID of the killing Character.

attacker_loadout_id: int

ID of the Loadout of the attacker.

attacker_vehicle_id: int

ID of the Vehicle of the attacker.

attacker_weapon_id: int

ID of the Item used by the attacker.

Important

The reference above is not an error, this field reports the item ID of the weapon, not the weapon ID.

attacker_team_id: int

The ID of the team of the attacker

character_id: int

ID of the Character that was killed.

facility_id: int

ID of the facility the vehicle was destroyed at.

Note

As of March 2021, this field is only populated for destroyed base turrets. All other vehicles do not contain facility data.

faction_id: int

The Faction of the vehicle.

team_id: int

Team ID of the destroyed vehicle.

vehicle_id: int | None

The type of Vehicle that was destroyed.

zone_id: int

The Zone the vehicle was destroyed in.

timestamp: int

The UTC timestamp of the event. May be used to infer latency to the event streaming endpoint.

world_id: int

ID of the World whose event streaming endpoint broadcast the event.

Event Client

class auraxium.event.EventClient(*args, ess_endpoint=None, **kwargs)

Advanced client with event streaming capability.

This subclass of auraxium.Client extends the interface to also provide access to the WebSocket endpoint at wss://push.planetside2.com/streaming.

To use the websocket endpoint, you have to define a Trigger and register it using the add_trigger() method. This will automatically open a WebSocket connection is there is not already one running. Likewise, removing all event triggers from the client will cause the underlying websocket connection to close.

Refer to the Trigger class’s documentation for details on how to use triggers and respond to events.

ess_endpoint: yarl.URL

The URL of the event streaming service endpoint. If not set, defaults to the daybreak games streaming endpoint.

triggers: list[auraxium.event.Trigger]

The list of Triggers registered for the client.

websocket: websockets.client.WebSocketClientProtocol | None

The websocket client used for the real-time event stream. This will be automatically opened and closed by the client as event triggers are added and removed.

endpoint_status() → dict[str, bool]

Return endpoint status info.

This returns a dictionary mapping API event server endpoints to their last reported status. This generally refreshes every 30 seconds as part of the WebSocket heartbeat messages.

add_trigger(trigger: Trigger) → None

Add a new event trigger to the client.

If there is currently no active websocket connection to the event streaming service, one will be created for this trigger.

Note

As this is a synchronous method, the WebSocket will not be active by the time this method returns. You can use EventClient.wait_ready() to wait for the WebSocket being ready to process subscriptions.

Parameters:

trigger (Trigger) – The trigger to add.

get_trigger(name: str) → Trigger

Retrieve a registered event trigger by name.

If the trigger cannot be found, a KeyError is raised.

Parameters:

name (str) – The name of the trigger to return.

Raises:

KeyError – Raised if no trigger with the given name is registered for the client.

remove_trigger(trigger: Trigger | str, *, keep_websocket_alive: bool = False) → None

Remove an existing event trigger.

You can either provide the trigger instance to remove, or specify the name of the trigger instead.

By default, the underlying websocket connection will be closed if this was the only trigger registered. Use the keep_websocket_alive flag to prevent this, for example if you intend to immediately add another trigger.

Parameters:

  • trigger (Trigger | str) – The trigger to remove, or its unique name.

  • keep_websocket_alive (bool) – If true, the websocket connection will be kept open even if the client has zero triggers remaining.

Raises:

  • KeyError – Raised if a trigger name was passed and no trigger of this name exists for the client.

  • ValueError – Raised if no trigger of the given name is currently registered for this client.

async close() → None

Gracefully shut down the client.

This will close the WebSocket connection and end any ongoing HTTP sessions used for requests to the REST API.

Call this to clean up before the client object is destroyed.

async connect() → None

Connect to the websocket endpoint and process responses.

This will continuously loop until EventClient.close() is called. If the WebSocket connection encounters and error, it will be automatically restarted.

Any event payloads received will be passed to EventClient.dispatch() for filtering and event dispatch.

async disconnect() → None

Disconnect the WebSocket.

Unlike EventClient.close(), this does not affect the HTTP sessions used by regular REST requests.

dispatch(event: Event) → None

Dispatch an event to the appropriate event triggers.

This goes through the list of triggers registered for this client and checks if the passed event matches the trigger’s requirements using Trigger.check.

The call-backs for the matching triggers will be scheduled for execution in the current event loop using asyncio.loop.create_task().

If a trigger’s single_shot attribute is set to true, the trigger will be removed from the client as soon as its call-back has been scheduled for execution. This means that when the action associated with a single-shot trigger runs, the associated trigger will no longer be registered for the client.

Parameters:

event (auraxium.event.Event) – An event received through the event stream.

trigger(self, event: str | type[Event], *args: str | type[Event], name: str | None = None, **kwargs) → Callable[Callable[[Event], Coroutine[None]], None]

Create and add a trigger for the given action.

If no name is specified, the call-back function’s name will be used as the trigger name.

Keep in mind that a trigger’s name must be unique. A KeyError will be raised if a trigger with this name already exists.

Parameters:
Raises:

KeyError – Raised if a trigger with the given name already exists.

async wait_for(trigger: Trigger, *args: Trigger, timeout: float | None = None) → Event

Wait for one or more triggers to fire.

This method will wait until any of the given triggers have fired, or until the timeout has been exceeded.

By default, any triggers passed will be automatically removed once the first has been triggered, regardless of the triggers’ single_shot setting.

Parameters:

  • trigger (Trigger) – A trigger to wait for.

  • args (Trigger) – Additional triggers that will also resume execution.

  • timeout (float | None) – The maximum number of seconds to wait for. Never expires if set to None.

Raises:

TimeoutError – Raised if timeout is exceeded.

Returns:

The first event matching the given trigger(s).

async wait_ready(interval: float = 0.05) → None

Wait for the WebSocket connection to be ready.

This will return once the WebSocket connection is open and active. This condition will be checked regularly as set by the interval argument.

If the WebSocket is already active at the time this method is called, this will return without delay.

Parameters:

interval (float) – The interval at which to check the WebSocket connection’s status.

Triggers

class auraxium.event.Trigger(event, *args, characters=None, worlds=None, conditions=None, action=None, name=None, single_shot=False)

An event trigger for the client’s websocket connection.

Event triggers encapsulate both the event type to trigger on, as well as the action to perform when the event is encountered.

They are also used to dynamically generate the subscription payload required to inform the event streaming service of the event types the client wishes to receive.

Note that some subscriptions are incompatible with each other and may require multiple clients to be stable.

action: collections.abc.Callable[[auraxium.event.Event], None] | collections.abc.Callable[[Coroutine[None]], None]

The method or coroutine to run if the matching event is encountered.

characters: list[int]

A list of characters to filter the incoming events by. For some events, like auraxium.event.Death, both the victim and killer can lead to a match.

conditions: list[collections.abc.Callable[[auraxium.event.Event], bool]]

Any number of callables that must return true for the trigger to run. Note that these filters are checked for any matching event types. Any callables used must be synchronous.

events: set[type[auraxium.event.Event] | str]

A set of events that the trigger will listen for.

last_run: datetime.datetime | None

A datetime.datetime instance that will be set to the last time the trigger has run. This will be None until the first run of te trigger.

name: str

The unique name of the trigger.

single_shot: bool

If True, the trigger will be automatically removed from the client when it first fires.

worlds: list[int]

A list of worlds to filter the incoming events by.

__init__(event: type[Event] | str, *args: type[Event] | str, characters: collections.abc.Iterable[auraxium.ps2.Character | int] | None = None, worlds: collections.abc.Iterable[auraxium.ps2.World | int] | None = None, conditions: list[Callable[[Event], bool]] | None = None, action: Callable[[Event], Coroutine[None] | None] | None = None, name: str | None = None, single_shot: bool = False) → None

Initialise a new trigger.

See also

auraxium.event.EventClient.trigger() – Decorator used to define a trigger around a given function.

Parameters:
callback(func: Callable[[Event], Coroutine[None] | None]) → None

Set the given function as the trigger action.

The action may be a regular callable or a coroutine. Any callable that is a coroutine function according to inspect.iscoroutinefunction() will be awaited.

This method can be used as a decorator.

my_trigger = Trigger('Death')

@my_trigger.callback
def pay_respect(event):
    char = event.character_id
    print('F ({char})')
Parameters:

func (collections.abc.Callable[[Event], None] | collections.abc.Callable[[Coroutine[None]], None]) – The method or coroutine to call when the event trigger fires.

check(event: Event) → bool

Return whether the given trigger should fire.

This only returns whether the trigger should fire, the trigger action will be scheduled separately, at which point Trigger.run() is called.

Parameters:

event (Event) – The event to check.

Returns:

Whether this trigger should run for the given event.

generate_subscription(logical_and: bool | None = None) → str

Generate the appropriate subscription for this trigger.

async run(event: Event) → None

Perform the action associated with this trigger.

Parameters:

event (Event) – The event to pass to the trigger action.