Spaces:
Paused
Paused
File size: 7,412 Bytes
6343479 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 | """Deterministic Chunk-Streamed Endless World Engine for FlyBrain V8/V9.
Implements Sections 41–45:
- WorldManager
- ChunkManager
- WorldGenerator (terrain, biomes, procedural structures)
- Bounded RAM streaming with persistent modification overlays
- Deterministic formula: hash(world_seed, cx, cy, generator_revision)
"""
import os
import math
import json
import hashlib
import numpy as np
from typing import Dict, Any, List, Tuple, Optional
class BiomeType:
TEMPERATE_VALLEY = "temperate_valley"
PINE_FOREST = "pine_forest"
RIVER_BASIN = "river_basin"
ROCKY_HIGHLANDS = "rocky_highlands"
SAVANNA_PLAINS = "savanna_plains"
class Chunk:
def __init__(self, cx: int, cy: int, size_m: float = 32.0, resolution: int = 32):
self.cx = cx
self.cy = cy
self.size_m = size_m
self.resolution = resolution
self.biome: str = BiomeType.TEMPERATE_VALLEY
self.heightmap: np.ndarray = np.zeros((resolution, resolution), dtype=np.float32)
self.objects: List[Dict[str, Any]] = []
self.modifications: List[Dict[str, Any]] = []
self.is_loaded: bool = False
self.last_accessed: float = 0.0
@property
def key(self) -> str:
return f"{self.cx}_{self.cy}"
def to_dict(self) -> Dict[str, Any]:
return {
"cx": self.cx,
"cy": self.cy,
"size_m": self.size_m,
"biome": self.biome,
"height_min": float(self.heightmap.min()),
"height_max": float(self.heightmap.max()),
"object_count": len(self.objects),
"modifications": self.modifications
}
class WorldGenerator:
"""Deterministic terrain and structure synthesizer."""
def __init__(self, seed: int = 42, revision: str = "v8.0"):
self.seed = seed
self.revision = revision
def _hash_coords(self, cx: int, cy: int) -> int:
h = hashlib.sha256(f"{self.seed}_{cx}_{cy}_{self.revision}".encode("utf-8")).hexdigest()
return int(h[:8], 16)
def generate_chunk(self, cx: int, cy: int, chunk_size: float = 32.0) -> Chunk:
chunk = Chunk(cx, cy, size_m=chunk_size)
rng = np.random.RandomState(self._hash_coords(cx, cy))
# Biome determination from macro-coordinates
macro = math.sin(cx * 0.15) + math.cos(cy * 0.15)
if macro > 0.8:
chunk.biome = BiomeType.ROCKY_HIGHLANDS
base_h = 4.0
elif macro < -0.8:
chunk.biome = BiomeType.RIVER_BASIN
base_h = -1.5
elif macro > 0.2:
chunk.biome = BiomeType.PINE_FOREST
base_h = 1.0
else:
chunk.biome = BiomeType.TEMPERATE_VALLEY
base_h = 0.0
# Deterministic perlin-like smooth procedural terrain
res = chunk.resolution
x = np.linspace(cx * chunk_size, (cx + 1) * chunk_size, res)
y = np.linspace(cy * chunk_size, (cy + 1) * chunk_size, res)
xv, yv = np.meshgrid(x, y)
h = base_h + 1.2 * np.sin(xv * 0.08) * np.cos(yv * 0.08)
h += 0.4 * np.sin(xv * 0.2 + yv * 0.1)
chunk.heightmap = h.astype(np.float32)
# Procedural natural props (rocks, trees)
num_trees = rng.randint(2, 8) if chunk.biome == BiomeType.PINE_FOREST else rng.randint(0, 3)
for i in range(num_trees):
tx = float(rng.uniform(cx * chunk_size + 2.0, (cx + 1) * chunk_size - 2.0))
ty = float(rng.uniform(cy * chunk_size + 2.0, (cy + 1) * chunk_size - 2.0))
chunk.objects.append({
"id": f"tree_{cx}_{cy}_{i}",
"type": "vegetation",
"subtype": "pine_tree",
"pos": [round(tx, 2), round(ty, 2), round(float(np.interp(tx, x, h[:, 0])), 2)],
"scale": [1.0, 1.0, round(float(rng.uniform(2.5, 4.5)), 2)]
})
chunk.is_loaded = True
return chunk
class ChunkManager:
"""Manages streaming radius, bounded RAM cache, and persistent delta overlays."""
def __init__(self, world_gen: WorldGenerator, max_loaded_chunks: int = 25, storage_dir: str = "data/world_chunks"):
self.generator = world_gen
self.max_loaded = max_loaded_chunks
self.storage_dir = storage_dir
os.makedirs(self.storage_dir, exist_ok=True)
self.loaded_chunks: Dict[str, Chunk] = {}
self.modifications_db: Dict[str, List[Dict[str, Any]]] = {}
def update_center(self, agent_x: float, agent_y: float, radius_chunks: int = 2) -> List[str]:
"""Loads chunks within radius and streams out far chunks to maintain bounded RAM."""
import time
center_cx = int(math.floor(agent_x / 32.0))
center_cy = int(math.floor(agent_y / 32.0))
needed_keys = set()
for dx in range(-radius_chunks, radius_chunks + 1):
for dy in range(-radius_chunks, radius_chunks + 1):
cx = center_cx + dx
cy = center_cy + dy
k = f"{cx}_{cy}"
needed_keys.add(k)
if k not in self.loaded_chunks:
chunk = self.generator.generate_chunk(cx, cy)
# Apply persistent modifications if existing
if k in self.modifications_db:
chunk.modifications.extend(self.modifications_db[k])
chunk.last_accessed = time.time()
self.loaded_chunks[k] = chunk
else:
self.loaded_chunks[k].last_accessed = time.time()
# Evict unneeded chunks beyond max limit
while len(self.loaded_chunks) > self.max_loaded:
# Evict LRU chunk that is not in needed_keys
candidates = [k for k in self.loaded_chunks if k not in needed_keys]
if not candidates:
break
lru_key = min(candidates, key=lambda k: self.loaded_chunks[k].last_accessed)
self._save_and_unload(lru_key)
return list(self.loaded_chunks.keys())
def apply_modification(self, cx: int, cy: int, mod: Dict[str, Any]) -> None:
"""Applies persistent world modification (e.g. built structure, dug hole, placed asset)."""
k = f"{cx}_{cy}"
if k not in self.modifications_db:
self.modifications_db[k] = []
self.modifications_db[k].append(mod)
if k in self.loaded_chunks:
self.loaded_chunks[k].modifications.append(mod)
def _save_and_unload(self, key: str) -> None:
chunk = self.loaded_chunks.pop(key, None)
if chunk and chunk.modifications:
self.modifications_db[key] = chunk.modifications
class WorldManager:
"""Master High-Level World Orchestrator."""
def __init__(self, seed: int = 42, max_loaded: int = 25):
self.seed = seed
self.generator = WorldGenerator(seed=seed)
self.chunks = ChunkManager(self.generator, max_loaded_chunks=max_loaded)
self.world_time_sec: float = 0.0
def tick(self, dt: float, agent_pos: Tuple[float, float, float]) -> Dict[str, Any]:
self.world_time_sec += dt
active_chunks = self.chunks.update_center(agent_pos[0], agent_pos[1])
return {
"world_time": round(self.world_time_sec, 2),
"active_chunk_count": len(active_chunks),
"active_chunks": active_chunks
}
|