Skip to content

pygase

Create smooth and scalable online and LAN multiplayer games easily.

PyGaSe, or Python Game Server, is a library (or framework, whichever term you prefer) that provides a complete set of high-level components for real-time networking for games.

As a user of this library you will only need classes and functions directly imported from pygase:

# For clients:
from pygase import Client

# For backends:
from pygase import GameState, GameStateStore, GameStateMachine, Server
# Not necessary but might come in handy:
from pygase import get_availabe_ip_addresses

GameState

GameState(self, time_order:int=0, game_status:int=0, **kwargs)

Customize a serializable game state model.

Contains game state information that will be synchronized between the server and the clients. Via pygase.utils.Sendable its instances will be serialized using the msgpack protocol and must only contain attributes of type str, bytes, Sqn, int, float, bool as well as lists or tuples of such.

Arguments

  • time_order (int): current time order number of the game state, higher means more recent
  • game_status (int): GameStatus enum value that describes whether or not the game loop is running

Provide custom game state attributes via keyword arguments or assign them later.

Attributes

  • game_status (int): see constructor argument of same name
  • time_order (pygase.utils.Sqn): see constructor argument of same name

GameState instances mainly consist of custom attributes that make up the game state.

is_paused

GameState.is_paused(self) -> bool

Return True if game is paused.

Backend

Backend(self, initial_game_state:pygase.gamestate.GameState, time_step_function, event_handlers:dict=None)

Easily create a fully integrated PyGaSe backend.

Arguments

  • initial_game_state (GameState): state of the game before the simulation begins
  • time_step_function (callable): function that takes a game state and a time difference and returns a dict of updated game state attributes (see GameStateMachine.time_step())
  • event_handlers (dict): a dict with event types as keys and event handler functions as values

Attributes

  • game_state_store (GameStateStore): the backends game state repository
  • game_state_machine (GameStateMachine): logic component that runs the game loop
  • server (Server): handles connections to PyGaSe clients

Example

# Run a game loop that continuously increments `foo` with velocity `bar`.
Backend(
    initial_gamestate=GameState(foo=0.0, bar=0.5),
    time_step_function=lambda game_state, dt: {foo: game_state.foo + game_state.bar*dt},
    # Handle client events to reset `foo` and set a new `bar` value.
    event_handlers={
        "RESET_FOO": lambda game_state, dt: {foo: 0.0},
        "SET_BAR": lambda new_bar, game_state, dt: {bar: new_bar}
    }
).run(hostname="localhost", port=8080)

run

Backend.run(self, hostname:str, port:int)

Run state machine and server and bind the server to a given address.

Arguments

  • hostname (str): hostname or IPv4 address the server will be bound to
  • port (int): port number the server will be bound to

shutdown

Backend.shutdown(self)

Shut down server and stop game loop.

GameStateStore

GameStateStore(self, initial_game_state:pygase.gamestate.GameState=None)

Provide access to a game state and manage state updates.

Arguments

  • inital_game_state (GameState): state of the game before the simulation begins

Raises

  • TypeError: if 'initial_game_state' is not an instance of GameState

get_update_cache

GameStateStore.get_update_cache(self) -> list

Return the latest state updates.

get_game_state

GameStateStore.get_game_state(self) -> pygase.gamestate.GameState

Return the current game state.

push_update

GameStateStore.push_update(self, update:pygase.gamestate.GameStateUpdate) -> None

Push a new state update to the update cache.

This method will usually be called by whatever is progressing the game state, usually a GameStateMachine.

GameStateMachine

GameStateMachine(self, game_state_store:pygase.backend.GameStateStore)

Run a simulation that propagates the game state.

A GameStateMachine progresses a game state through time, applying all game simulation logic. This class is meant either as a base class from which you inherit and implement the GameStateMachine.time_step() method, or you assign an implementation after instantiation.

Arguments

  • game_state_store (GameStateStore): part of the PyGaSe backend that provides the state

Attributes

  • game_time (float): duration the game has been running in seconds

register_event_handler

