262 lines
13 KiB
Python
262 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import math
|
|
from dataclasses import dataclass, field
|
|
|
|
from engine.character import Character
|
|
from engine.entity import Entity, EntityPosition
|
|
from engine.item import ItemInstance
|
|
from engine.map import Map
|
|
|
|
LEAP_DISTANCE = 2
|
|
|
|
|
|
@dataclass
|
|
class TimeWarpZone:
|
|
"""A timed spherical zone that slows everything inside it except a whitelist of entities.
|
|
|
|
Generic on purpose: size (radius), speed_factor, whitelist (immune_entity_ids), and time
|
|
(expires_at, an absolute GameClock.total_time) are all just constructor args - nothing here
|
|
assumes it was cast by a character ability vs. triggered by an item (see
|
|
resources/logic/abilities.py's cast_time_warp_sphere for both attachment points).
|
|
"""
|
|
|
|
map_id: str
|
|
center: tuple[float, float, float]
|
|
radius: float
|
|
speed_factor: float
|
|
expires_at: float
|
|
immune_entity_ids: set[str] = field(default_factory=set)
|
|
|
|
def contains(self, map_id: str, x: float, y: float, z: float) -> bool:
|
|
if map_id != self.map_id or math.floor(z) != math.floor(self.center[2]):
|
|
return False
|
|
dx, dy = x - self.center[0], y - self.center[1]
|
|
return math.hypot(dx, dy) <= self.radius
|
|
|
|
def affects(self, entity_id: str) -> bool:
|
|
return entity_id not in self.immune_entity_ids
|
|
|
|
|
|
class World:
|
|
"""Owns the map graph, all entities, and the def registry; drives movement."""
|
|
|
|
def __init__(self, registry=None):
|
|
self.registry = registry
|
|
self.maps: dict[str, Map] = {}
|
|
self.entities: dict[str, Entity] = {}
|
|
self.player: Entity | None = None
|
|
self.time_warp_zones: list[TimeWarpZone] = []
|
|
self._pending_fall_distance: dict[str, int] = {}
|
|
self.ground_items: dict[tuple[str, int, int, int], list[ItemInstance]] = {} # (map_id,
|
|
# x, y, z) -> items dropped/lying there - see drop_item_at/ground_items_at/
|
|
# remove_ground_item and resources/logic/pickup.py for the actual pickup/drop logic.
|
|
# In-memory only for now: not persisted through WorldRepository save/load.
|
|
|
|
def consume_fall_distance(self, entity_id: str) -> int:
|
|
"""How many z-levels _apply_gravity just dropped this entity, if any - 0 otherwise.
|
|
One-shot, same "consume_*" idiom as engine/input.py's pending actions: main.py calls
|
|
this right after a move to decide whether to roll fall damage (see
|
|
resources/logic/falling.py), without World itself depending on that game-logic layer.
|
|
"""
|
|
return self._pending_fall_distance.pop(entity_id, 0)
|
|
|
|
def drop_item_at(self, map_id: str, x: float, y: float, z: float, item: ItemInstance) -> None:
|
|
key = (map_id, math.floor(x), math.floor(y), math.floor(z))
|
|
self.ground_items.setdefault(key, []).append(item)
|
|
|
|
def ground_items_at(self, map_id: str, x: float, y: float, z: float) -> list[ItemInstance]:
|
|
return list(self.ground_items.get((map_id, math.floor(x), math.floor(y), math.floor(z)), []))
|
|
|
|
def remove_ground_item(self, map_id: str, x: float, y: float, z: float, item_id: str) -> ItemInstance | None:
|
|
key = (map_id, math.floor(x), math.floor(y), math.floor(z))
|
|
items = self.ground_items.get(key)
|
|
if not items:
|
|
return None
|
|
for i, candidate in enumerate(items):
|
|
if candidate.item_id == item_id:
|
|
removed = items.pop(i)
|
|
if not items:
|
|
del self.ground_items[key]
|
|
return removed
|
|
return None
|
|
|
|
def speed_multiplier_at(self, entity_id: str, map_id: str, x: float, y: float, z: float, now: float) -> float:
|
|
"""Combined slow from the terrain underfoot (e.g. mud/water - see Tile.speed_multiplier)
|
|
and every active, non-expired time-warp zone that affects this entity here.
|
|
"""
|
|
multiplier = self.maps[map_id].get_tile(x, y, z).speed_multiplier(self.registry)
|
|
for zone in self.time_warp_zones:
|
|
if zone.expires_at > now and zone.affects(entity_id) and zone.contains(map_id, x, y, z):
|
|
multiplier *= zone.speed_factor
|
|
return multiplier
|
|
|
|
def expire_time_warp_zones(self, now: float) -> None:
|
|
self.time_warp_zones = [z for z in self.time_warp_zones if z.expires_at > now]
|
|
|
|
def add_map(self, map_: Map) -> None:
|
|
self.maps[map_.map_id] = map_
|
|
|
|
def add_entity(self, entity: Entity) -> None:
|
|
self.entities[entity.entity_id] = entity
|
|
|
|
def entities_on_map(self, map_id: str) -> list[Entity]:
|
|
return [e for e in self.entities.values() if e.position is not None and e.position.map_id == map_id]
|
|
|
|
def visible_embeddings(self, m: Map) -> list:
|
|
"""`m`'s embeddings the player is actually aboard right now - i.e. whose child map is
|
|
the player's own current map. Rendering only descends into these: an embedded map (e.g.
|
|
a ship's interior) is otherwise invisible from outside it, so standing off-ship never
|
|
exposes its submap. No-op (empty) with no player positioned anywhere yet.
|
|
"""
|
|
if self.player is None or self.player.position is None:
|
|
return []
|
|
return [e for e in m.embeddings if e.child_map.map_id == self.player.position.map_id]
|
|
|
|
def entity_at(self, map_id: str, x: int, y: int, z: int) -> Entity | None:
|
|
"""Tile-membership lookup: matches any entity whose (possibly continuous) position
|
|
falls in tile (x, y, z), i.e. floor(pos.x) == x etc. Floor of an int is itself, so this
|
|
is unchanged for entities that happen to sit exactly on a tile.
|
|
"""
|
|
for entity in self.entities.values():
|
|
pos = entity.position
|
|
if pos is None or pos.map_id != map_id:
|
|
continue
|
|
if (math.floor(pos.x), math.floor(pos.y), math.floor(pos.z)) == (x, y, z):
|
|
return entity
|
|
return None
|
|
|
|
def try_move(self, entity: Entity, dx: int, dy: int, dz: int) -> bool:
|
|
assert entity.position is not None
|
|
cur_map = self.maps[entity.position.map_id]
|
|
tx = entity.position.x + dx
|
|
ty = entity.position.y + dy
|
|
tz = entity.position.z + dz
|
|
moved = self._move_within(entity, cur_map, tx, ty, tz)
|
|
if moved:
|
|
self._apply_staircase(entity)
|
|
self._apply_gravity(entity)
|
|
return moved
|
|
|
|
def try_leap(self, entity: Entity, dx: int, dy: int, dz: int, distance: int = LEAP_DISTANCE) -> bool:
|
|
"""A leap in the given unit direction, e.g. to clear a gap and land aboard a ship.
|
|
|
|
Reuses try_move's own semantics unchanged: it only ever validates the landing tile,
|
|
never anything in between, so scaling the delta is all a "jump over a gap" needs.
|
|
|
|
If the entity's species defines a "leap" ability (e.g. requiring both legs), it's
|
|
gated on that; species/entities with no such ability defined are left unrestricted.
|
|
"""
|
|
if isinstance(entity, Character) and entity.species_id is not None and self.registry is not None:
|
|
species = self.registry.get_species_def(entity.species_id)
|
|
if any(a.id == "leap" for a in species.abilities) and not entity.can_perform_ability("leap", self.registry):
|
|
return False
|
|
return self.try_move(entity, dx * distance, dy * distance, dz * distance)
|
|
|
|
def _move_within(self, entity: Entity, map_: Map, x: int, y: int, z: int) -> bool:
|
|
if map_.in_bounds(x, y, z):
|
|
embedding = map_.embedding_at(x, y, z)
|
|
if embedding is not None:
|
|
lx, ly, lz = embedding.parent_to_local(x, y, z)
|
|
if not embedding.child_map.get_tile(lx, ly, lz).is_walkable(self.registry):
|
|
return False
|
|
entity.position = EntityPosition(embedding.child_map.map_id, lx, ly, lz)
|
|
return True
|
|
if not map_.get_tile(x, y, z).is_walkable(self.registry):
|
|
return False
|
|
entity.position = EntityPosition(map_.map_id, x, y, z)
|
|
return True
|
|
|
|
if map_.parent_embedding is not None:
|
|
embedding = map_.parent_embedding
|
|
px, py, pz = embedding.local_to_parent(x, y, z)
|
|
return self._move_within(entity, embedding.parent_map, px, py, pz)
|
|
|
|
return False
|
|
|
|
def try_move_continuous(self, entity: Entity, dx: float, dy: float, dz: float, distance: float) -> bool:
|
|
"""True stepless movement: moves `entity` by `distance` along the (dx, dy, dz) direction
|
|
(need not be a unit vector - it's normalized here), landing at a fractional position
|
|
rather than snapping to a tile. Collision is checked against whichever tile the
|
|
destination falls in (floor(x), floor(y)) - same walkability/embedding rules as
|
|
try_move, just point-based rather than a full sweep, and the path between old and new
|
|
position isn't swept for obstacles (a fast-enough mover could tunnel a thin wall in one
|
|
frame - a known simplification, not a concern at ordinary per-frame speeds/dt).
|
|
"""
|
|
assert entity.position is not None
|
|
length = math.hypot(dx, dy)
|
|
if length == 0:
|
|
return False
|
|
ux, uy = dx / length, dy / length
|
|
nx = entity.position.x + ux * distance
|
|
ny = entity.position.y + uy * distance
|
|
nz = entity.position.z
|
|
moved = self._move_within_continuous(entity, self.maps[entity.position.map_id], nx, ny, nz)
|
|
if moved:
|
|
self._apply_staircase(entity)
|
|
self._apply_gravity(entity)
|
|
return moved
|
|
|
|
def _apply_staircase(self, entity: Entity) -> None:
|
|
"""A single z-shift if the entity just landed on a staircase/ladder tile, provided the
|
|
target floor is in bounds and walkable at the same (x, y). Not recursive - a stack of
|
|
staircases each carries you up/down exactly one level per step onto them.
|
|
"""
|
|
assert entity.position is not None
|
|
map_ = self.maps[entity.position.map_id]
|
|
x, y, z = entity.position.x, entity.position.y, entity.position.z
|
|
direction = map_.get_tile(x, y, z).staircase_direction(self.registry)
|
|
if direction is None:
|
|
return
|
|
nz = math.floor(z) + (1 if direction == "up" else -1)
|
|
if not map_.in_bounds(x, y, nz):
|
|
return
|
|
if not map_.get_tile(x, y, nz).is_walkable(self.registry):
|
|
return
|
|
entity.position = EntityPosition(map_.map_id, x, y, nz)
|
|
|
|
def _apply_gravity(self, entity: Entity) -> None:
|
|
"""Drops an entity straight down to the first walkable, floored tile below, if the
|
|
tile they're now on has no floor at all - open air, e.g. stepping off a cliff steeper
|
|
than any placed slope/cave connector (see engine/worldgen/terrain.py: everywhere else
|
|
in this engine's hand-authored maps, every tile gets a flooring layer by default, so
|
|
this only ever fires for that generated-terrain gap). A single relocation, not a
|
|
physics simulation - same "instant, not animated" spirit as _apply_staircase.
|
|
"""
|
|
assert entity.position is not None
|
|
map_ = self.maps[entity.position.map_id]
|
|
x, y, z = entity.position.x, entity.position.y, entity.position.z
|
|
if map_.get_tile(x, y, z).flooring is not None:
|
|
return
|
|
start_z = math.floor(z)
|
|
for nz in range(start_z - 1, -1, -1):
|
|
candidate = map_.get_tile(x, y, nz)
|
|
if candidate.flooring is not None and candidate.is_walkable(self.registry):
|
|
entity.position = EntityPosition(map_.map_id, x, y, nz)
|
|
self._pending_fall_distance[entity.entity_id] = start_z - nz
|
|
return
|
|
|
|
def _move_within_continuous(self, entity: Entity, map_: Map, x: float, y: float, z: float) -> bool:
|
|
tile_x, tile_y, tile_z = math.floor(x), math.floor(y), math.floor(z)
|
|
if map_.in_bounds(tile_x, tile_y, tile_z):
|
|
embedding = map_.embedding_at(tile_x, tile_y, tile_z)
|
|
if embedding is not None:
|
|
local_tile = embedding.parent_to_local(tile_x, tile_y, tile_z)
|
|
assert local_tile is not None
|
|
if not embedding.child_map.get_tile(*local_tile).is_walkable(self.registry):
|
|
return False
|
|
lx, ly, lz = embedding.parent_to_local_pos(x, y, z)
|
|
entity.position = EntityPosition(embedding.child_map.map_id, lx, ly, lz)
|
|
return True
|
|
if not map_.get_tile(tile_x, tile_y, tile_z).is_walkable(self.registry):
|
|
return False
|
|
entity.position = EntityPosition(map_.map_id, x, y, z)
|
|
return True
|
|
|
|
if map_.parent_embedding is not None:
|
|
embedding = map_.parent_embedding
|
|
px, py, pz = embedding.local_to_parent_pos(x, y, z)
|
|
return self._move_within_continuous(entity, embedding.parent_map, px, py, pz)
|
|
|
|
return False
|