Score every Ecobalyse activity with VoLCA

"""VoLCA Ecobalyse import - load all SimaPro CSV databases + EF 3.1 scoring.

Usage:
    cd volca-deploy/volca/examples
    uv run ecobalyse/ecobalyse_import.py

https://www.volca.run/docs/python/
"""

import time
import tomllib
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

from volca import Client, Server, VoLCAError

HERE = Path(__file__).resolve().parent
METHOD = "Environmental Footprint 3.1 (adapted)"


def load_config(path):
    with open(path, "rb") as f:
        return tomllib.load(f)


def main():
    config_path = HERE / "volca-ecobalyse.toml"
    config = load_config(config_path)

    databases = config["databases"]
    methods = config["methods"]

    print("=" * 70)
    print("  VoLCA Demo - SimaPro CSV databases + EF 3.1 scoring")
    print("=" * 70)

    # --- Start server ---
    print(f"\n► Starting VoLCA server (port {config['server']['port']})...")
    binary = config_path.parent / config["server"]["binary"]
    srv = Server(config=str(config_path), binary=str(binary))
    srv.start(wait_timeout=30)
    client = Client(base_url=srv.base_url, password=srv.password)
    print(f"  Server ready at {srv.base_url}")

    # --- Show configured databases ---
    print(f"\n► {len(databases)} databases configured:")
    for db in databases:
        deps = db.get("depends", [])
        dep_info = f" (depends on {', '.join(deps)})" if deps else ""
        print(f"  · {db['displayName']}{dep_info}")

    # --- Show configured methods ---
    print(f"\n► {len(methods)} method(s) configured:")
    for m in methods:
        scoring_sets = m.get("scoring", [])
        scoring_info = f" - scoring: {', '.join(s['name'] for s in scoring_sets)}" if scoring_sets else ""
        print(f"  · {m['name']}{scoring_info}")

    # --- Load databases ---
    print("\n► Loading databases into memory...")
    for db in databases:
        name, display = db["name"], db["displayName"]
        print(f"  Loading {display}...", end=" ", flush=True)
        t = time.perf_counter()
        try:
            client.load_database(name)
            print(f"done ({time.perf_counter() - t:.1f}s)")
        except VoLCAError as e:
            print(f"FAILED ({e})")

    # --- Search for a test activity ---
    test_db = "agribalyse-3-2"
    query = "mozzarella"
    print(f"\n► Searching '{query}' in {test_db}...")
    activity = client.use(test_db).search_activities(name=query, limit=1)[0]
    print(f"  Found: {activity.activity_name}")
    print(f"  Process ID: {activity.process_id}")

    # --- Compute impacts ---
    print(f"\n► Computing LCIA impacts ({METHOD})...")
    result = client.use(test_db).get_impacts_batch(activity.process_id, collection=METHOD)

    print(f"  {len(result.results)} impact categories computed\n")
    print(f"  {'Category':<50s} {'Score':>12s}  Unit")
    print(f"  {'-' * 76}")
    for cat in result.results:
        print(f"  {cat.category:<50s} {cat.score:12.4e}  {cat.unit}")

    # --- Scoring results ---
    if result.scoring_results:
        print("\n► Scoring results (formula-based single scores):")
        for set_name, scores in result.scoring_results.items():
            unit = result.scoring_units.get(set_name, "Pt")
            for score_name, value in scores.items():
                label = set_name if len(scores) == 1 else f"{set_name}.{score_name}"
                print(f"  {label} = {value:.2f} {unit}")

    # --- Pre-warm method tables for every loaded database (parallel) ---
    # First /impacts hit on a (db, method-collection) pair pays the
    # regionalized weight precomputation up-front. Doing it here, in
    # parallel across DBs, moves that cost OUT of any later batch run -
    # ecobalyse_export.py then measures only the steady-state per-pid
    # characterization.
    print(f"\n► Warming up {len(databases)} databases (one /impacts call each, parallel)...")
    warm_start = time.perf_counter()

    def _warm_one(db_cfg):
        name, display = db_cfg["name"], db_cfg["displayName"]
        t0 = time.perf_counter()
        c = client.use(name)
        # Grab any activity to hand to /impacts; the work is per-(db, method),
        # not per-pid, so the choice of pid doesn't matter.
        results = c.search_activities(limit=1)
        if not results:
            return display, "no activities", time.perf_counter() - t0
        try:
            c.score_activities([results[0].process_id], collection=METHOD, top_flows=0)
            status = "ok"
        except VoLCAError as e:
            status = str(e)
        return display, status, time.perf_counter() - t0

    with ThreadPoolExecutor(max_workers=len(databases)) as pool:
        futures = {pool.submit(_warm_one, db): db for db in databases}
        for fut in as_completed(futures):
            display, status, dt = fut.result()
            print(f"  [{dt:5.1f}s] {display}: {status}")
    print(f"  Warmup wall-clock: {time.perf_counter() - warm_start:.1f}s")

    print()