GameStateMachine.register_event_handler(self, event_type:str, event_handler_function) -> None

Register an event handler for a specific event type.

For event handlers to have any effect, the events have to be wired from a Server to the GameStateMachine via the event_wire argument of the Server.run() method.

Arguments

  • event_type (str): which type of event to link the handler function to
  • handler_func (callable, coroutine): function or coroutine to be invoked for events of the given type

In addition to the event data, a GameStateMachine handler function gets passed the following keyword arguments

  • - game_state: game state at the time of the event
  • - dt: time since the last time step
  • - client_address: client which sent the event that is being handled

It is expected to return an update dict like the time_step method.

run_game_loop

GameStateMachine.run_game_loop(self, interval:float=0.02) -> None

Simulate the game world.

This function blocks as it continously progresses the game state through time but it can also be spawned as a coroutine or in a thread via Server.run_game_loop_in_thread(). As long as the simulation is running, the game_state.status will be GameStatus.get('Active').

Arguments

  • interval (float): (minimum) duration in seconds between consecutive time steps

run_game_loop_in_thread

GameStateMachine.run_game_loop_in_thread(self, interval:float=0.02) -> threading.Thread

Simulate the game in a seperate thread.

See GameStateMachine.run_game_loop().

Returns

threading.Thread: the thread the game loop runs in

stop

GameStateMachine.stop(self, timeout:float=1.0) -> bool

Pause the game simulation.

This sets self.status to Gamestatus.get('Paused'). This method can also be spawned as a coroutine. A subsequent call of GameStateMachine.run_game_loop() will resume the simulation at the point where it was stopped.

Arguments

  • timeout (float): time in seconds to wait for the simulation to stop

Returns

bool: wether or not the simulation was successfully stopped

time_step

GameStateMachine.time_step(self, game_state:pygase.gamestate.GameState, dt:float) -> dict

Calculate a game state update.

This method should be implemented to return a dict with all the updated state attributes.

Arguments

  • game_state (GameState): the state of the game prior to the time step
  • dt (float): time in seconds since the last time step, use it to simulate at a consistent speed

Returns

dict: updated game state attributes

Server

Server(self, game_state_store:pygase.backend.GameStateStore)

Listen to clients and orchestrate the flow of events and state updates.

The Server instance does not contain game logic or state, it is only responsible for connections to clients. The state is provided by a GameStateStore and game logic by a GameStateMachine.

Arguments

  • game_state_store (GameStateStore): part of the backend that provides an interface to the pygase.GameState

Attributes

  • connections (list): contains each clients address as a key leading to the corresponding pygase.connection.ServerConnection instance
  • host_client (tuple): address of the host client (who has permission to shutdown the server), if there is any
  • game_state_store (GameStateStore): game state repository

Members

  • hostname (str): read-only access to the servers hostname
  • port (int): read-only access to the servers port number

hostname

Get the hostname or IP address on which the server listens.

Returns None when the server is not running.

port

Get the port number on which the server listens.

Returns None when the server is not running.

run

Server.run(self, port:int=0, hostname:str='localhost', event_wire=None) -> None

Start the server under a specified address.

This is a blocking function but can also be spawned as a coroutine or in a thread via Server.run_in_thread().

Arguments

  • port (int): port number the server will be bound to, default will be an available port chosen by the computers network controller
  • hostname (str): hostname or IP address the server will be bound to. Defaults to 'localhost'.
  • event_wire (GameStateMachine): object to which events are to be repeated (has to implement a _push_event(event) method and is typically a GameStateMachine)

run_in_thread

Server.run_in_thread(self, port:int=0, hostname:str='localhost', event_wire=None, daemon=True) -> threading.Thread

Start the server in a seperate thread.

See Server.run().

Returns

threading.Thread: the thread the server loop runs in

shutdown

Server.shutdown(self) -> None

Shut down the server.

The server can be restarted via Server.run() in which case it will remember previous connections. This method can also be spawned as a coroutine.

dispatch_event

Server.dispatch_event(self, event_type:str, *args, target_client='all', retries:int=0, ack_callback=None, **kwargs) -> None

