Database Module

The database module handles persistence, logging, and analytics.

Database Manager

class simulation_framework.src.database.database.Database(db_path: str = 'simulation.db')

Bases: object

Main database interface for the simulation framework

cleanup_old_data() → None

Clean up old data according to retention policies

close() → None

Close database connection

create_simulation_run(simulation: SimulationRun) → int

Create a new simulation run record

execute_custom_query(query: str, params: Tuple = ()) → List[Dict[str, Any]]

Execute custom SQL query (use with caution)

get_action_logs(simulation_id: int, agent_id: int | None = None, action_type: str | None = None, start_tick: int | None = None, end_tick: int | None = None, limit: int = 1000) → List[ActionLog]

Get action logs with filtering

get_action_summary(simulation_id: int) → List[Dict[str, Any]]

Get action summary using database view

get_agent_snapshots(simulation_id: int, agent_id: int | None = None, start_tick: int | None = None, end_tick: int | None = None, limit: int = 1000) → List[AgentSnapshot]

Get agent snapshots with filtering

get_agent_summary(simulation_id: int) → List[Dict[str, Any]]

Get agent summary using database view

get_analytics(simulation_id: int, metric_name: str | None = None, category: str | None = None, start_tick: int | None = None, end_tick: int | None = None, limit: int = 1000) → List[Analytics]

Get analytics data with filtering

get_combat_summary(simulation_id: int) → Dict[str, Any] | None

Get combat summary using database view

get_connection()

Get database connection with proper cleanup

get_database_stats() → Dict[str, Any]

Get database statistics

get_simulation_run(simulation_id: int) → SimulationRun | None

Get simulation run by ID

get_trade_summary(simulation_id: int) → Dict[str, Any] | None

Get trade summary using database view

get_world_snapshots(simulation_id: int, start_tick: int | None = None, end_tick: int | None = None, limit: int = 1000) → List[WorldSnapshot]

Get world snapshots with filtering

list_simulation_runs(limit: int = 100, offset: int = 0) → List[SimulationRun]

List all simulation runs

log_action(action: ActionLog) → int

Log an action

log_combat(combat: CombatLog) → int

Log a combat action

log_trade(trade: TradeLog) → int

Log a trade transaction

save_agent_snapshot(snapshot: AgentSnapshot) → int

Save agent snapshot

save_agent_snapshots_batch(snapshots: List[AgentSnapshot]) → None

Save multiple agent snapshots efficiently

save_analytics(analytics: Analytics) → int

Save analytics metric

save_world_snapshot(snapshot: WorldSnapshot) → int

Save world snapshot

update_simulation_run(simulation: SimulationRun) → bool

Update simulation run record

Data Models

class simulation_framework.src.database.models.ActionLog(id: int | None = None, simulation_id: int = 0, tick: int = 0, agent_id: int = 0, action_type: str = '', action_data: Dict | None = None, success: bool = False, result_message: str = '', duration: int = 1)

Bases: object

Model for logging agent actions

action_data: Dict = None

action_type: str = ''

agent_id: int = 0

duration: int = 1

id: int | None = None

result_message: str = ''

simulation_id: int = 0

success: bool = False

tick: int = 0

class simulation_framework.src.database.models.AgentSnapshot(id: int | None = None, simulation_id: int = 0, agent_id: int = 0, tick: int = 0, name: str = '', position_x: int = 0, position_y: int = 0, health: int = 0, max_health: int = 0, stamina: int = 0, max_stamina: int = 0, personality: Dict | None = None, character_class: str = '', skills: Dict | None = None, current_goals: List | None = None, relationships: Dict | None = None, inventory_items: int = 0, gold: int = 0)

Bases: object

Model for agent state snapshots

agent_id: int = 0

character_class: str = ''

current_goals: List = None

gold: int = 0

health: int = 0

id: int | None = None

inventory_items: int = 0

max_health: int = 0

max_stamina: int = 0

name: str = ''

personality: Dict = None

position_x: int = 0

position_y: int = 0

relationships: Dict = None

simulation_id: int = 0

skills: Dict = None

stamina: int = 0

tick: int = 0

class simulation_framework.src.database.models.Analytics(id: int | None = None, simulation_id: int = 0, metric_name: str = '', metric_value: float = 0.0, tick: int = 0, category: str = '', metadata: Dict | None = None)

Bases: object

Model for simulation analytics and metrics

category: str = ''

id: int | None = None

metadata: Dict = None

metric_name: str = ''

metric_value: float = 0.0

simulation_id: int = 0

tick: int = 0

class simulation_framework.src.database.models.CombatLog(id: int | None = None, simulation_id: int = 0, tick: int = 0, attacker_id: int = 0, target_id: int = 0, damage_dealt: int = 0, damage_type: str = '', was_critical: bool = False, weapon_used: str = '', target_died: bool = False)

