my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	"""
#2	Mnemosyne Cost Logger
#3	Tracks memory injection costs over time for benchmarking.
#4	"""
#5
#6	import json
#7	import sqlite3
#8	from datetime import datetime
#9	from pathlib import Path
#10	from typing import Dict, List
#11
#12	DEFAULT_LOG_DIR = Path.home() / ".mnemosyne" / "data"
#13	DEFAULT_LOG_DB = DEFAULT_LOG_DIR / "cost_log.db"
#14
#15
#16	def _get_conn(db_path: Path = None) -> sqlite3.Connection:
#17	path = db_path or DEFAULT_LOG_DB
#18	path.parent.mkdir(parents=True, exist_ok=True)
#19	conn = sqlite3.connect(str(path), check_same_thread=False)
#20	conn.row_factory = sqlite3.Row
#21	return conn
#22
#23
#24	def init_cost_log(db_path: Path = None):
#25	conn = _get_conn(db_path)
#26	cursor = conn.cursor()
#27	cursor.execute("""
#28	CREATE TABLE IF NOT EXISTS cost_entries (
#29	id INTEGER PRIMARY KEY AUTOINCREMENT,
#30	session_id TEXT,
#31	memory_count INTEGER,
#32	token_count INTEGER,
#33	estimated_cost_usd REAL,
#34	model TEXT DEFAULT 'default',
#35	timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
#36	)
#37	""")
#38	conn.commit()
#39
#40
#41	def log_cost(session_id: str, memory_count: int, token_count: int,
#42	estimated_cost_usd: float, model: str = "default",
#43	db_path: Path = None):
#44	init_cost_log(db_path)
#45	conn = _get_conn(db_path)
#46	cursor = conn.cursor()
#47	cursor.execute("""
#48	INSERT INTO cost_entries (session_id, memory_count, token_count, estimated_cost_usd, model, timestamp)
#49	VALUES (?, ?, ?, ?, ?, ?)
#50	""", (session_id, memory_count, token_count, estimated_cost_usd, model, datetime.now().isoformat()))
#51	conn.commit()
#52
#53
#54	def get_cost_stats(session_id: str = None, db_path: Path = None) -> Dict:
#55	init_cost_log(db_path)
#56	conn = _get_conn(db_path)
#57	cursor = conn.cursor()
#58
#59	if session_id:
#60	cursor.execute("""
#61	SELECT COUNT(*) as calls, SUM(memory_count) as total_memories,
#62	SUM(token_count) as total_tokens, SUM(estimated_cost_usd) as total_cost
#63	FROM cost_entries WHERE session_id = ?
#64	""", (session_id,))
#65	else:
#66	cursor.execute("""
#67	SELECT COUNT(*) as calls, SUM(memory_count) as total_memories,
#68	SUM(token_count) as total_tokens, SUM(estimated_cost_usd) as total_cost
#69	FROM cost_entries
#70	""")
#71
#72	row = cursor.fetchone()
#73	return {
#74	"total_calls": row["calls"] or 0,
#75	"total_memories_injected": row["total_memories"] or 0,
#76	"total_tokens": row["total_tokens"] or 0,
#77	"total_estimated_cost_usd": round(row["total_cost"] or 0, 6),
#78	}
#79

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public