fix: dashboard instant boot, pipeline map layer, theater data bugs
- Add /api/static endpoint for instant geospatial data on boot (158 items) - Add per-coroutine 45s timeout to prevent SSE first-frame blocking - Dismiss loading overlay after static fetch instead of waiting for SSE - Render 24 oil/gas/hydrogen pipelines as colored polylines on map - Fix theaters dict-vs-list bug in posture.py and fleet.py - Add exposure detail modal and color/label entries Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Marc Shade
@@ -0,0 +1,83 @@
|
||||
# CLAUDE.md
|
||||
|
||||
This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
|
||||
|
||||
## What This Is
|
||||
|
||||
World Intelligence MCP Server — 68 tools across 27 domains providing real-time global intelligence from free public APIs. Serves three interfaces: MCP stdio (for Claude Code/Cursor), a live Starlette dashboard with SSE, and a Click CLI with Rich output.
|
||||
|
||||
## Commands
|
||||
|
||||
```bash
|
||||
# Install
|
||||
pip install -e ".[dev,dashboard]"
|
||||
|
||||
# Run MCP server (stdio mode)
|
||||
world-intel-mcp
|
||||
|
||||
# Run tests (pytest-asyncio, auto mode)
|
||||
pytest
|
||||
pytest --cov=world_intel_mcp
|
||||
pytest src/world_intel_mcp/tests/test_sources.py::test_fetch_market_quotes -v # single test
|
||||
|
||||
# CLI
|
||||
intel markets # stock indices
|
||||
intel earthquakes --min-mag 5.0 # USGS quakes
|
||||
intel report daily # generate HTML report
|
||||
intel status # cache + circuit breaker health
|
||||
|
||||
# Dashboard (requires [dashboard] extra)
|
||||
intel-dashboard --port 8501
|
||||
```
|
||||
|
||||
## Architecture
|
||||
|
||||
Three consumers share the same source modules and infrastructure stack:
|
||||
|
||||
```
|
||||
server.py (MCP stdio) ─┐
|
||||
cli.py (Click CLI) ├─> sources/*.py ─> Fetcher ─> CircuitBreaker ─> Cache (SQLite)
|
||||
dashboard/app.py (SSE) ─┘ │
|
||||
~/.cache/world-intel-mcp/cache.db
|
||||
```
|
||||
|
||||
**Infrastructure layer** (`fetcher.py`, `cache.py`, `circuit_breaker.py`):
|
||||
- `Fetcher`: Centralized async HTTP client (httpx). All external calls go through `get_json()`, `get_text()`, or `get_xml()`. Handles retries (2 max), per-source rate limiting, and stale-data fallback (never returns blank if old data exists).
|
||||
- `CircuitBreaker`: Per-source tracking. 3 consecutive failures trips the breaker for 5 minutes. Each RSS feed gets its own breaker (`rss:bbc_world`).
|
||||
- `Cache`: SQLite WAL-mode TTL cache. `get()` returns live data, `get_stale()` returns expired data for fallback.
|
||||
|
||||
**Source modules** (`sources/*.py`): Each module exports `async def fetch_*(fetcher: Fetcher, **kwargs) -> dict`. Pure data fetching — no MCP awareness. 25 modules covering markets, seismology, military, cyber, health, etc.
|
||||
|
||||
**Analysis modules** (`analysis/*.py`): Cross-domain intelligence that consumes outputs from multiple sources. Includes signal aggregation, instability indexing, NLP (entity extraction, classification, clustering, spike detection via Welford's algorithm), and strategic synthesis.
|
||||
|
||||
**Static config** (`config/*.py`): Curated datasets — 22 intel hotspots, 70+ military bases, 40 ports, 24 pipelines, 24 nuclear facilities, 105 major cities, 28 world leaders, 36 APT groups.
|
||||
|
||||
## Adding a New Tool
|
||||
|
||||
1. Create `sources/your_source.py` with `async def fetch_your_data(fetcher: Fetcher, **kwargs) -> dict`
|
||||
2. Use `fetcher.get_json(url, source="your-source", cache_key=..., cache_ttl=300)` — this gives you caching, retries, circuit breaking, and rate limiting automatically
|
||||
3. In `server.py`: import the module, add a `Tool(...)` to the `TOOLS` list, add a `case` to `_dispatch()`
|
||||
4. Optionally add to `dashboard/app.py` (SSE endpoint) and `cli.py` (Click command)
|
||||
5. Add tests using `respx` to mock HTTP (see `tests/test_sources.py` for pattern)
|
||||
|
||||
## Key Patterns
|
||||
|
||||
- **Source name string**: The `source` parameter in `fetcher.get_json()` identifies the API for circuit breaking and rate limiting. Must match entries in `_SOURCE_RATE_LIMITS` if rate-limited (e.g., `"yahoo-finance"`, `"coingecko"`, `"adsblol"`).
|
||||
- **Tool dispatch**: `server.py` uses Python `match/case` to route tool names to source functions. Tool names follow `intel_*` convention.
|
||||
- **All source functions take `fetcher` as first arg** — never construct your own httpx client.
|
||||
- **Dashboard SSE**: `dashboard/app.py` fetches all domains in parallel via `asyncio.gather()`, streams updates every 30 seconds.
|
||||
- **Tests strip proxy env vars** automatically via `conftest.py` fixture (prevents SOCKS proxy interference).
|
||||
|
||||
## Environment Variables
|
||||
|
||||
Only these unlock additional data sources (everything else works unauthenticated):
|
||||
- `ACLED_ACCESS_TOKEN` — conflict events
|
||||
- `NASA_FIRMS_API_KEY` — satellite wildfire data
|
||||
- `EIA_API_KEY` — energy prices
|
||||
- `CLOUDFLARE_API_TOKEN` — internet outage data
|
||||
- `FRED_API_KEY` — macro economic data
|
||||
- `OPENSKY_CLIENT_ID` / `OPENSKY_CLIENT_SECRET` — military flight fallback
|
||||
|
||||
## Testing
|
||||
|
||||
Tests use `respx` to mock httpx responses. Fixtures in `conftest.py` provide `cache` (tmp_path SQLite) and `fetcher` (with clean breaker). Tests are async (`pytest-asyncio` in auto mode). Proxy env vars are stripped automatically.
|
||||
@@ -59,11 +59,15 @@ def _score_military(surge_data: dict, posture_data: dict) -> tuple[float, list[s
|
||||
for s in surges[:3]:
|
||||
signals.append(f"Surge: {s.get('region', 'unknown')} ({s.get('aircraft_count', '?')} aircraft)")
|
||||
|
||||
theaters = posture_data.get("theaters", [])
|
||||
active_theaters = [t for t in theaters if t.get("aircraft_count", 0) > 10]
|
||||
theaters = posture_data.get("theaters", {})
|
||||
# theaters is a dict keyed by theater name, values are dicts with "count"
|
||||
if isinstance(theaters, dict):
|
||||
active_theaters = [(name, t) for name, t in theaters.items() if isinstance(t, dict) and t.get("count", 0) > 10]
|
||||
else:
|
||||
active_theaters = []
|
||||
score += min(50.0, len(active_theaters) * 12.0)
|
||||
for t in active_theaters[:3]:
|
||||
signals.append(f"{t.get('name', '?')}: {t.get('aircraft_count', 0)} aircraft")
|
||||
for name, t in active_theaters[:3]:
|
||||
signals.append(f"{name}: {t.get('count', 0)} aircraft")
|
||||
|
||||
return min(100.0, score), signals
|
||||
|
||||
|
||||
@@ -116,8 +116,18 @@ async def _fetch_overview() -> dict:
|
||||
"population_exposure": fetch_population_exposure(fetcher),
|
||||
}
|
||||
|
||||
# Per-coro timeout so no single slow source blocks the entire dashboard.
|
||||
# Without this, 80+ RSS feeds timing out sequentially can delay the
|
||||
# first SSE frame for minutes, leaving the dashboard stuck at all-zeros.
|
||||
async def _with_timeout(name: str, coro, timeout: float = 45.0):
|
||||
try:
|
||||
return await asyncio.wait_for(coro, timeout=timeout)
|
||||
except asyncio.TimeoutError:
|
||||
logger.warning("Dashboard fetch %s timed out after %.0fs", name, timeout)
|
||||
return {"error": f"timeout after {timeout}s", "_timeout": True}
|
||||
|
||||
gathered = await asyncio.gather(
|
||||
*[asyncio.create_task(c) for c in coros.values()],
|
||||
*[_with_timeout(name, c) for name, c in zip(coros.keys(), coros.values())],
|
||||
return_exceptions=True,
|
||||
)
|
||||
|
||||
@@ -215,6 +225,20 @@ async def api_stream(request):
|
||||
)
|
||||
|
||||
|
||||
async def api_static(request):
|
||||
"""Return static geospatial datasets instantly (no API calls).
|
||||
|
||||
The dashboard fetches this on boot so the infrastructure layer
|
||||
populates immediately without waiting for the full SSE gather.
|
||||
"""
|
||||
return JSONResponse({
|
||||
"military_bases": {"bases": MILITARY_BASES, "count": len(MILITARY_BASES)},
|
||||
"strategic_ports": {"ports": STRATEGIC_PORTS, "count": len(STRATEGIC_PORTS)},
|
||||
"pipelines": {"pipelines": PIPELINES, "count": len(PIPELINES)},
|
||||
"nuclear_facilities": {"facilities": NUCLEAR_FACILITIES, "count": len(NUCLEAR_FACILITIES)},
|
||||
}, headers={"Access-Control-Allow-Origin": "*"})
|
||||
|
||||
|
||||
async def api_health(request):
|
||||
"""Health check."""
|
||||
return JSONResponse({"status": "ok"})
|
||||
@@ -278,6 +302,7 @@ app = Starlette(
|
||||
Route("/", index),
|
||||
Route("/api/overview", api_overview),
|
||||
Route("/api/stream", api_stream),
|
||||
Route("/api/static", api_static),
|
||||
Route("/api/health", api_health),
|
||||
Route("/api/report/pdf", api_report_pdf),
|
||||
],
|
||||
|
||||
@@ -391,6 +391,8 @@ a { color: var(--accent); text-decoration: none; }
|
||||
}
|
||||
.mk-nuke-fac:hover { transform: scale(1.4); }
|
||||
.mk-nuke-fac.hot { animation: glow-gold 2s ease-in-out infinite; }
|
||||
.pipeline-line { cursor: pointer; transition: opacity 0.15s; }
|
||||
.pipeline-line:hover { opacity: 1 !important; stroke-width: 3 !important; }
|
||||
@keyframes glow-gold {
|
||||
0%,100% { box-shadow: 0 0 4px rgba(240,184,64,0.3); }
|
||||
50% { box-shadow: 0 0 16px rgba(240,184,64,0.7); }
|
||||
@@ -595,8 +597,8 @@ $('#drawerToggle').addEventListener('click', function() {
|
||||
});
|
||||
|
||||
// ════════════ DETAIL MODAL ════════════
|
||||
var COLORS = { earthquake: 'var(--red)', military: 'var(--blue)', conflict: 'var(--amber)', fire: 'var(--gold)', convergence: 'var(--purple)', nuclear: 'var(--green)', military_base: 'var(--teal)', port: 'var(--blue)', nuclear_facility: 'var(--gold)', pipeline: 'var(--purple)' };
|
||||
var LABELS = { earthquake: 'EARTHQUAKE', military: 'MILITARY AIRCRAFT', conflict: 'CONFLICT EVENT', fire: 'WILDFIRE CLUSTER', convergence: 'SIGNAL CONVERGENCE', nuclear: 'NUCLEAR MONITOR', military_base: 'MILITARY BASE', port: 'STRATEGIC PORT', nuclear_facility: 'NUCLEAR FACILITY', pipeline: 'PIPELINE' };
|
||||
var COLORS = { earthquake: 'var(--red)', military: 'var(--blue)', conflict: 'var(--amber)', fire: 'var(--gold)', convergence: 'var(--purple)', nuclear: 'var(--green)', military_base: 'var(--teal)', port: 'var(--blue)', nuclear_facility: 'var(--gold)', pipeline: 'var(--purple)', exposure: '#e040fb' };
|
||||
var LABELS = { earthquake: 'EARTHQUAKE', military: 'MILITARY AIRCRAFT', conflict: 'CONFLICT EVENT', fire: 'WILDFIRE CLUSTER', convergence: 'SIGNAL CONVERGENCE', nuclear: 'NUCLEAR MONITOR', military_base: 'MILITARY BASE', port: 'STRATEGIC PORT', nuclear_facility: 'NUCLEAR FACILITY', pipeline: 'PIPELINE', exposure: 'POPULATION EXPOSURE' };
|
||||
|
||||
function showDetail(type, d) {
|
||||
var title = '', subtitle = '', fields = [], link = '';
|
||||
@@ -720,6 +722,17 @@ function showDetail(type, d) {
|
||||
['Notes', d.notes || '\u2014'],
|
||||
['Coordinates', d.latitude != null ? d.latitude.toFixed(3) + ', ' + d.longitude.toFixed(3) : '\u2014']
|
||||
];
|
||||
} else if (type === 'exposure') {
|
||||
title = d.city || 'Exposed City';
|
||||
subtitle = (d.country || '') + ' \u2014 ' + (d.population || '');
|
||||
fields = [
|
||||
['City', d.city || '\u2014'],
|
||||
['Country', d.country || '\u2014'],
|
||||
['Population', d.population || '\u2014'],
|
||||
['Nearest Event', d.nearest_event || '\u2014', d.nearest_event === 'conflict' ? 'crit' : 'high'],
|
||||
['Event Detail', d.event_detail || '\u2014'],
|
||||
['Distance', d.distance_km != null ? d.distance_km + ' km' : '\u2014']
|
||||
];
|
||||
}
|
||||
|
||||
var hdr = '<div class="detail-cat"><span class="detail-cat-dot" style="background:' + c + ';--cat-color:' + c + '"></span><span class="detail-cat-label">' + esc(LABELS[type] || type) + '</span></div>' +
|
||||
@@ -1120,6 +1133,28 @@ function updateMapInfra(data) {
|
||||
});
|
||||
mk.addTo(mapLayers.infra);
|
||||
});
|
||||
// Pipelines (polylines: lat_start/lon_start → lat_end/lon_end)
|
||||
var pipeColors = { oil: '#ff9100', gas: '#4da8ff', hydrogen: '#34d399', lng: '#2dd4bf' };
|
||||
var pipeStatusOpacity = { active: 0.55, destroyed: 0.25, terminated: 0.25, cancelled: 0.2, intermittent: 0.45, reduced: 0.4, stalled: 0.3, construction: 0.35, proposed: 0.2 };
|
||||
var pipes = (data.pipelines && data.pipelines.pipelines) || [];
|
||||
pipes.forEach(function(p) {
|
||||
if (p.lat_start == null || p.lon_start == null || p.lat_end == null || p.lon_end == null) return;
|
||||
total++;
|
||||
var color = pipeColors[p.type] || '#a78bfa';
|
||||
var opacity = pipeStatusOpacity[p.status] || 0.4;
|
||||
var dash = (p.status === 'proposed' || p.status === 'construction') ? '6 4' : (p.status === 'destroyed' || p.status === 'terminated' || p.status === 'cancelled') ? '3 6' : null;
|
||||
var opts = { color: color, weight: 2, opacity: opacity, className: 'pipeline-line' };
|
||||
if (dash) opts.dashArray = dash;
|
||||
var line = L.polyline([[p.lat_start, p.lon_start], [p.lat_end, p.lon_end]], opts);
|
||||
line.bindTooltip(esc(p.name) + '<br><span style="opacity:0.7">' + esc(p.type) + ' \u2022 ' + esc(p.status) + '</span>', { className: 'mk-tip', sticky: true });
|
||||
line.on('click', function() {
|
||||
showDetail('pipeline', {
|
||||
name: p.name, route: p.route, type: p.type,
|
||||
capacity: p.capacity, status: p.status, notes: p.notes
|
||||
});
|
||||
});
|
||||
line.addTo(mapLayers.infra);
|
||||
});
|
||||
$('#cntInfra').textContent = total;
|
||||
}
|
||||
|
||||
@@ -1954,6 +1989,20 @@ function connectSSE() {
|
||||
|
||||
// ════════════ BOOT ════════════
|
||||
initMap();
|
||||
|
||||
// Load static geospatial data immediately (bases, ports, nuclear facilities).
|
||||
// This populates the infrastructure layer without waiting for the full SSE
|
||||
// gather, which can take 30-60s when external APIs are slow.
|
||||
fetch('/api/static')
|
||||
.then(function(r) { return r.json(); })
|
||||
.then(function(data) {
|
||||
try { updateMapInfra(data); } catch(e) { console.warn('static infra failed:', e); }
|
||||
// Dismiss loading overlay once static infra is rendered — the map is
|
||||
// usable with 158 items while live feeds continue loading via SSE.
|
||||
document.getElementById('loading').classList.add('gone');
|
||||
})
|
||||
.catch(function(e) { console.warn('static fetch failed:', e); });
|
||||
|
||||
connectSSE();
|
||||
</script>
|
||||
</body>
|
||||
|
||||
@@ -22,15 +22,18 @@ async def _safe(coro, label: str) -> dict:
|
||||
return {}
|
||||
|
||||
|
||||
def _fleet_readiness(theaters: list, waterways: list, surges: list) -> tuple[str, int]:
|
||||
def _fleet_readiness(theaters: dict, waterways: list, surges: list) -> tuple[str, int]:
|
||||
"""Assess overall fleet readiness from component data.
|
||||
|
||||
*theaters* is a dict keyed by theater name (e.g. ``{"europe": {"count": 5, ...}}``).
|
||||
Returns (level, score 0-100).
|
||||
"""
|
||||
score = 0.0
|
||||
|
||||
# Theater activity: more aircraft = higher activity
|
||||
total_aircraft = sum(t.get("aircraft_count", 0) for t in theaters)
|
||||
total_aircraft = sum(
|
||||
t.get("count", 0) for t in theaters.values() if isinstance(t, dict)
|
||||
)
|
||||
score += min(30.0, total_aircraft * 0.5)
|
||||
|
||||
# Waterway status: elevated/critical waterways raise score
|
||||
@@ -81,19 +84,23 @@ async def fetch_fleet_report(fetcher) -> dict:
|
||||
naval_base_count = naval_bases.get("count", 0)
|
||||
|
||||
# Extract key data
|
||||
theaters = posture_data.get("theaters", [])
|
||||
theaters = posture_data.get("theaters", {})
|
||||
if not isinstance(theaters, dict):
|
||||
theaters = {}
|
||||
waterways = vessel_data.get("waterways", [])
|
||||
surges = surge_data.get("surges", [])
|
||||
|
||||
# Compute fleet readiness
|
||||
readiness_level, readiness_score = _fleet_readiness(theaters, waterways, surges)
|
||||
|
||||
# Theater summary
|
||||
# Theater summary — theaters is {name: {count, countries, top_types, ...}}
|
||||
theater_summary = []
|
||||
for t in theaters:
|
||||
for name, t in theaters.items():
|
||||
if not isinstance(t, dict):
|
||||
continue
|
||||
theater_summary.append({
|
||||
"name": t.get("name", "Unknown"),
|
||||
"aircraft_count": t.get("aircraft_count", 0),
|
||||
"name": name,
|
||||
"aircraft_count": t.get("count", 0),
|
||||
"top_types": t.get("top_types", [])[:3],
|
||||
})
|
||||
|
||||
@@ -126,7 +133,9 @@ async def fetch_fleet_report(fetcher) -> dict:
|
||||
"active_surges": active_surges,
|
||||
"surge_count": len(active_surges),
|
||||
"naval_base_count": naval_base_count,
|
||||
"total_tracked_aircraft": sum(t.get("aircraft_count", 0) for t in theaters),
|
||||
"total_tracked_aircraft": sum(
|
||||
t.get("count", 0) for t in theaters.values() if isinstance(t, dict)
|
||||
),
|
||||
"source": "fleet-activity-report",
|
||||
"timestamp": datetime.now(timezone.utc).isoformat(),
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user