Bases: object

Model for logging combat encounters

attacker_id: int = 0

damage_dealt: int = 0

damage_type: str = ''

id: int | None = None

simulation_id: int = 0

target_died: bool = False

target_id: int = 0

tick: int = 0

was_critical: bool = False

weapon_used: str = ''

class simulation_framework.src.database.models.DatabaseHelper

Bases: object

Helper functions for database operations

static dataclass_to_dict(obj: Any, for_insert: bool = False) → Dict

Convert dataclass to dictionary for database operations

static dict_to_json(data: Dict) → str

Convert dictionary to JSON string for database storage

static json_to_dict(data: str) → Dict

Convert JSON string from database to dictionary

static json_to_list(data: str) → List

Convert JSON string from database to list

static list_to_json(data: List) → str

Convert list to JSON string for database storage

static row_to_dataclass(row: Row, model_class) → Any

Convert SQLite row to dataclass instance

class simulation_framework.src.database.models.SimulationRun(id: int | None = None, name: str = '', description: str = '', world_seed: int = 0, world_width: int = 0, world_height: int = 0, start_time: datetime | None = None, end_time: datetime | None = None, current_tick: int = 0, total_agents: int = 0, config: Dict | None = None)

Bases: object

Model for simulation run metadata

config: Dict = None

current_tick: int = 0

description: str = ''

end_time: datetime | None = None

id: int | None = None

name: str = ''

start_time: datetime | None = None

total_agents: int = 0

world_height: int = 0

world_seed: int = 0

world_width: int = 0

class simulation_framework.src.database.models.TradeLog(id: int | None = None, simulation_id: int = 0, tick: int = 0, initiator_id: int = 0, target_id: int = 0, offered_items: Dict | None = None, requested_items: Dict | None = None, offered_gold: int = 0, requested_gold: int = 0, completed: bool = False)

Bases: object

Model for logging trade transactions

completed: bool = False

id: int | None = None

initiator_id: int = 0

offered_gold: int = 0

offered_items: Dict = None

requested_gold: int = 0

requested_items: Dict = None

simulation_id: int = 0

target_id: int = 0

tick: int = 0

class simulation_framework.src.database.models.WorldSnapshot(id: int | None = None, simulation_id: int = 0, tick: int = 0, total_entities: int = 0, active_agents: int = 0, active_npcs: int = 0, resource_nodes: int = 0, world_events: List | None = None, market_prices: Dict | None = None)

Bases: object

Model for world state snapshots

active_agents: int = 0

active_npcs: int = 0

id: int | None = None

market_prices: Dict = None

resource_nodes: int = 0

simulation_id: int = 0

tick: int = 0

total_entities: int = 0

world_events: List = None

Schema

Database schema definitions and migration management

class simulation_framework.src.database.schema.DatabaseSchema

Bases: object

Database schema management and SQL definitions