if __name__ == "__main__":
    main()

"""VoLCA Ecobalyse export - compute impacts for all Ecobalyse activities.

Reads activities from ecobalyse-data4/activities.json, connects to a running
VoLCA server (started by ecobalyse_import.py), computes EF 3.1 impacts + ECS
score for each process, and writes results in Ecobalyse JSON format.

Usage:
    cd volca-deploy/volca/examples
    uv run ecobalyse/ecobalyse_import.py   # start server + load databases
    uv run ecobalyse/ecobalyse_export.py   # compute impacts (in another terminal)

https://www.volca.run/docs/python/
"""

import argparse
import json
import sys
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

from volca import Client, VoLCAError

HERE = Path(__file__).resolve().parent
OUTPUT_JSON = HERE / "ecobalyse_activities.json"
BASE_URL = "http://localhost:8080"
METHOD = "Environmental Footprint 3.1 (adapted)"
WORKERS = 8

# Source field in lci_catalog activity JSON → DB slug loaded in the running server.
# "Ecobalyse" and "Custom" sources are virtual aggregates without a backing
# SimaPro/EcoSpold export - they're skipped silently here.
SOURCE_TO_DB = {
    "Agribalyse 3.2": "agribalyse-3-2",
    "Ecoinvent 3.9.1": "ecoinvent-3-9-1-adapted",
    "Ecoinvent 3.11": "ecoinvent-3-11-adapted",
    "Ginko 2025": "ginko-2025-v2",
    "PastoEco": "pastoeco",
    "Woolmark": "woolmark",
    "WFLDB": "wfldb",
}

# EF 3.1 impact category name → Ecobalyse trigram
CATEGORY_TO_TRIGRAM = {
    "Acidification": "acd",
    "Climate change": "cch",
    "Eutrophication, freshwater": "fwe",
    "Eutrophication, marine": "swe",
    "Eutrophication, terrestrial": "tre",
    "Ecotoxicity, freshwater - part 1": "etf1",
    "Ecotoxicity, freshwater - part 2": "etf2",
    "Human toxicity, cancer": "htc",
    "Human toxicity, non-cancer": "htn",
    "Ionising radiation": "ior",
    "Land use": "ldu",
    "Ozone depletion": "ozd",
    "Particulate matter": "pma",
    "Photochemical ozone formation": "pco",
    "Resource use, fossils": "fru",
    "Resource use, minerals and metals": "mru",
    "Water use": "wtu",
}


def check_server(client):
    """Verify the VoLCA server is running and return loaded database names."""
    try:
        client.get_version()
    except Exception:
        print("ERROR: VoLCA server not running. Start it first with:")
        print("  uv run ecobalyse_import.py")
        sys.exit(1)

    return {db.name for db in client.list_databases() if db.status == "loaded"}


def resolve_pid(client, db_name, activity):
    """Resolve one activity name to a processId. Returns (activity, pid_or_None, status)."""
    try:
        found = client.use(db_name).search_activities(
            name=activity["activityName"], limit=1
        ).results
    except VoLCAError:
        return activity, None, "failed"
    if not found:
        return activity, None, "not_found"
    return activity, found[0].process_id, "ok"


def extract_impacts(batch_result):
    """Map an LCIABatchResult to a {trigram: score} dict."""
    impacts = {}
    for cat in batch_result.results:
        trigram = CATEGORY_TO_TRIGRAM.get(cat.category)
        if trigram:
            impacts[trigram] = cat.score
    for set_name, scores in batch_result.scoring_results.items():
        if set_name == "ECS":
            impacts["ecs"] = scores.get("total", 0)
        elif set_name == "ecobalyse":
            for score_name, value in scores.items():
                impacts[score_name] = value
    return impacts


