""" Map Style Relational Service for FreeExile. Provides query interfaces for biomes, tile mappings, props, and idempotent SQLite seeder. """ from __future__ import annotations import os import sqlite3 from contextlib import contextmanager from pathlib import Path from typing import Generator, List, Optional, Union from server.world.map_style_types import ( MapStyleDefinition, MapStylePalette, MapStyleLighting, MapStyleHazard, MapStyleTileMapping, MapStylePropDefinition, ) from server.world.map_style_schema import init_db, SQL_PRAGMA_SETUP import server.world.map_style_catalog as catalog class MapStyleService: """Service providing relational access to Dark Fantasy Cổ Võ Hoang Vực map styles.""" def __init__(self, db_path: Optional[Union[Path, str]] = None): if db_path is None: base_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "data")) os.makedirs(base_dir, exist_ok=True) db_path = os.path.join(base_dir, "map_styles.db") self.db_path = str(db_path) self._memory_conn: Optional[sqlite3.Connection] = None if self.db_path == ":memory:": self._memory_conn = init_db(":memory:") self.seed_database(force=False) else: with self._get_connection() as conn: pass # ensures tables and pragmas are initialized via init_db self.seed_database(force=False) @contextmanager def _get_connection(self) -> Generator[sqlite3.Connection, None, None]: if self._memory_conn is not None: yield self._memory_conn self._memory_conn.commit() else: conn = init_db(self.db_path) try: yield conn conn.commit() except Exception: conn.rollback() raise finally: conn.close() def close(self) -> None: """Close connection if running in-memory.""" if self._memory_conn is not None: self._memory_conn.close() self._memory_conn = None def seed_database(self, force: bool = False) -> int: """Seed all 30 styles, tile mappings, and props idempotently.""" with self._get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT count(*) AS cnt FROM map_styles") row = cursor.fetchone() if row and row["cnt"] >= 30 and not force: return int(row["cnt"]) for s in catalog.list_styles(): cursor.execute( """ INSERT OR REPLACE INTO map_styles ( style_id, biome_code, name_vi, name_en, theme_category, theme_category_name, description, floor_color_hex, wall_color_hex, path_color_hex, liquid_color_hex, ambient_light_hex, fog_density, movement_modifier, hazard_damage, hazard_slow_factor, asset_dir, monster_family, legacy_alias ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( s.style_id, s.biome_code, s.name_vi, s.name_en, s.theme_category, s.theme_category_name, s.description, s.palette.floor, s.palette.wall, s.palette.path, s.palette.liquid, s.lighting.ambient_hex, s.lighting.fog_density, s.movement_modifier, s.hazard.dps, s.hazard.slow_factor, s.asset_dir, s.monster_family, s.legacy_alias, ), ) for t in catalog.get_tiles_for_style(s.style_id): cursor.execute( """ INSERT OR REPLACE INTO map_style_tiles ( style_id, tile_type_id, tile_type_name, texture_filename, normal_filename, texture_width, texture_height, is_seamless, elevation_px, roughness, base_color_fallback ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( t.style_id, t.tile_code, t.tile_type_name, t.texture_path, t.normal_path, t.texture_width, t.texture_height, t.is_seamless, t.elevation_px, t.roughness, t.procedural_fallback_color, ), ) for p in catalog.get_props_for_style(s.style_id): cursor.execute( """ INSERT OR REPLACE INTO map_style_props ( prop_id, style_id, prop_name_vi, prop_name_en, sprite_key, sprite_path, spawn_weight, placement_rule, blocks_movement, footprint_w, footprint_h, elevation_z ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( p.prop_id, p.style_id, p.name_vi, p.name_en, p.sprite_key, p.sprite_path, p.spawn_weight, p.placement_rule, p.blocks_movement, p.footprint_w, p.footprint_h, p.elevation_z, ), ) cursor.execute("SELECT count(*) AS cnt FROM map_styles") return int(cursor.fetchone()["cnt"]) @staticmethod def _row_to_style(row: sqlite3.Row) -> MapStyleDefinition: pal = MapStylePalette( floor=row["floor_color_hex"], wall=row["wall_color_hex"], path=row["path_color_hex"], liquid=row["liquid_color_hex"], ambient=row["ambient_light_hex"], ) light = MapStyleLighting( ambient_hex=row["ambient_light_hex"], intensity=1.0, fog_density=row["fog_density"], ) haz = MapStyleHazard( slow_factor=row["hazard_slow_factor"], dps=row["hazard_damage"], ) return MapStyleDefinition( style_id=row["style_id"], biome_code=row["biome_code"], name_vi=row["name_vi"], name_en=row["name_en"], theme_category=row["theme_category"], theme_category_name=row["theme_category_name"], description=row["description"], palette=pal, lighting=light, hazard=haz, movement_modifier=row["movement_modifier"], asset_dir=row["asset_dir"], monster_family=row["monster_family"], legacy_alias=row["legacy_alias"], ) def get_style(self, style_id: str) -> Optional[MapStyleDefinition]: """Query style by canonical style_id or legacy alias.""" with self._get_connection() as conn: cursor = conn.cursor() cursor.execute( "SELECT * FROM map_styles WHERE style_id = ? OR legacy_alias = ?", (style_id, style_id), ) row = cursor.fetchone() return self._row_to_style(row) if row else None def get_style_by_code(self, biome_code: int) -> Optional[MapStyleDefinition]: """Query style by biome_code (1-30).""" with self._get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT * FROM map_styles WHERE biome_code = ?", (biome_code,)) row = cursor.fetchone() return self._row_to_style(row) if row else None def list_styles(self) -> List[MapStyleDefinition]: """Return all 30 styles ordered by biome_code.""" with self._get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT * FROM map_styles ORDER BY biome_code ASC") return [self._row_to_style(r) for r in cursor.fetchall()] def get_tiles_for_style(self, style_id: str) -> List[MapStyleTileMapping]: """Retrieve the 20 TileType bindings for a style.""" resolved = catalog.LEGACY_BIOME_TO_STYLE_ID.get(style_id, style_id) with self._get_connection() as conn: cursor = conn.cursor() cursor.execute( "SELECT * FROM map_style_tiles WHERE style_id = ? ORDER BY tile_type_id ASC", (resolved,), ) return [ MapStyleTileMapping( style_id=r["style_id"], tile_code=r["tile_type_id"], tile_type_name=r["tile_type_name"], texture_path=r["texture_filename"], normal_path=r["normal_filename"], procedural_fallback_color=r["base_color_fallback"], texture_width=r["texture_width"], texture_height=r["texture_height"], is_seamless=r["is_seamless"], elevation_px=r["elevation_px"], roughness=r["roughness"], ) for r in cursor.fetchall() ] def get_tile_mapping(self, style_id: str, tile_code: int) -> Optional[MapStyleTileMapping]: """Retrieve single TileType mapping for a style and tile_code.""" resolved = catalog.LEGACY_BIOME_TO_STYLE_ID.get(style_id, style_id) with self._get_connection() as conn: cursor = conn.cursor() cursor.execute( "SELECT * FROM map_style_tiles WHERE style_id = ? AND tile_type_id = ?", (resolved, tile_code), ) r = cursor.fetchone() if not r: return None return MapStyleTileMapping( style_id=r["style_id"], tile_code=r["tile_type_id"], tile_type_name=r["tile_type_name"], texture_path=r["texture_filename"], normal_path=r["normal_filename"], procedural_fallback_color=r["base_color_fallback"], texture_width=r["texture_width"], texture_height=r["texture_height"], is_seamless=r["is_seamless"], elevation_px=r["elevation_px"], roughness=r["roughness"], ) def get_props_for_style(self, style_id: str) -> List[MapStylePropDefinition]: """Retrieve decorative props for a style.""" resolved = catalog.LEGACY_BIOME_TO_STYLE_ID.get(style_id, style_id) with self._get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT * FROM map_style_props WHERE style_id = ?", (resolved,)) return [ MapStylePropDefinition( style_id=r["style_id"], prop_id=r["prop_id"], name_vi=r["prop_name_vi"], name_en=r["prop_name_en"], placement_rule=r["placement_rule"], spawn_weight=r["spawn_weight"], footprint_w=r["footprint_w"], footprint_h=r["footprint_h"], sprite_key=r["sprite_key"], sprite_path=r["sprite_path"], blocks_movement=r["blocks_movement"], elevation_z=r["elevation_z"], ) for r in cursor.fetchall() ] def list_props(self, style_id: str) -> List[MapStylePropDefinition]: """Alias for get_props_for_style.""" return self.get_props_for_style(style_id)