CREATE_TABLES = ["\n        CREATE TABLE IF NOT EXISTS simulation_runs (\n            id INTEGER PRIMARY KEY AUTOINCREMENT,\n            name TEXT NOT NULL,\n            description TEXT,\n            world_seed INTEGER NOT NULL,\n            world_width INTEGER NOT NULL,\n            world_height INTEGER NOT NULL,\n            start_time TEXT,\n            end_time TEXT,\n            current_tick INTEGER DEFAULT 0,\n            total_agents INTEGER DEFAULT 0,\n            config TEXT DEFAULT '{}',\n            created_at DATETIME DEFAULT CURRENT_TIMESTAMP,\n            updated_at DATETIME DEFAULT CURRENT_TIMESTAMP\n        )\n        ", "\n        CREATE TABLE IF NOT EXISTS agent_snapshots (\n            id INTEGER PRIMARY KEY AUTOINCREMENT,\n            simulation_id INTEGER NOT NULL,\n            agent_id INTEGER NOT NULL,\n            tick INTEGER NOT NULL,\n            name TEXT NOT NULL,\n            position_x INTEGER NOT NULL,\n            position_y INTEGER NOT NULL,\n            health INTEGER NOT NULL,\n            max_health INTEGER NOT NULL,\n            stamina INTEGER NOT NULL,\n            max_stamina INTEGER NOT NULL,\n            personality TEXT DEFAULT '{}',\n            character_class TEXT DEFAULT '',\n            skills TEXT DEFAULT '{}',\n            current_goals TEXT DEFAULT '[]',\n            relationships TEXT DEFAULT '{}',\n            inventory_items INTEGER DEFAULT 0,\n            gold INTEGER DEFAULT 0,\n            created_at DATETIME DEFAULT CURRENT_TIMESTAMP,\n            FOREIGN KEY (simulation_id) REFERENCES simulation_runs(id)\n        )\n        ", "\n        CREATE TABLE IF NOT EXISTS world_snapshots (\n            id INTEGER PRIMARY KEY AUTOINCREMENT,\n            simulation_id INTEGER NOT NULL,\n            tick INTEGER NOT NULL,\n            total_entities INTEGER DEFAULT 0,\n            active_agents INTEGER DEFAULT 0,\n            active_npcs INTEGER DEFAULT 0,\n            resource_nodes INTEGER DEFAULT 0,\n            world_events TEXT DEFAULT '[]',\n            market_prices TEXT DEFAULT '{}',\n            created_at DATETIME DEFAULT CURRENT_TIMESTAMP,\n            FOREIGN KEY (simulation_id) REFERENCES simulation_runs(id)\n        )\n        ", "\n        CREATE TABLE IF NOT EXISTS action_logs (\n            id INTEGER PRIMARY KEY AUTOINCREMENT,\n            simulation_id INTEGER NOT NULL,\n            tick INTEGER NOT NULL,\n            agent_id INTEGER NOT NULL,\n            action_type TEXT NOT NULL,\n            action_data TEXT DEFAULT '{}',\n            success BOOLEAN DEFAULT FALSE,\n            result_message TEXT DEFAULT '',\n            duration INTEGER DEFAULT 1,\n            created_at DATETIME DEFAULT CURRENT_TIMESTAMP,\n            FOREIGN KEY (simulation_id) REFERENCES simulation_runs(id)\n        )\n        ", "\n        CREATE TABLE IF NOT EXISTS trade_logs (\n            id INTEGER PRIMARY KEY AUTOINCREMENT,\n            simulation_id INTEGER NOT NULL,\n            tick INTEGER NOT NULL,\n            initiator_id INTEGER NOT NULL,\n            target_id INTEGER NOT NULL,\n            offered_items TEXT DEFAULT '{}',\n            requested_items TEXT DEFAULT '{}',\n            offered_gold INTEGER DEFAULT 0,\n            requested_gold INTEGER DEFAULT 0,\n            completed BOOLEAN DEFAULT FALSE,\n            created_at DATETIME DEFAULT CURRENT_TIMESTAMP,\n            FOREIGN KEY (simulation_id) REFERENCES simulation_runs(id)\n        )\n        ", "\n        CREATE TABLE IF NOT EXISTS combat_logs (\n            id INTEGER PRIMARY KEY AUTOINCREMENT,\n            simulation_id INTEGER NOT NULL,\n            tick INTEGER NOT NULL,\n            attacker_id INTEGER NOT NULL,\n            target_id INTEGER NOT NULL,\n            damage_dealt INTEGER DEFAULT 0,\n            damage_type TEXT DEFAULT '',\n            was_critical BOOLEAN DEFAULT FALSE,\n            weapon_used TEXT DEFAULT '',\n            target_died BOOLEAN DEFAULT FALSE,\n            created_at DATETIME DEFAULT CURRENT_TIMESTAMP,\n            FOREIGN KEY (simulation_id) REFERENCES simulation_runs(id)\n        )\n        ", "\n        CREATE TABLE IF NOT EXISTS analytics (\n            id INTEGER PRIMARY KEY AUTOINCREMENT,\n            simulation_id INTEGER NOT NULL,\n            metric_name TEXT NOT NULL,\n            metric_value REAL NOT NULL,\n            tick INTEGER NOT NULL,\n            category TEXT DEFAULT '',\n            metadata TEXT DEFAULT '{}',\n            created_at DATETIME DEFAULT CURRENT_TIMESTAMP,\n            FOREIGN KEY (simulation_id) REFERENCES simulation_runs(id)\n        )\n        ", '\n        CREATE TABLE IF NOT EXISTS schema_version (\n            version INTEGER PRIMARY KEY,\n            applied_at DATETIME DEFAULT CURRENT_TIMESTAMP\n        )\n        ', 'CREATE INDEX IF NOT EXISTS idx_agent_snapshots_sim_tick ON agent_snapshots (simulation_id, tick)', 'CREATE INDEX IF NOT EXISTS idx_agent_snapshots_agent_id ON agent_snapshots (agent_id)', 'CREATE INDEX IF NOT EXISTS idx_world_snapshots_sim_tick ON world_snapshots (simulation_id, tick)', 'CREATE INDEX IF NOT EXISTS idx_action_logs_sim_tick ON action_logs (simulation_id, tick)', 'CREATE INDEX IF NOT EXISTS idx_action_logs_agent_id ON action_logs (agent_id)', 'CREATE INDEX IF NOT EXISTS idx_action_logs_type ON action_logs (action_type)', 'CREATE INDEX IF NOT EXISTS idx_trade_logs_sim_tick ON trade_logs (simulation_id, tick)', 'CREATE INDEX IF NOT EXISTS idx_trade_logs_initiator ON trade_logs (initiator_id)', 'CREATE INDEX IF NOT EXISTS idx_trade_logs_target ON trade_logs (target_id)', 'CREATE INDEX IF NOT EXISTS idx_combat_logs_sim_tick ON combat_logs (simulation_id, tick)', 'CREATE INDEX IF NOT EXISTS idx_combat_logs_attacker ON combat_logs (attacker_id)', 'CREATE INDEX IF NOT EXISTS idx_combat_logs_target ON combat_logs (target_id)', 'CREATE INDEX IF NOT EXISTS idx_analytics_sim_metric ON analytics (simulation_id, metric_name)', 'CREATE INDEX IF NOT EXISTS idx_analytics_category ON analytics (category)', 'CREATE INDEX IF NOT EXISTS idx_analytics_tick ON analytics (tick)']

