progames.core.engine package¶

Submodules¶

progames.core.engine.abc module¶

class progames.core.engine.abc.Store¶

Bases: collections.abc.Sized

empty() → bool¶: check if the store is empty

pop(*args, **kwargs) → Item¶

get the first item

Returns:	the first item

store(entity: Item, *args, **kwargs) → None¶

store item

Parameters:	entity – item to store

class progames.core.engine.abc.GameWorld¶

Bases: object

accept(command: progames.core.engine.messages.Command) → Iterable[progames.core.engine.messages.Event]¶

Parameters:	command – command

rollback(command: progames.core.engine.messages.Command) → None¶

Parameters:	command – the failed command which causes rollback

update() → None¶: Some game need update game world each frame

class progames.core.engine.abc.Message¶: Bases: object

progames.core.engine.concurrent module¶

progames.core.engine.concurrent.run(func, args=None, kwargs=None)¶

execute a function in default thread pool

Parameters:	func (function) – function to be executed args (tuple) – args kwargs (dict) – kwargs
Returns:	None

progames.core.engine.messages module¶

class progames.core.engine.messages.Command¶

Bases: progames.core.engine.abc.Message

base command

visit(world: progames.core.engine.abc.GameWorld) → Iterable[progames.core.engine.messages.Event]¶

class progames.core.engine.messages.Event¶

Bases: progames.core.engine.abc.Message

base event

class progames.core.engine.messages.GameEnded¶

Bases: progames.core.engine.messages.Event

default event raise when game is ended

progames.core.engine.processors module¶

class progames.core.engine.processors.Processor(game: progames.core.engine.GameWorld, store: progames.core.engine.Store, *args, **kwargs)¶

Bases: object

Process game and produce events

Notes

Player’s input data will be processed and transform to commands.
Each frame, game only consume 1 command from command store.
Command is a visitor, travel around game world and produce events. These events will be collected and sent to listeners.
Rollback is required to implement

command_store = None¶: command store

cps = 50¶: commands per seconds (max), 0 < cps

get_next_command() → typing.Optional[progames.core.engine.Command]¶

get command to process

:raise RuntimeWarning

handle(event: progames.core.engine.Event) → None¶: base handle event, don’t forget call: super().handle(event)

is_over() → bool¶: check if game is over

over = None¶: flag for checking if game is over

process_command() → None¶

process commands one by one

Flows:

pop 1 command from command store
the command go through game world, make changes and produce events
notify events to listeners (register)
update game world

Raises:	RuntimeError – game must stop RuntimeWarning – game continue

start()¶: start processor

state = None¶

stop() → None¶: immediate stop game loop

progames.core.engine.producers module¶

class progames.core.engine.producers.Producer(store: progames.core.engine.abc.Store, generator: collections.abc.AsyncGenerator)¶

Bases: object

produce_async()¶

start()¶: listen to inputs

progames.core.engine.stores module¶

class progames.core.engine.stores.SimpleQueueStore¶

Bases: progames.core.engine.abc.Store

empty()¶: check if the store is empty

pop(*args, **kwargs)¶

get the first item

Returns:	the first item

store(entity, *args, **kwargs)¶

store item

Parameters:	entity – item to store

progames.core.engine.worlds module¶

class progames.core.engine.worlds.AbstractGameWorld¶

Bases: progames.core.engine.abc.GameWorld

accept(command: progames.core.engine.messages.Command) → List[progames.core.engine.messages.Event]¶

Parameters:	command – command

process(command: progames.core.engine.messages.Command) → List[progames.core.engine.messages.Event]¶

put your logic here

Parameters:	command – the command
Returns:	list of events
Raises:	`RuntimeError` – game crashed `RuntimeWarning` – continue

rollback(command: progames.core.engine.messages.Command) → None¶

Parameters:	command – the failed command which causes rollback

update() → None¶: Some game need update game world each frame

validate(command: progames.core.engine.messages.Command)¶

validate command

Raises:	`ValidationError` – invalid command

Module contents¶

class progames.core.engine.Command¶

Bases: progames.core.engine.abc.Message

base command

visit(world: progames.core.engine.abc.GameWorld) → Iterable[progames.core.engine.messages.Event]¶

class progames.core.engine.Event¶

Bases: progames.core.engine.abc.Message

base event

class progames.core.engine.GameEnded¶

Bases: progames.core.engine.messages.Event

default event raise when game is ended

class progames.core.engine.GameWorld¶

Bases: object

accept(command: progames.core.engine.messages.Command) → Iterable[progames.core.engine.messages.Event]¶

Parameters:	command – command

rollback(command: progames.core.engine.messages.Command) → None¶

Parameters:	command – the failed command which causes rollback

update() → None¶: Some game need update game world each frame

class progames.core.engine.AbstractGameWorld¶

Bases: progames.core.engine.abc.GameWorld

accept(command: progames.core.engine.messages.Command) → List[progames.core.engine.messages.Event]¶

Parameters:	command – command

process(command: progames.core.engine.messages.Command) → List[progames.core.engine.messages.Event]¶

put your logic here

Parameters:	command – the command
Returns:	list of events
Raises:	`RuntimeError` – game crashed `RuntimeWarning` – continue

rollback(command: progames.core.engine.messages.Command) → None¶

Parameters:	command – the failed command which causes rollback

update() → None¶: Some game need update game world each frame

validate(command: progames.core.engine.messages.Command)¶

validate command

Raises:	`ValidationError` – invalid command

class progames.core.engine.Producer(store: progames.core.engine.abc.Store, generator: collections.abc.AsyncGenerator)¶

Bases: object

produce_async()¶

start()¶: listen to inputs

class progames.core.engine.Processor(game: progames.core.engine.GameWorld, store: progames.core.engine.Store, *args, **kwargs)¶

Bases: object

Process game and produce events

Notes

Player’s input data will be processed and transform to commands.
Each frame, game only consume 1 command from command store.
Command is a visitor, travel around game world and produce events. These events will be collected and sent to listeners.
Rollback is required to implement

command_store = None¶: command store

cps = 50¶: commands per seconds (max), 0 < cps

get_next_command() → typing.Optional[progames.core.engine.Command]¶

get command to process

:raise RuntimeWarning

handle(event: progames.core.engine.Event) → None¶: base handle event, don’t forget call: super().handle(event)

is_over() → bool¶: check if game is over

over = None¶: flag for checking if game is over

process_command() → None¶

process commands one by one

Flows:

pop 1 command from command store
the command go through game world, make changes and produce events
notify events to listeners (register)
update game world

Raises:	RuntimeError – game must stop RuntimeWarning – game continue

start()¶: start processor

state = None¶

stop() → None¶: immediate stop game loop

class progames.core.engine.SimpleQueueStore¶

Bases: progames.core.engine.abc.Store

empty()¶: check if the store is empty

pop(*args, **kwargs)¶

get the first item

Returns:	the first item

store(entity, *args, **kwargs)¶

store item

Parameters:	entity – item to store