def batch_impacts(client, db_name, pid_to_activity):
    """Score all pids for this DB in one MUMPS multi-RHS solve.
    Returns {pid: impacts_dict}. Missing pids → empty dict entries filled by caller.

    Sends `top_flows=0` to skip the O(|inventory|×|methods|) contribution walk
    we don't consume; raw scores + scoring sets are unaffected."""
    if not pid_to_activity:
        return {}
    scored = client.use(db_name).score_activities(
        list(pid_to_activity.keys()), collection=METHOD, top_flows=0
    )
    return {sa.process_id: extract_impacts(sa.impacts) for sa in scored.results}


def build_record(activity, impacts):
    return {
        "activityName": activity["activityName"],
        "comment": activity.get("comment", ""),
        "displayName": activity.get("displayName", ""),
        "impacts": {k: format_number(v) for k, v in sorted(impacts.items())},
        "location": activity.get("location", ""),
        "massPerUnit": activity.get("massPerUnit"),
        "scopes": activity.get("scopes", []),
        "source": activity["source"],
        "unit": activity.get("unit", ""),
        "waste": activity.get("waste", 0),
    }


def format_number(v):
    """Format like Ecobalyse: 5 significant digits, use int for zero.

    Tolerates non-numeric server values (e.g. an error string surfaced by a
    scoring formula that referenced a Left-tainted regionalized score). We
    pass them through unchanged so the export still emits a record; the
    downstream consumer can spot and triage non-numeric entries.
    """
    if v is None:
        return None
    if not isinstance(v, (int, float)):
        return v
    if v == 0:
        return 0
    return float(f"{v:.5g}")


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--only-db",
        metavar="NAME",
        action="append",
        default=[],
        help="Restrict to one TOML db name (repeatable, e.g. --only-db ecoinvent-3.11)",
    )
    parser.add_argument(
        "--warmup",
        action="store_true",
        help="Issue a GET /impacts for one activity per DB first to pay factorization outside the timed section",
    )
    parser.add_argument(
        "--lci-catalog",
        metavar="DIR",
        type=Path,
        default=Path("lci_catalog"),
        help="Path to the ecobalyse-data lci_catalog directory (one JSON file per activity under <db>/)",
    )
    args = parser.parse_args()
    if not args.lci_catalog.is_dir():
        parser.error(f"--lci-catalog: no such directory: {args.lci_catalog}")

    # Read input activities - one JSON file per activity under lci_catalog/<db>/
    print(f"► Reading activities from {args.lci_catalog}/<db>/*.json...")
    activities = [
        json.load(open(p))
        for p in sorted(args.lci_catalog.glob("*/*.json"))
    ]
    computable = [
        a
        for a in activities
        if a.get("activityName") and a.get("source") in SOURCE_TO_DB
    ]
    if args.only_db:
        allowed = set(args.only_db)
        computable = [a for a in computable if SOURCE_TO_DB[a["source"]] in allowed]
        print(
            f"  --only-db filter: keeping {len(computable)} activities across {len(allowed)} databases"
        )
    skipped_no_name = sum(1 for a in activities if not a.get("activityName"))
    skipped_unknown_source = sum(
        1
        for a in activities
        if a.get("activityName") and a.get("source") not in SOURCE_TO_DB
    )
    print(f"  {len(activities)} total, {len(computable)} computable")
    print(
        f"  Skipped: {skipped_no_name} without activityName, {skipped_unknown_source} unknown source"
    )

    # Check server is running
    print(f"\n► Connecting to VoLCA server at {BASE_URL}...")
    client = Client(base_url=BASE_URL)
    loaded_dbs = check_server(client)
    print(
        f"  Connected. {len(loaded_dbs)} databases loaded: {', '.join(sorted(loaded_dbs))}"
    )

    # Warn about missing databases
    needed_dbs = {SOURCE_TO_DB[a["source"]] for a in computable}
    missing = needed_dbs - loaded_dbs
    if missing:
        print(f"  WARNING: databases not loaded: {', '.join(sorted(missing))}")
        computable = [a for a in computable if SOURCE_TO_DB[a["source"]] in loaded_dbs]
        print(f"  Reduced to {len(computable)} computable activities")

    # Build (db_name, activity) pairs
    tasks = [(SOURCE_TO_DB[a["source"]], a) for a in computable]

    results = []
    not_found = []
    failed = []
    total = len(tasks)
    total_start = time.perf_counter()

    # Phase 1: resolve names → processIds in parallel (light I/O-bound work)
    print(f"\n► Resolving {total} processIds ({WORKERS} parallel searches)...")
    phase1_start = time.perf_counter()
    by_db = {}  # db_name -> {pid: activity}
    with ThreadPoolExecutor(max_workers=WORKERS) as pool:
        futures = [pool.submit(resolve_pid, client, db, act) for db, act in tasks]
        for done, fut in enumerate(as_completed(futures), start=1):
            activity, pid, status = fut.result()
            display = activity.get("displayName", activity.get("activityName", "?"))
            db_name = SOURCE_TO_DB[activity["source"]]
            if status == "ok":
                by_db.setdefault(db_name, {})[pid] = activity
            elif status == "not_found":
                not_found.append((db_name, display, activity["activityName"]))
            else:
                failed.append((db_name, display))
            if done % 200 == 0 or done == total:
                print(f"  [{done}/{total}]")
    print(f"  Resolved in {time.perf_counter() - phase1_start:.1f}s")

    # Optional warmup: force MUMPS factorization + method flow-mapping cache per DB
    # so Phase 2 measures only the steady-state per-activity cost.
    if args.warmup:
        print(
            f"\n► Warming {len(by_db)} databases (GET /impacts on first pid of each)..."
        )
        warm_start = time.perf_counter()
        for db_name, pid_map in by_db.items():
            first_pid = next(iter(pid_map))
            t = time.perf_counter()
            client.use(db_name).get_impacts_batch(first_pid, collection=METHOD)
            print(f"  [{db_name}] warmup {time.perf_counter() - t:.2f}s")
        print(f"  Warmup total {time.perf_counter() - warm_start:.1f}s")

    # Phase 2: one batch-impacts call per DB, run in parallel across DBs.
    # Different DBs use different MUMPS handles so concurrent solves don't
    # contend; wall-clock becomes max(per-DB) instead of sum.
    print(f"\n► Batch impacts across {len(by_db)} databases (parallel)...")
    phase2_start = time.perf_counter()

    def run_batch(db_name, pid_map):
        t0 = time.perf_counter()
        impacts_by_pid = batch_impacts(client, db_name, pid_map)
        return db_name, pid_map, impacts_by_pid, time.perf_counter() - t0

    with ThreadPoolExecutor(max_workers=max(1, len(by_db))) as pool:
        futures = [pool.submit(run_batch, db, pm) for db, pm in by_db.items()]
        for fut in as_completed(futures):
            db_name, pid_map, impacts_by_pid, elapsed = fut.result()
            print(
                f"  [{db_name}] {len(impacts_by_pid)}/{len(pid_map)} in {elapsed:.1f}s"
                f" ({1000 * elapsed / max(1, len(pid_map)):.0f}ms/process)"
            )
            for pid, activity in pid_map.items():
                impacts = impacts_by_pid.get(pid)
                if impacts is None:
                    failed.append((
                        db_name,
                        activity.get("displayName", activity["activityName"]),
                    ))
                else:
                    results.append(build_record(activity, impacts))
    print(f"  Phase 2 wall-clock: {time.perf_counter() - phase2_start:.1f}s")

    done = len(results) + len(not_found) + len(failed)
    total_time = time.perf_counter() - total_start

    # Write output
    print(f"\n► Writing {len(results)} activities to {OUTPUT_JSON.name}...")
    with open(OUTPUT_JSON, "w", encoding="utf-8") as f:
        json.dump(results, f, indent=2, ensure_ascii=False)
        f.write("\n")

    # Summary
    print(f"\n{'=' * 60}")
    print(
        f"  Results:  {len(results)} computed, {len(not_found)} not found, {len(failed)} failed"
    )
    if done > 0:
        mean_ms = 1000 * total_time / done
        print(f"  Timing:   {total_time:.1f}s total, {mean_ms:.0f}ms mean per process")
    if not_found:
        print(f"\n  Not found ({len(not_found)}):")
        for db, display, name in not_found[:10]:
            print(f"    [{db}] {display}")
        if len(not_found) > 10:
            print(f"    ... and {len(not_found) - 10} more")
    print()


