From 2f12a0325722637e2fdc92027c50e66f2a5f1fd5 Mon Sep 17 00:00:00 2001 From: Latte Date: Sat, 27 Jun 2026 20:43:02 +0200 Subject: [PATCH] feat(sde): optional local SDE store for offline name/type lookups Add src/familiar/sde.py backing type lookups from a local SQLite SDE with graceful fallback to ESI when absent. lookup.resolve_id prefers the SDE for inventory_types; universe.get_item serves SDE-backed fields with no network call. Register a read-only sde_status() MCP tool. Tested against an in-memory SQLite fixture. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/familiar/lookup.py | 7 ++ src/familiar/sde.py | 164 +++++++++++++++++++++++++++++++++++++++ src/familiar/server.py | 4 + src/familiar/universe.py | 15 ++++ tests/test_sde.py | 125 +++++++++++++++++++++++++++++ 5 files changed, 315 insertions(+) create mode 100644 src/familiar/sde.py create mode 100644 tests/test_sde.py diff --git a/src/familiar/lookup.py b/src/familiar/lookup.py index a527306..33a2bb9 100644 --- a/src/familiar/lookup.py +++ b/src/familiar/lookup.py @@ -6,6 +6,7 @@ instead of a numeric ID. from __future__ import annotations +from . import sde from .esi import client # ESI /universe/ids/ groups results by category. We care about these. @@ -47,5 +48,11 @@ async def resolve_id(value: str | int, *categories: str) -> int: direct = _as_id(value) if direct is not None: return direct + # Prefer the offline SDE for inventory types so a present local store avoids + # the ESI round-trip; fall back to ESI when the SDE is absent or has no hit. + if "inventory_types" in categories and sde.available(): + offline = sde.resolve_type_name(str(value)) + if offline is not None: + return offline resolved, _ = await resolve_name(str(value), *categories) return resolved diff --git a/src/familiar/sde.py b/src/familiar/sde.py new file mode 100644 index 0000000..851655f --- /dev/null +++ b/src/familiar/sde.py @@ -0,0 +1,164 @@ +"""Optional local SDE (Static Data Export) store backing offline name/type lookups. + +EVE's Static Data Export ships the static universe (item types, groups, ...) as a +bulk download. When a caller has placed a prepared SQLite copy on disk, familiar +can resolve type names and serve item details without a network round-trip to +ESI. The store is entirely optional: when the file (or a needed table) is absent +every accessor degrades gracefully — :func:`available` reports ``False`` and the +callers fall back to ESI. + +The path is taken from the ``FAMILIAR_SDE_PATH`` environment variable, else +``config.data_dir() / "sde.sqlite"``. + +Minimal expected schema (only what familiar reads):: + + CREATE TABLE invTypes ( + typeID INTEGER PRIMARY KEY, + typeName TEXT, + groupID INTEGER, + volume REAL + ); + +Anything beyond these columns is ignored, so a full official SDE conversion works +as well as a hand-built subset. +""" + +from __future__ import annotations + +import os +import sqlite3 +from typing import Any + +from fastmcp import FastMCP + +from .config import data_dir + + +def sde_path() -> str: + """Filesystem path to the SQLite SDE, env-overridable via ``FAMILIAR_SDE_PATH``.""" + override = os.environ.get("FAMILIAR_SDE_PATH") + if override: + return override + return str(data_dir() / "sde.sqlite") + + +_conn: sqlite3.Connection | None = None +_connect_attempted = False + + +def connect() -> sqlite3.Connection | None: + """Return a cached read-only sqlite connection, or ``None`` if unavailable. + + The connection is opened lazily on first use and cached. A missing file + yields ``None`` rather than raising, so callers can treat the SDE as best + effort. + """ + global _conn, _connect_attempted + if _conn is not None: + return _conn + if _connect_attempted: + return None + _connect_attempted = True + path = sde_path() + if not os.path.isfile(path): + return None + try: + conn = sqlite3.connect(path) + conn.row_factory = sqlite3.Row + _conn = conn + except sqlite3.Error: + return None + return _conn + + +def set_connection(conn: sqlite3.Connection | None) -> None: + """Replace the cached connection (used by tests to inject an in-memory DB). + + Mirrors :func:`familiar.esi.set_client`. Passing ``None`` clears the cache and + resets the "already attempted" flag so a later :func:`connect` re-evaluates + the on-disk path. + """ + global _conn, _connect_attempted + if conn is not None: + conn.row_factory = sqlite3.Row + _conn = conn + _connect_attempted = False + + +def available() -> bool: + """Whether a usable SDE with an ``invTypes`` table is present.""" + conn = connect() + if conn is None: + return False + try: + conn.execute("SELECT 1 FROM invTypes LIMIT 1") + except sqlite3.Error: + return False + return True + + +def _row_to_type(row: sqlite3.Row | None) -> dict[str, Any] | None: + """Normalize an ``invTypes`` row to the familiar type dict.""" + if row is None: + return None + return { + "type_id": row["typeID"], + "name": row["typeName"], + "group_id": row["groupID"], + "volume": row["volume"], + } + + +def type_by_id(type_id: int) -> dict[str, Any] | None: + """Look up a type by its numeric id, or ``None`` if absent/unavailable.""" + conn = connect() + if conn is None: + return None + try: + cur = conn.execute( + "SELECT typeID, typeName, groupID, volume FROM invTypes WHERE typeID = ?", + (type_id,), + ) + return _row_to_type(cur.fetchone()) + except sqlite3.Error: + return None + + +def type_by_name(name: str) -> dict[str, Any] | None: + """Look up a type by exact (case-insensitive) name, or ``None``.""" + conn = connect() + if conn is None: + return None + try: + cur = conn.execute( + "SELECT typeID, typeName, groupID, volume FROM invTypes " + "WHERE typeName = ? COLLATE NOCASE", + (name,), + ) + return _row_to_type(cur.fetchone()) + except sqlite3.Error: + return None + + +def resolve_type_name(name: str) -> int | None: + """Resolve a type name to its id offline (case-insensitive), or ``None``.""" + found = type_by_name(name) + return found["type_id"] if found is not None else None + + +def sde_status() -> dict[str, Any]: + """Report whether the local SDE is available and where it is expected. + + A read-only diagnostic: tells the caller if offline lookups are active and + the path familiar is checking, without revealing any account data. + """ + return {"available": available(), "path": sde_path()} + + +TOOLS = [sde_status] + + +def register(mcp: FastMCP) -> None: + """Register the SDE tools onto the given server.""" + for fn in TOOLS: + mcp.tool(fn) diff --git a/src/familiar/server.py b/src/familiar/server.py index 5a6d3a1..18a1b72 100644 --- a/src/familiar/server.py +++ b/src/familiar/server.py @@ -26,6 +26,7 @@ from . import ( misc, planets, profile, + sde, structures, trade, ui, @@ -64,6 +65,9 @@ def ping() -> dict[str, str]: universe.register(mcp) market.register(mcp) +# Optional local SDE store (offline name/type lookups; reports availability). +sde.register(mcp) + # Phase 2: authenticated, read-only account lookups (requires `familiar login`). account.register(mcp) diff --git a/src/familiar/universe.py b/src/familiar/universe.py index f7588b9..b7b31f2 100644 --- a/src/familiar/universe.py +++ b/src/familiar/universe.py @@ -10,6 +10,7 @@ from typing import Any from fastmcp import FastMCP +from . import sde from .esi import client from .lookup import resolve_id @@ -28,6 +29,20 @@ async def resolve_names(names: list[str]) -> dict[str, Any]: async def get_item(item: str | int) -> dict[str, Any]: """Look up an item/type by name or type_id (volume, group, description).""" type_id = await resolve_id(item, "inventory_types") + # Serve from the offline SDE when present — no network call. The SDE lacks + # packaged_volume/description, so those come back as None; ESI is the source + # of record when the SDE is absent or doesn't know the type. + if sde.available(): + offline = sde.type_by_id(type_id) + if offline is not None: + return { + "type_id": offline["type_id"], + "name": offline["name"], + "group_id": offline["group_id"], + "volume": offline["volume"], + "packaged_volume": None, + "description": None, + } data = await client().get(f"/universe/types/{type_id}/") return { "type_id": data["type_id"], diff --git a/tests/test_sde.py b/tests/test_sde.py new file mode 100644 index 0000000..b26892f --- /dev/null +++ b/tests/test_sde.py @@ -0,0 +1,125 @@ +"""Local SDE store: offline lookups and SDE-preferred / ESI-fallback wiring.""" + +import asyncio +import sqlite3 + +import httpx +import pytest + +from familiar import sde, universe + + +def _make_sde(rows=((34, "Tritanium", 18, 0.01), (35, "Pyerite", 18, 0.01))): + """In-memory SQLite seeded with the minimal invTypes schema and rows.""" + conn = sqlite3.connect(":memory:") + conn.execute( + "CREATE TABLE invTypes (" + "typeID INTEGER PRIMARY KEY, typeName TEXT, groupID INTEGER, volume REAL)" + ) + conn.executemany("INSERT INTO invTypes VALUES (?, ?, ?, ?)", rows) + conn.commit() + return conn + + +@pytest.fixture +def use_sde(): + """Inject an in-memory SDE and reset the cached connection afterwards.""" + conn = _make_sde() + sde.set_connection(conn) + yield conn + sde.set_connection(None) + conn.close() + + +def test_available_true_with_table(use_sde): + assert sde.available() is True + + +def test_available_false_without_connection(): + sde.set_connection(None) + # No file on the default/overridden path in the test environment. + assert sde.available() is False + + +def test_available_false_when_table_missing(): + conn = sqlite3.connect(":memory:") # connected, but no invTypes table + sde.set_connection(conn) + try: + assert sde.available() is False + # The graceful-degradation accessors must not crash either. + assert sde.type_by_id(34) is None + assert sde.type_by_name("Tritanium") is None + assert sde.resolve_type_name("Tritanium") is None + finally: + sde.set_connection(None) + conn.close() + + +def test_type_by_id(use_sde): + item = sde.type_by_id(34) + assert item == {"type_id": 34, "name": "Tritanium", "group_id": 18, "volume": 0.01} + + +def test_type_by_id_missing(use_sde): + assert sde.type_by_id(999999) is None + + +def test_type_by_name(use_sde): + item = sde.type_by_name("Pyerite") + assert item["type_id"] == 35 + + +def test_resolve_type_name_case_insensitive(use_sde): + assert sde.resolve_type_name("tritanium") == 34 + assert sde.resolve_type_name("TRITANIUM") == 34 + assert sde.resolve_type_name("Tritanium") == 34 + + +def test_resolve_type_name_unknown(use_sde): + assert sde.resolve_type_name("Nope") is None + + +def test_sde_status(use_sde): + status = sde.sde_status() + assert status["available"] is True + assert isinstance(status["path"], str) + + +def test_get_item_uses_sde_without_esi(use_sde, mock_esi): + def boom(request): # any ESI call here is a bug + raise AssertionError(f"ESI was called: {request.url}") + + mock_esi(boom) + result = asyncio.run(universe.get_item("Tritanium")) + assert result["type_id"] == 34 + assert result["name"] == "Tritanium" + assert result["volume"] == 0.01 + # SDE has no description / packaged_volume. + assert result["description"] is None + assert result["packaged_volume"] is None + + +def test_get_item_falls_back_to_esi_when_sde_absent(mock_esi): + sde.set_connection(None) # no SDE + fixtures = { + "/universe/ids/": {"inventory_types": [{"id": 34, "name": "Tritanium"}]}, + "/universe/types/": { + "type_id": 34, + "name": "Tritanium", + "group_id": 18, + "volume": 0.01, + "packaged_volume": 0.01, + "description": "The most common ore type.", + }, + } + + def route(request): + for needle, body in fixtures.items(): + if needle in request.url.path: + return httpx.Response(200, json=body) + return httpx.Response(404, json={"error": "no fixture"}) + + mock_esi(route) + result = asyncio.run(universe.get_item("Tritanium")) + assert result["type_id"] == 34 + assert result["description"] == "The most common ore type."