CREATE_TRIGGERS = ['\n        CREATE TRIGGER IF NOT EXISTS update_simulation_runs_timestamp\n        AFTER UPDATE ON simulation_runs\n        BEGIN\n            UPDATE simulation_runs SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id;\n        END\n        ']

CREATE_VIEWS = ['\n        CREATE VIEW IF NOT EXISTS agent_summary AS\n        SELECT\n            s.name as simulation_name,\n            a.simulation_id,\n            a.agent_id,\n            a.name as agent_name,\n            a.character_class,\n            COUNT(*) as snapshot_count,\n            MIN(a.tick) as first_tick,\n            MAX(a.tick) as last_tick,\n            AVG(a.health * 1.0 / a.max_health) as avg_health_ratio,\n            AVG(a.stamina * 1.0 / a.max_stamina) as avg_stamina_ratio\n        FROM agent_snapshots a\n        JOIN simulation_runs s ON a.simulation_id = s.id\n        GROUP BY a.simulation_id, a.agent_id, a.name, a.character_class\n        ', '\n        CREATE VIEW IF NOT EXISTS action_summary AS\n        SELECT\n            simulation_id,\n            action_type,\n            COUNT(*) as total_actions,\n            SUM(CASE WHEN success THEN 1 ELSE 0 END) as successful_actions,\n            AVG(CASE WHEN success THEN 1.0 ELSE 0.0 END) as success_rate,\n            AVG(duration) as avg_duration\n        FROM action_logs\n        GROUP BY simulation_id, action_type\n        ', '\n        CREATE VIEW IF NOT EXISTS trade_summary AS\n        SELECT\n            simulation_id,\n            COUNT(*) as total_trades,\n            SUM(CASE WHEN completed THEN 1 ELSE 0 END) as completed_trades,\n            AVG(CASE WHEN completed THEN 1.0 ELSE 0.0 END) as completion_rate,\n            AVG(offered_gold + requested_gold) as avg_gold_volume\n        FROM trade_logs\n        GROUP BY simulation_id\n        ', '\n        CREATE VIEW IF NOT EXISTS combat_summary AS\n        SELECT\n            simulation_id,\n            COUNT(*) as total_combats,\n            SUM(damage_dealt) as total_damage,\n            AVG(damage_dealt) as avg_damage,\n            SUM(CASE WHEN was_critical THEN 1 ELSE 0 END) as critical_hits,\n            SUM(CASE WHEN target_died THEN 1 ELSE 0 END) as deaths,\n            AVG(CASE WHEN was_critical THEN 1.0 ELSE 0.0 END) as critical_rate\n        FROM combat_logs\n        GROUP BY simulation_id\n        ']

SCHEMA_VERSION = 1

classmethod get_all_creation_sql() → List[str]

Get all SQL statements needed to create the complete schema

classmethod get_cleanup_sql() → List[str]

Get SQL statements to clean up old data

classmethod get_migration_sql(from_version: int, to_version: int) → List[str]

Get SQL statements for migrating between schema versions

classmethod get_sample_data_sql() → List[str]

Get SQL statements to insert sample data for testing

Analytics Engine

class simulation_framework.src.database.analytics_engine.AnalyticsEngine(database: Database)

Bases: object

Analytics engine for calculating simulation metrics and insights

calculate_all_metrics(simulation_id: int, current_tick: int) → None

Calculate and save all available metrics for a simulation

generate_simulation_report(simulation_id: int) → Dict[str, Any]

Generate comprehensive simulation report

get_correlation_analysis(simulation_id: int, metric1: str, metric2: str) → Dict[str, Any]

Analyze correlation between two metrics

get_trend_analysis(simulation_id: int, metric_name: str, window_size: int = 100) → Dict[str, Any]

Analyze trends for a specific metric