Send an event to one or all clients.

Arguments

  • event_type (str): identifies the event and links it to a handler
  • target_client (tuple, str): either 'all' for an event broadcast, or a clients address as a tuple
  • retries (int): number of times the event is to be resent in case it times out
  • ack_callback (callable, coroutine): will be executed after the event was received and be passed a reference to the corresponding pygase.connection.ServerConnection instance

Additional positional and keyword arguments will be sent as event data and passed to the clients handler function.

register_event_handler

Server.register_event_handler(self, event_type:str, event_handler_function) -> None

Register an event handler for a specific event type.

Arguments

  • event_type (str): event type to link the handler function to
  • handler_func (callable, coroutine): will be called for received events of the given type

Client

Client(self)

Exchange events with a PyGaSe server and access a synchronized game state.

Attributes

  • connection (pygase.connection.ClientConnection): object that contains all networking information

Example

from time import sleep
# Connect a client to the server from the Backend code example
client = Client()
client.connect_in_thread(hostname="localhost", port=8080)
# Increase `bar` five times, then reset `foo`
for i in range(5):
    client.dispatch_event("SET_BAR", new_bar=i)
    sleep(1)
client.dispatch_event("RESET_FOO")

connect

Client.connect(self, port:int, hostname:str='localhost') -> None

Open a connection to a PyGaSe server.

This is a blocking function but can also be spawned as a coroutine or in a thread via Client.connect_in_thread().

Arguments

  • port (int): port number of the server to which to connect
  • hostname (str): hostname or IPv4 address of the server to which to connect

connect_in_thread

Client.connect_in_thread(self, port:int, hostname:str='localhost') -> threading.Thread

Open a connection in a seperate thread.

See Client.connect().

Returns

threading.Thread: the thread the client loop runs in

disconnect

Client.disconnect(self, shutdown_server:bool=False) -> None

Close the client connection.

This method can also be spawned as a coroutine.

Arguments

  • shutdown_server (bool): wether or not the server should be shut down (only has an effect if the client has host permissions)

access_game_state

Client.access_game_state(self)

Return a context manager to access the shared game state.

Can be used in a with block to lock the synchronized game_state while working with it.

Example

with client.access_game_state() as game_state:
    do_stuff(game_state)

wait_until

Client.wait_until(self, game_state_condition, timeout:float=1.0) -> None

Block until a condition on the game state is satisfied.

Arguments

  • game_state_condition (callable): function that takes a pygase.GameState instance and returns a bool
  • timeout (float): time in seconds after which to raise a TimeoutError

Raises

  • TimeoutError: if the condition is not met after timeout seconds

try_to

Client.try_to(self, function, timeout:float=1.0)

Execute a function using game state attributes that might not yet exist.

This method repeatedly tries to execute function(game_state), ignoring KeyError exceptions, until it either works or times out.

Arguments

  • function (callable): function that takes a pygase.GameState instance and returns anything
  • timeout (float): time in seconds after which to raise a TimeoutError

Returns

any: whatever function(game_state) returns

Raises

  • TimeoutError: if the function doesn't run through after timeout seconds

dispatch_event

Client.dispatch_event(self, event_type:str, *args, retries:int=0, ack_callback=None, **kwargs) -> None

Send an event to the server.

Arguments

  • event_type (str): event type identifier that links to a handler
  • retries (int): number of times the event is to be resent in case it times out
  • ack_callback (callable, coroutine): will be invoked after the event was received

Additional positional and keyword arguments will be sent as event data and passed to the handler function.

ack_callback should not perform any long-running blocking operations (say a while True loop), as that will block the connections asynchronous event loop. Use a coroutine instead, with appropriately placed awaits.

register_event_handler

Client.register_event_handler(self, event_type:str, event_handler_function) -> None

Register an event handler for a specific event type.

Arguments

  • event_type (str): event type to link the handler function to
  • handler_func (callable, coroutine): will be called for events of the given type

get_available_ip_addresses

get_available_ip_addresses() -> list

Return a list of all locally available IPv4 addresses.