class GhostVariant(Enum):
+ """
+ Enumeration for Ghost variant and behavior.
+ """
+
A = auto()
B = auto()
C = auto()
slf: "Ghost",
blinky_pos: tuple[int, int],
) -> tuple[int, int]:
+ """
+ Return ghost target coordinates.
+ """
match self:
case GhostVariant.A:
return player
return (x_target * 2 - x_blinky, y_target * 2 - y_blinky)
+class GhostState(Enum):
+ """
+ Enumeration for Ghost state and attitudes.
+ """
+
+ chase = auto()
+ wander = auto()
+ flee = auto()
+ dead = auto()
+ respawning = auto()
+
+
@dataclass
class InitialPauseTimer(Resource):
+ """
+ Resource use for the initial pause before starting a level.
+ """
+
tick: int = 0
@classmethod
class Config(Resource, BaseModel):
+ """
+ Configuration of the game.
+ """
+
ghost_chase_speed: float = Field(default=2.5, ge=0.1, le=100.0)
ghost_flee_speed: float = Field(default=1.5, ge=0.1, le=100.0)
pacman_speed: float = Field(default=3.0, ge=0.1, le=100.0)
class Highscores(Resource, BaseModel):
+ """
+ Store the hichscores in a dictionnary.
+ """
+
scores: dict[str, int]
class GameIsOver(Resource):
+ """
+ Resource set when game is over.
+ """
+
pass
@dataclass
class GameState(Resource):
+ """
+ Resource storing current score, level, timer and lives.
+ """
+
score: int = 0
level: int = 0
timer: int = 0
class MainMenu(Component):
+ """
+ Component for main menu items.
+ """
+
pass
class GameMenu(Component):
+ """
+ Component for game menu items.
+ """
+
pass
class GameGui(Component):
+ """
+ Component for game graphical user interface items.
+ """
+
pass
class PacGum(Component):
+ """
+ Component for pacgums items.
+ """
+
pass
class SuperPacGum(Component):
+ """
+ Component for super pacgums items.
+ """
+
pass
@dataclass
class Pacman(Component):
+ """
+ Component for pacman items storing the home coordinates and entity.
+ """
+
home: tuple[int, int]
home_entity: Entity
-class GhostState(Enum):
- chase = auto()
- wander = auto()
- flee = auto()
- dead = auto()
- respawning = auto()
-
-
@dataclass
class GhostStatus(Component):
+ """
+ Component for ghost state and cycle.
+ """
+
tick: int = 0
cycle_wander: int = 60 * 10
state: GhostState = GhostState.wander
@dataclass
class Ghost(Component):
+ """
+ Component for ghost items storing the ghost variant and the home
+ coordinates and entity.
+ """
+
variant: GhostVariant
home: tuple[int, int]
home_entity: Entity
@dataclass
class TargetTile(Component):
+ """
+ Component for the coordinates of a target used for ghost movement.
+ """
+
pos: tuple[int, int]
@dataclass
class TilePos(Component):
+ """
+ Component tile position as entity.
+ """
+
pos: Entity
@dataclass
class InMotion(Component):
+ """
+ Component for pacman or ghost movement storing the position, the provenance
+ and the direction.
+ """
+
src: Entity
pos: Entity
direction: int
@dataclass
class MoveSpeed(Component):
+ """
+ Component for movement storing the movement speed by pixels per frames.
+ """
+
pixels_per_frame: float
class Maze(Component):
+ """
+ Component for maze storing the x and y dimension and the cells in a
+ dictionnary (coordonate entity).
+ """
+
def __init__(self, x: int, y: int) -> None:
self._cells: dict[tuple[int, int], Entity] = {}
self._x = x
@dataclass
class Cell(Component):
+ """
+ Component for the cells of the maze storing the value and the position.
+ """
+
n: int
grid_pos: tuple[int, int]
def neighbor(self, idx: int) -> tuple[int, int]:
+ """
+ Get neighbor position by direction (1: up, 2: right ...).
+ """
dx, dy = _DIR_OFFSETS[idx]
return (self.grid_pos[0] + dx, self.grid_pos[1] + dy)
def walls(self) -> list[bool]:
+ """
+ Get list of walls.
+ """
return [(1 & (self.n >> i)) == 1 for i in range(4)]
def remove_comment(s: str) -> str:
+ """
+ Return s with all comment lines removed.
+ """
return "\n".join(line for line in s.split("\n") if not s.startswith("#"))
def load_config(world: World) -> None:
+ """
+ Load the configuration json file given in first argument and spawn the
+ Config resource.
+ """
if len(sys.argv) != 2:
raise Exception(
f"Expected exactly 1 argument, got {len(sys.argv) - 1}"
def load_highscores(world: World, config: Config) -> None:
+ """
+ Load the highscores json file and spawn the Highscores resource.
+ """
try:
with open(config.highscores_path) as f:
scores = Highscores.model_validate_json(f.read())
def save_highscores(config: Config, scores: Highscores) -> None:
+ """
+ Write the json highscores file from the Highscores resource.
+ """
path = config.highscores_path
with open(path, "w+") as f:
f.write(scores.model_dump_json(indent=4))
def create_maze(
world: World, assets: Assets[Sprite2D]
) -> Callable[[MazeGenerator], Entity]:
+ """
+ System creating and adding the maze to the world.
+ """
+
def inner(maze: MazeGenerator) -> Entity:
root_maze = Maze(len(maze.maze[0]), len(maze.maze))
cell_dims = Vector2(64, 64)
def spawn_superpacgums(
world: World, mazes: Query[Maze], assets: Assets[Sprite2D]
) -> None:
+ """
+ System creating and adding super pacgums to the maze.
+ """
pacgums = []
for maze in mazes:
for pos in [
superpacgums: Query[ChildOf, SuperPacGum],
assets: Assets[Sprite2D],
) -> None:
+ """
+ System creating and adding pacgums to the maze.
+ """
ignore = {e.parent for e in superpacgums}
pacgums = []
for ent, cell in cells:
def setup_maze(world: World) -> None:
+ """
+ System running the 'create_maze' system with a give MazeGenerator.
+ """
world.run_system(create_maze, MazeGenerator((14, 14)))
entities: Query[tuple[InMotion, Coord2D, MoveSpeed]],
positions: Query[Coord2D],
) -> None:
+ """
+ System enabling movement operation for entity wit InMotion component.
+ """
for motion, pos, speed in entities:
if (dst := positions.get(motion.pos)) is None:
continue
entities: Query[tuple[InMotion, Coord2D, TilePos]],
cells: Query[Coord2D],
) -> None:
+ """
+ System updating the tile position for entity wit InMotion component.
+ """
for motion, coord, pos in entities:
src = cells[motion.src]
dst = cells[motion.pos]
pos.pos = motion.pos
-def move_to_target(
- entities: Query[tuple[TargetTile, InMotion, Coord2D, GhostStatus]],
- cells: Query[tuple[Cell, Coord2D]],
- maze_q: Query[Maze],
-) -> None:
- if (maze := maze_q.single()) is None:
- return
- for target, motion, pos, status in entities:
- if status.state in (GhostState.dead, GhostState.respawning):
- continue
- cell, cell_pos = cells[motion.pos]
- if cell_pos.xy.distance_squared_to(pos.xy) > 0.01:
- continue
- candidates = []
- for direction, wall in enumerate(cell.walls()):
- if wall:
- continue
- dst_next = cell.neighbor(direction)
- dst_ent = maze._cells[dst_next]
- candidates.append((dst_ent, direction))
- candidates.sort(
- key=lambda e: (
- 2 << 32
- if e[0] is motion.src
- else Vector2(cells[e[0]][0].grid_pos).distance_squared_to(
- target.pos
- )
- ),
- )
- if status.state is GhostState.flee:
- lst = candidates[:-1]
- shuffle(lst)
- candidates = lst + [candidates[-1]]
- nxt, nxt_dir = candidates[0]
- motion.src = motion.pos
- motion.pos = nxt
- motion.direction = nxt_dir
-
-
def spawn_pacman(
world: World,
assets: Assets[Sprite2D],
cells: Query[Cell],
config: Config,
) -> None:
+ """
+ System creating and adding pacman to the maze.
+ """
if (maze_t := maze_q.single()) is None:
return
maze, maze_ent = maze_t
maze_q: Query[tuple[Maze, Entity]],
config: Config,
) -> None:
+ """
+ System creating and adding ghosts to the maze.
+ """
if (maze_t := maze_q.single()) is None:
return
maze, maze_ent = maze_t
cells: Query[tuple[Cell, Coord2D]],
maze_q: Query[Maze],
) -> None:
+ """
+ System giving target movement to pacman with the input given by the user.
+ """
if (maze := maze_q.single()) is None:
return
for coord, motion, pos in pacman_q:
score: GameState,
config: Config,
) -> None:
+ """
+ System managing the collision and collection of pacgum by pacman.
+ """
collided = [
pacgum
for coord, pacgum in pacgums
score: GameState,
config: Config,
) -> None:
+ """
+ System managing the collision and collection of superpacgum by pacman and
+ starting the ghost flee state if a superpacgum is collected.
+ """
collided = [
pacgum
for pacgum_pos, pacgum in pacgums
def respawn_pacman(pacmans: Query[tuple[Pacman, Coord2D, InMotion]]) -> None:
+ """
+ System reseting pacman position when touched by ghost.
+ """
for pacman, pos, motion in pacmans:
pos.xy = Vector2(*pacman.home).elementwise() * 64
motion.pos = pacman.home_entity
def respawn_ghost(ghosts: Query[tuple[Ghost, Coord2D, InMotion]]) -> None:
+ """
+ System reseting ghost fleeing position when touched by pacman.
+ """
for ghost, pos, motion in ghosts:
pos.xy = Vector2(*ghost.home).elementwise() * 64
motion.pos = ghost.home_entity
state: GameState,
config: Config,
) -> None:
+ """
+ System managing the collision between by pacman and ghost.
+ """
for pacman_pos in pacmans:
for ghost_pos, status, ghost, motion, pos in ghosts:
if (
pass
+def move_to_target(
+ entities: Query[tuple[TargetTile, InMotion, Coord2D, GhostStatus]],
+ cells: Query[tuple[Cell, Coord2D]],
+ maze_q: Query[Maze],
+) -> None:
+ """
+ System giving target movement to ghost.
+ """
+ if (maze := maze_q.single()) is None:
+ return
+ for target, motion, pos, status in entities:
+ if status.state in (GhostState.dead, GhostState.respawning):
+ continue
+ cell, cell_pos = cells[motion.pos]
+ if cell_pos.xy.distance_squared_to(pos.xy) > 0.01:
+ continue
+ candidates = []
+ for direction, wall in enumerate(cell.walls()):
+ if wall:
+ continue
+ dst_next = cell.neighbor(direction)
+ dst_ent = maze._cells[dst_next]
+ candidates.append((dst_ent, direction))
+ candidates.sort(
+ key=lambda e: (
+ 2 << 32
+ if e[0] is motion.src
+ else Vector2(cells[e[0]][0].grid_pos).distance_squared_to(
+ target.pos
+ )
+ ),
+ )
+ if status.state is GhostState.flee:
+ lst = candidates[:-1]
+ shuffle(lst)
+ candidates = lst + [candidates[-1]]
+ nxt, nxt_dir = candidates[0]
+ motion.src = motion.pos
+ motion.pos = nxt
+ motion.direction = nxt_dir
+
+
def update_ghost_targets(
ghosts: Query[tuple[Ghost, TilePos, TargetTile, GhostStatus]],
pacman_q: Query[tuple[InMotion, TilePos], Pacman],
cells: Query[Cell],
) -> None:
+ """
+ System updating target to ghost.
+ """
player = pacman_q.single()
if player is None:
return
def update_chase_status(status_q: Query[GhostStatus], config: Config) -> None:
+ """
+ System updating ghost status and cycle.
+ """
for status in status_q:
target_time: int
next_state: GhostState
def update_ghost_speed(
ghost: Query[tuple[MoveSpeed, GhostStatus]], config: Config
) -> None:
+ """
+ System updating ghost speed according to his state.
+ """
for speed, status in ghost:
match status.state:
case GhostState.chase | GhostState.wander:
def update_pacman_anim(
pacman_q: Query[tuple[InMotion, AnimationSet], Pacman],
) -> None:
+ """
+ System updating pacman animation according to his direction.
+ """
for motion, anim in pacman_q:
anim.idx = motion.direction
ghost_q: Query[tuple[InMotion, AnimationSet, GhostStatus], Ghost],
config: Config,
) -> None:
+ """
+ System updating pacman animation according to his direction and state.
+ """
for motion, anim, status in ghost_q:
match status.state:
case GhostState.chase | GhostState.wander:
def start_pause_timer(timer: "InitialPauseTimer") -> None:
+ """
+ System starting the initial pause timer.
+ """
timer.tick = 0
def setup_new_maze(world: World) -> None:
+ """
+ System setting up the maze by running 'setup_maze', 'spawn_superpacgums',
+ 'spawn_pacgums', 'spawn_pacman', 'spawn_ghosts' and
+ 'start_pause_timer' systems.
+ """
for system in (
setup_maze,
spawn_superpacgums,
def maze_exists(itr: Query[Maze]) -> bool:
+ """
+ System cheking maze existance.
+ """
return next(itr.__iter__(), None) is not None
def in_game(world: World) -> bool:
+ """
+ System cheking that the game is running.
+ """
return world.run_raw_system(maze_exists) and not world.run_raw_system(
in_game_menu
)
def level_is_finished(
pacgums: Query[SuperPacGum | PacGum], _: Not[GameIsOver]
) -> bool:
+ """
+ System cheking the level completion.
+ """
return next(iter(pacgums), None) is None
queue: CommandQueue,
state: GameState,
) -> None:
+ """
+ System called when level is completed, generating a new level or finishing
+ the game if all levels are completed.
+ """
for maze in mazes:
queue.queue(hierarchy.despawn_with_children(maze))
if state.level == 2:
def level_is_lost(
config: Config, state: GameState, _: Not[GameIsOver]
) -> bool:
+ """
+ System cheking the level is lost.
+ """
return config.level_timer < state.timer or state.lives <= 0
mazes: Query[Entity, Maze],
queue: CommandQueue,
) -> None:
+ """
+ System called when the player lose the game.
+ """
for maze in mazes:
queue.queue(hierarchy.despawn_with_children(maze))
queue.queue(lambda world: world.add_raw_system(game_over)())
def game_over(queue: CommandQueue) -> None:
+ # FIXME
+ """
+ System called when game is finish.
+ """
queue.queue(lambda world: world.spawn_resource(GameIsOver()))
print("Game over")
def update_game_timer(state: GameState) -> None:
+ """
+ System updating the levels timer.
+ """
state.timer += 1
def initial_pause_done(timer: InitialPauseTimer, config: Config) -> bool:
+ """
+ System cheking if the inital timer is done.
+ """
if timer.tick < config.initial_pos_timer:
timer.tick += 1
return False
def print_score(state: GameState) -> str:
+ """
+ Return a string with the player score.
+ """
return f"Score: {state.score}"
def print_lives(state: GameState) -> str:
+ """
+ Return a string with the player lives.
+ """
return f"Lives: {state.lives}"
def print_timer(state: GameState, config: Config) -> str:
+ """
+ Return a string with the level timer.
+ """
time = config.level_timer - state.timer
return f"Time left: {time // 60}"
def spawn_hud(world: World, fonts: Assets[Font]) -> None:
+ """
+ System that print the Heads-Up Display in game with lives, levels and timer
+ informations.
+ """
world.spawn_many(
tuple[
WindowPositionned,
hierarchy: Hierarchy[tuple[()]],
menus: Query[Entity, MainMenu],
) -> None:
+ """
+ System closing main menu and starting the game and Heads-Up Display.
+ """
for menu in menus:
hierarchy.despawn_with_children(menu)(world)
world.despawn_resource(SelectedButton)
def quit_game(world: World) -> None:
+ """
+ System closing the game and quiting the program.
+ """
world.spawn_resource(WorldShouldExit())
def spawn_button(
world: World, fonts: Assets[Font], sprites: Assets[Sprite2D]
) -> Callable[[WindowPositionned, str, System], Entity]:
+ """
+ System creating and adding a button to the world.
+ """
+
def inner(pos: WindowPositionned, s: str, cb: System) -> Entity:
parent = world.spawn(
tuple[
gui: Query[Entity, GameGui],
queue: CommandQueue,
) -> None:
+ """
+ System closing the current game and opening the main menu.
+ """
world.run_raw_system(close_game_menu)
world.despawn_resource(GameState)
world.despawn_many(list(gui))
def in_game_menu(menus: Query[GameMenu]) -> bool:
+ """
+ System cheking if the main menu is running.
+ """
return next(menus.__iter__(), None) is not None
def toggle_game_menu(world: World) -> None:
+ """
+ System opening or closing the game menu.
+ """
if world.run_raw_system(in_game_menu):
world.run_raw_system(close_game_menu)
else:
menus: Query[Entity, GameMenu],
animated: Query[Entity, Pacman | Ghost],
) -> None:
+ """
+ System closing the game menu.
+ """
world.remove_many(StopAnimation, list(animated))
for menu in menus:
hierarchy.despawn_with_children(menu)(world)
def setup_game_menu(
world: World, animated: Query[Entity, Pacman | Ghost]
) -> None:
+ """
+ System opening the game menu.
+ """
world.insert_many(StopAnimation, [(e, StopAnimation()) for e in animated])
selected = world.run_system(
spawn_button,
def set_up_instructions(world: World, fonts: Assets[Font]) -> list[Entity]:
+ """
+ System adding instructions to main menu.
+ """
text = """Use 'WASD'to move pacman in the maze.
Collect all pacgums to go to the next level.
Complete all levels to win the game.
def set_up_highsores(
world: World, scores: Highscores, fonts: Assets[Font]
) -> list[Entity]:
+ """
+ System adding highscores to main menu.
+ """
highscores = [(name, value) for name, value in scores.scores.items()]
highscores.sort(key=lambda x: x[1], reverse=True)
entities = [
def setup_main_menu(world: World) -> None:
-
+ """
+ System creating the main menu by adding button, instruction and highscores.
+ """
selected = world.run_system(
spawn_button,
WindowPositionned(Vector2(0.5), Vector2(-150, -100)),
def run_game() -> None:
+ """
+ Start with all necessary plugins and system to run the game.
+ """
World().with_plugins(graphics_plugins, minimal_plugins).with_systems(
StartupSchedule,
Systems(load_config, load_highscores, setup_main_menu)
projects / axy / ft / pacman.git / commitdiff