if __name__ == "__main__":
    main()

"""VoLCA EcoFoodChoice export - compute EFC + Biomaps impacts.

Reads activities from ecobalyse-data/lci_catalog/<db>/*.json, connects to a
running VoLCA server (started by ecobalyse_import.py), computes EF 3.1 impacts
+ ECS + EFC + regionalized Biomaps for each process, and writes a JSON file
with one record per activity carrying:

  - the 14 PEF/EF 3.1 trigram scores (raw characterized values)
  - the regionalized Biomaps biodiversity-loss potential ("bio")
  - the EFC single-score total ("efc")

The Biomaps method is regionalized (CFs vary by activity location). VoLCA
resolves the per-activity location → CF cell via the location hierarchy
configured in the TOML's `geographies = "..."`.

Usage:
    cd volca-deploy/volca/examples
    uv run ecobalyse/ecobalyse_import.py     # start server + load + warm-up
    uv run ecobalyse/ecofoodchoice_export.py # compute impacts (in another terminal)

https://www.volca.run/docs/python/
"""

import argparse
import json
import sys
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

from volca import Client, VoLCAError

HERE = Path(__file__).resolve().parent
OUTPUT_JSON = HERE / "ecofoodchoice_activities.json"
BASE_URL = "http://localhost:8080"
METHOD = "Environmental Footprint 3.1 (adapted)"
WORKERS = 8

