Skip to content

pygase.client

Connect to PyGaSe servers.

Contents

  • Client: main API class for PyGaSe clients

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