from typing import override
from mazegenerator import MazeGenerator # type: ignore[import-untyped]
+from pydantic import BaseModel
from pygame import K_a, K_d, K_s, K_w, Vector2
from pygame.math import clamp
CommandQueue,
Component,
Entity,
+ Not,
Query,
Resource,
Systems,
@dataclass
-class Score(Resource):
+class InitialPauseTimer(Resource):
+ tick: int = 0
+
+ @classmethod
+ @override
+ def extract_default(
+ cls, ty: type, world: "World"
+ ) -> Callable[[], "InitialPauseTimer | None"]:
+ return InitialPauseTimer
+
+
+class Config(Resource, BaseModel):
+ ghost_chase_speed: float = 2.5
+ ghost_flee_speed: float = 1.5
+ pacman_speed: float = 3.0
+ initial_pos_timer: int = 60
+ level_timer: int = 60 * 180
+ initial_lives: int = 3
+ score_pacgum: int = 1
+ score_superpacgum: int = 5
+ score_ghost: int = 20
+ superpacgum_duration: int = 60 * 8
+
+
+class GameIsOver(Resource):
+ pass
+
+
+@dataclass
+class GameState(Resource):
score: int = 0
+ level: int = 0
+ timer: int = 0
+ lives: int = 3
@dataclass
assets: Assets[Sprite2D],
maze_q: Query[tuple[Maze, Entity]],
cells: Query[Cell],
+ config: Config,
) -> None:
if (maze_t := maze_q.single()) is None:
return
Pacman(pos, maze._cells[pos]),
ShouldRender(),
InMotion(maze._cells[(pos)], maze._cells[(pos)], 1),
- MoveSpeed(3.0),
+ MoveSpeed(config.pacman_speed),
TilePos(maze._cells[(pos)]),
ChildOf(maze_ent),
),
def spawn_ghosts(
- world: World, assets: Assets[Sprite2D], maze_q: Query[tuple[Maze, Entity]]
+ world: World,
+ assets: Assets[Sprite2D],
+ maze_q: Query[tuple[Maze, Entity]],
+ config: Config,
) -> None:
if (maze_t := maze_q.single()) is None:
return
]
for direction in "URDL"
]
- + [[assets.load("./asset/enemy/ghost_blue.png")]],
+ + [
+ [assets.load("./asset/enemy/ghost_blue.png")],
+ [assets.load("./asset/enemy/ghost_dead.png")],
+ [
+ assets.load("./asset/enemy/ghost_flash.png"),
+ assets.load("./asset/enemy/ghost_blue.png"),
+ ],
+ ],
6,
),
ShouldRender(),
InMotion(maze._cells[pos], maze._cells[pos], 1),
- MoveSpeed(2.5),
+ MoveSpeed(config.ghost_chase_speed),
TilePos(maze._cells[pos]),
Ghost(variant, pos, maze._cells[pos]),
TargetTile(pos),
world: World,
pacgums: Query[tuple[CoordAbs2D, Entity], PacGum],
pacmans: Query[CoordAbs2D, Pacman],
- score: Score,
+ score: GameState,
+ config: Config,
) -> None:
collided = [
pacgum
for pos in pacmans
if pos._vec.xy.distance_squared_to(coord._vec.xy) <= 256
]
- score.score += len(collided)
+ score.score += len(collided * config.score_pacgum)
world.despawn_many(collided)
pacgums: Query[tuple[CoordAbs2D, Entity], SuperPacGum],
pacmans: Query[CoordAbs2D, Pacman],
ghosts: Query[tuple[InMotion, GhostStatus], Ghost],
- score: Score,
+ score: GameState,
+ config: Config,
) -> None:
collided = [
pacgum
for pacman_pos in pacmans
if pacman_pos._vec.xy.distance_squared_to(pacgum_pos._vec.xy) <= 256
]
- score.score += len(collided * 10)
+ score.score += len(collided * config.score_superpacgum)
if len(collided) == 0:
return
for ghost, status in ghosts:
tuple[CoordAbs2D, GhostStatus, Ghost, InMotion, Coord2D],
Ghost,
],
- score: Score,
+ score: GameState,
world: World,
+ state: GameState,
+ config: Config,
) -> None:
for pacman_pos in pacmans:
for ghost_pos, status, ghost, motion, pos in ghosts:
if (
pacman_pos._vec.xy.distance_squared_to(ghost_pos._vec.xy)
- > 3600
+ > 48 * 48
):
continue
match status.state:
world.add_raw_system(respawn_pacman)()
world.add_raw_system(respawn_ghost)()
world.add_raw_system(start_pause_timer)()
+ state.lives -= 1
+ break
case GhostState.flee:
- score.score += 100 # FIXME
+ score.score += config.score_ghost
status.state = GhostState.dead
status.tick = 0
motion.src = ghost.home_entity
pass
-def level_is_finished(pacgums: Query[SuperPacGum | PacGum]) -> bool:
- return next(iter(pacgums), None) is None
-
-
def update_ghost_targets(
ghosts: Query[tuple[Ghost, TilePos, TargetTile, GhostStatus]],
pacman_q: Query[tuple[InMotion, TilePos], Pacman],
pass
-def update_chase_status(status_q: Query[GhostStatus]) -> None:
+def update_chase_status(status_q: Query[GhostStatus], config: Config) -> None:
for status in status_q:
target_time: int
next_state: GhostState
cycle_wander_next: int = status.cycle_wander
match status.state:
case GhostState.flee:
- target_time, next_state = (8 * 60, GhostState.wander)
+ target_time, next_state = (
+ config.superpacgum_duration,
+ GhostState.wander,
+ )
case GhostState.chase:
if cycle_wander_next > 60 * 4:
cycle_wander_next -= 120
status.cycle_wander = cycle_wander_next
-def update_ghost_speed(ghost: Query[tuple[MoveSpeed, GhostStatus]]) -> None:
+def update_ghost_speed(
+ ghost: Query[tuple[MoveSpeed, GhostStatus]], config: Config
+) -> None:
for speed, status in ghost:
match status.state:
case GhostState.chase | GhostState.wander:
- speed.pixels_per_frame = 2.5
+ speed.pixels_per_frame = config.ghost_chase_speed
case GhostState.flee:
- speed.pixels_per_frame = 1.5
+ speed.pixels_per_frame = config.ghost_flee_speed
def update_pacman_anim(
def update_ghost_anim(
ghost_q: Query[tuple[InMotion, AnimationSet, GhostStatus], Ghost],
+ config: Config,
) -> None:
for motion, anim, status in ghost_q:
match status.state:
case GhostState.chase | GhostState.wander:
anim.idx = motion.direction
- case GhostState.flee:
+ case GhostState.flee if (
+ config.superpacgum_duration - 2.5 * 60 > status.tick
+ ):
anim.idx = 4
- case _:
- pass
-
-
-def setup_resources(world: World) -> None:
- world.spawn_resource(Inputs({}))
- world.spawn_resource(Score())
+ case GhostState.dead:
+ anim.idx = 5
+ case GhostState.respawning | GhostState.flee:
+ anim.idx = 6
def start_pause_timer(timer: "InitialPauseTimer") -> None:
del inputs.inputs[key]
+def level_is_finished(
+ pacgums: Query[SuperPacGum | PacGum], _: Not[GameIsOver]
+) -> bool:
+ return next(iter(pacgums), None) is None
+
+
def victory(
hierarchy: Hierarchy[tuple[()]],
mazes: Query[Entity, Maze],
queue: CommandQueue,
+ state: GameState,
) -> None:
for maze in mazes:
queue.queue(hierarchy.despawn_with_children(maze))
+ if state.level == 2:
+ queue.queue(lambda world: world.add_raw_system(game_over)())
+ return
+ state.level += 1
+ state.lives += 1
+ state.timer = 0
queue.queue(setup_new_maze)
-@dataclass
-class InitialPauseTimer(Resource):
- tick: int = 0
+def level_is_lost(
+ config: Config, state: GameState, _: Not[GameIsOver]
+) -> bool:
+ return config.level_timer < state.timer or state.lives <= 0
- @classmethod
- @override
- def extract_default(
- cls, ty: type, world: "World"
- ) -> Callable[[], "InitialPauseTimer | None"]:
- return InitialPauseTimer
+
+def lose(
+ hierarchy: Hierarchy[tuple[()]],
+ mazes: Query[Entity, Maze],
+ queue: CommandQueue,
+) -> None:
+ for maze in mazes:
+ queue.queue(hierarchy.despawn_with_children(maze))
+ queue.queue(lambda world: world.add_raw_system(game_over)())
-def initial_pause_done(timer: InitialPauseTimer) -> bool:
- if timer.tick < 60:
+def game_over(queue: CommandQueue) -> None:
+ queue.queue(lambda world: world.spawn_resource(GameIsOver()))
+ print("Game over")
+
+
+def update_game_timer(state: GameState) -> None:
+ state.timer += 1
+
+
+def initial_pause_done(timer: InitialPauseTimer, config: Config) -> bool:
+ if timer.tick < config.initial_pos_timer:
timer.tick += 1
return False
return True
+def setup_resources(world: World) -> None:
+ world.spawn_resource(Inputs({}))
+ world.spawn_resource(GameState())
+ world.spawn_resource(Config())
+
+
def run_game() -> None:
World().with_plugins(graphics_plugins, minimal_plugins).with_systems(
StartupSchedule,
move_in_motion,
update_chase_status,
update_ghost_speed,
+ update_game_timer,
)
.chain()
.run_if(initial_pause_done)
.after(PropagateTransform)
.before(GraphicsSystems),
Systems(victory).run_if(level_is_finished),
+ Systems(lose).run_if(level_is_lost),
).run_main()
projects / axy / ft / pacman.git / commitdiff