# Source field in lci_catalog activity JSON → DB slug loaded in the running server.
# "Ecobalyse" and "Custom" sources are virtual aggregates without a backing
# SimaPro/EcoSpold export - they're skipped silently here.
SOURCE_TO_DB = {
    "Agribalyse 3.2": "agribalyse-3-2",
    "Ecoinvent 3.9.1": "ecoinvent-3-9-1-adapted",
    "Ecoinvent 3.11": "ecoinvent-3-11-adapted",
    "Ginko 2025": "ginko-2025-v2",
    "PastoEco": "pastoeco",
    "Woolmark": "woolmark",
    "WFLDB": "wfldb",
}

# Impact category name → EFC short code (trigrams kept in sync with EF 3.1).
# Biomaps is a single regionalized category; EFC is the formula-based total.
CATEGORY_TO_TRIGRAM = {
    "Acidification": "acd",
    "Climate change": "cch",
    "Eutrophication, freshwater": "fwe",
    "Eutrophication, marine": "swe",
    "Eutrophication, terrestrial": "tre",
    "Ecotoxicity, freshwater - inorganics": "etfi",
    "Ecotoxicity, freshwater - organics - p.1": "etfo1",
    "Ecotoxicity, freshwater - organics - p.2": "etfo2",
    "Ionising radiation": "ior",
    "Land use": "ldu",
    "Ozone depletion": "ozd",
    "Particulate matter": "pma",
    "Photochemical ozone formation": "pco",
    "Resource use, fossils": "fru",
    "Resource use, minerals and metals": "mru",
    "Water use": "wtu",
    "Biomaps - Biodiversity Loss Potential": "bio",
}


def check_server(client):
    """Verify the VoLCA server is running and return loaded database names."""
    try:
        client.get_version()
    except Exception:
        print("ERROR: VoLCA server not running. Start it first with:")
        print("  uv run ecobalyse_import.py")
        sys.exit(1)

    return {db.name for db in client.list_databases() if db.status == "loaded"}


def resolve_pid(client, db_name, activity):
    """Resolve one activity name to a processId. Returns (activity, pid_or_None, status)."""
    try:
        found = client.use(db_name).search_activities(
            name=activity["activityName"], limit=1
        ).results
    except VoLCAError:
        return activity, None, "failed"
    if not found:
        return activity, None, "not_found"
    return activity, found[0].process_id, "ok"


def extract_impacts(batch_result):
    """Map an LCIABatchResult to an EFC-shaped {trigram: score} dict.

    Carries the 14 PEF-mapped EF 3.1 categories + the regionalized Biomaps
    score + the EFC formula total. The raw inorganic / organics-p.1 / -p.2
    Ecotoxicity, freshwater categories ship as-is (etfi / etfo1 / etfo2) so
    downstream consumers can re-aggregate the way they want; EFC's own
    `organics = etfo1 + etfo2` happens server-side via the scoring formula
    and is exposed in `scoring_results`.
    """
    impacts = {}
    for cat in batch_result.results:
        trigram = CATEGORY_TO_TRIGRAM.get(cat.category)
        if trigram:
            impacts[trigram] = cat.score
    for set_name, scores in batch_result.scoring_results.items():
        if set_name == "EFC":
            impacts["efc"] = scores.get("total", 0)
        elif set_name == "ECS":
            impacts["ecs"] = scores.get("total", 0)
        elif set_name == "PEF":
            impacts["pef"] = scores.get("total", 0)
    return impacts


