Skip to content

pygase.backend

Serve PyGaSe clients.

Provides the Server class and all PyGaSe components that deal with progression and syncing of game states.

Contents

  • GameStateStore: main API class for game state repositories
  • Server: main API class for PyGaSe servers
  • GameStateMachine: main API class for game logic components
  • Backend: main API class for a fully integrated PyGaSe backend

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.

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

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

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.