def batch_impacts(client, db_name, pid_to_activity):
    """Score all pids for this DB in one MUMPS multi-RHS solve.
    Returns {pid: impacts_dict}. Missing pids → empty dict entries filled by caller.

    Sends `top_flows=0` to skip the O(|inventory|×|methods|) contribution walk
    we don't consume; raw scores + scoring sets are unaffected."""
    if not pid_to_activity:
        return {}
    scored = client.use(db_name).score_activities(
        list(pid_to_activity.keys()), collection=METHOD, top_flows=0
    )
    return {sa.process_id: extract_impacts(sa.impacts) for sa in scored.results}


def build_record(activity, impacts):
    return {
        "activityName": activity["activityName"],
        "comment": activity.get("comment", ""),
        "displayName": activity.get("displayName", ""),
        "impacts": {k: format_number(v) for k, v in sorted(impacts.items())},
        "location": activity.get("location", ""),
        "massPerUnit": activity.get("massPerUnit"),
        "scopes": activity.get("scopes", []),
        "source": activity["source"],
        "unit": activity.get("unit", ""),
        "waste": activity.get("waste", 0),
    }


def format_number(v):
    """Format like Ecobalyse: 5 significant digits, use int for zero.

    Tolerates non-numeric server values (e.g. an error string surfaced by a
    scoring formula that referenced a Left-tainted regionalized score). We
    pass them through unchanged so the export still emits a record; the
    downstream consumer can spot and triage non-numeric entries.
    """
    if v is None:
        return None
    if not isinstance(v, (int, float)):
        return v
    if v == 0:
        return 0
    return float(f"{v:.5g}")


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--only-db",
        metavar="NAME",
        action="append",
        default=[],
        help="Restrict to one TOML db name (repeatable, e.g. --only-db ecoinvent-3.11)",
    )
    parser.add_argument(
        "--warmup",
        action="store_true",
        help="Issue a GET /impacts for one activity per DB first to pay factorization outside the timed section",
    )
    parser.add_argument(
        "--lci-catalog",
        metavar="DIR",
        type=Path,
        default=Path("lci_catalog"),
        help="Path to the ecobalyse-data lci_catalog directory (one JSON file per activity under <db>/)",
    )
    args = parser.parse_args()
    if not args.lci_catalog.is_dir():
        parser.error(f"--lci-catalog: no such directory: {args.lci_catalog}")

    # Read input activities - one JSON file per activity under lci_catalog/<db>/
    print(f"► Reading activities from {args.lci_catalog}/<db>/*.json...")
    activities = [
        json.load(open(p))
        for p in sorted(args.lci_catalog.glob("*/*.json"))
    ]
    computable = [
        a
        for a in activities
        if a.get("activityName") and a.get("source") in SOURCE_TO_DB
    ]
    if args.only_db:
        allowed = set(args.only_db)
        computable = [a for a in computable if SOURCE_TO_DB[a["source"]] in allowed]
        print(
            f"  --only-db filter: keeping {len(computable)} activities across {len(allowed)} databases"
        )
    skipped_no_name = sum(1 for a in activities if not a.get("activityName"))
    skipped_unknown_source = sum(
        1
        for a in activities
        if a.get("activityName") and a.get("source") not in SOURCE_TO_DB
    )
    print(f"  {len(activities)} total, {len(computable)} computable")
    print(
        f"  Skipped: {skipped_no_name} without activityName, {skipped_unknown_source} unknown source"
    )

    # Check server is running
    print(f"\n► Connecting to VoLCA server at {BASE_URL}...")
    client = Client(base_url=BASE_URL)
    loaded_dbs = check_server(client)
    print(
        f"  Connected. {len(loaded_dbs)} databases loaded: {', '.join(sorted(loaded_dbs))}"
    )

    # Warn about missing databases
    needed_dbs = {SOURCE_TO_DB[a["source"]] for a in computable}
    missing = needed_dbs - loaded_dbs
    if missing:
        print(f"  WARNING: databases not loaded: {', '.join(sorted(missing))}")
        computable = [a for a in computable if SOURCE_TO_DB[a["source"]] in loaded_dbs]
        print(f"  Reduced to {len(computable)} computable activities")

    # Build (db_name, activity) pairs
    tasks = [(SOURCE_TO_DB[a["source"]], a) for a in computable]

    results = []
    not_found = []
    failed = []
    total = len(tasks)
    total_start = time.perf_counter()

    # Phase 1: resolve names → processIds in parallel (light I/O-bound work)
    print(f"\n► Resolving {total} processIds ({WORKERS} parallel searches)...")
    phase1_start = time.perf_counter()
    by_db = {}  # db_name -> {pid: activity}
    with ThreadPoolExecutor(max_workers=WORKERS) as pool:
        futures = [pool.submit(resolve_pid, client, db, act) for db, act in tasks]
        for done, fut in enumerate(as_completed(futures), start=1):
            activity, pid, status = fut.result()
            display = activity.get("displayName", activity.get("activityName", "?"))
            db_name = SOURCE_TO_DB[activity["source"]]
            if status == "ok":
                by_db.setdefault(db_name, {})[pid] = activity
            elif status == "not_found":
                not_found.append((db_name, display, activity["activityName"]))
            else:
                failed.append((db_name, display))
            if done % 200 == 0 or done == total:
                print(f"  [{done}/{total}]")
    print(f"  Resolved in {time.perf_counter() - phase1_start:.1f}s")

    # Optional warmup: force MUMPS factorization + method flow-mapping cache per DB
    # so Phase 2 measures only the steady-state per-activity cost.
    if args.warmup:
        print(
            f"\n► Warming {len(by_db)} databases (GET /impacts on first pid of each)..."
        )
        warm_start = time.perf_counter()
        for db_name, pid_map in by_db.items():
            first_pid = next(iter(pid_map))
            t = time.perf_counter()
            client.use(db_name).get_impacts_batch(first_pid, collection=METHOD)
            print(f"  [{db_name}] warmup {time.perf_counter() - t:.2f}s")
        print(f"  Warmup total {time.perf_counter() - warm_start:.1f}s")

    # Phase 2: one batch-impacts call per DB, run in parallel across DBs.
    # Different DBs use different MUMPS handles so concurrent solves don't
    # contend; wall-clock becomes max(per-DB) instead of sum.
    print(f"\n► Batch impacts across {len(by_db)} databases (parallel)...")
    phase2_start = time.perf_counter()

    def run_batch(db_name, pid_map):
        t0 = time.perf_counter()
        impacts_by_pid = batch_impacts(client, db_name, pid_map)
        return db_name, pid_map, impacts_by_pid, time.perf_counter() - t0

    with ThreadPoolExecutor(max_workers=max(1, len(by_db))) as pool:
        futures = [pool.submit(run_batch, db, pm) for db, pm in by_db.items()]
        for fut in as_completed(futures):
            db_name, pid_map, impacts_by_pid, elapsed = fut.result()
            print(
                f"  [{db_name}] {len(impacts_by_pid)}/{len(pid_map)} in {elapsed:.1f}s"
                f" ({1000 * elapsed / max(1, len(pid_map)):.0f}ms/process)"
            )
            for pid, activity in pid_map.items():
                impacts = impacts_by_pid.get(pid)
                if impacts is None:
                    failed.append((
                        db_name,
                        activity.get("displayName", activity["activityName"]),
                    ))
                else:
                    results.append(build_record(activity, impacts))
    print(f"  Phase 2 wall-clock: {time.perf_counter() - phase2_start:.1f}s")

    done = len(results) + len(not_found) + len(failed)
    total_time = time.perf_counter() - total_start

    # Write output
    print(f"\n► Writing {len(results)} activities to {OUTPUT_JSON.name}...")
    with open(OUTPUT_JSON, "w", encoding="utf-8") as f:
        json.dump(results, f, indent=2, ensure_ascii=False)
        f.write("\n")

    # Summary
    print(f"\n{'=' * 60}")
    print(
        f"  Results:  {len(results)} computed, {len(not_found)} not found, {len(failed)} failed"
    )
    if done > 0:
        mean_ms = 1000 * total_time / done
        print(f"  Timing:   {total_time:.1f}s total, {mean_ms:.0f}ms mean per process")
    if not_found:
        print(f"\n  Not found ({len(not_found)}):")
        for db, display, name in not_found[:10]:
            print(f"    [{db}] {display}")
        if len(not_found) > 10:
            print(f"    ... and {len(not_found) - 10} more")
    print()


if __name__ == "__main__":
    main()