initial prototype based on nat-as-server

scripts/generate_dev_fixture.py

@@ -0,0 +1,237 @@
+#!/usr/bin/env python3
+"""Generate dev-test-fixture.json for admin Dev tab import."""
+import json
+import random
+from datetime import datetime, timedelta
+from pathlib import Path
+
+PREFIX = "dev_"
+PASSWORD = "socialvrlab"
+CLIENT_PREFIX = "DEV-CL-"
+SCALE = 5
+
+# Fixed RNG + anchor time so the file is reproducible but not grid-like.
+RNG = random.Random(42)
+ANCHOR = datetime(2026, 5, 27, 15, 30, 0)
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+
+FALLBACK_QUESTIONNAIRE_IDS = [
+    "questionnaire_1_demographic_information",
+    "questionnaire_2_rhs",
+    "questionnaire_3_integration_index",
+    "questionnaire_4_consultation_results",
+    "questionnaire_5_final_interview",
+    "questionnaire_6_follow_up_survey",
+]
+
+NUM_ADMINS = 2 * SCALE
+NUM_SUPERVISORS = 3 * SCALE
+NUM_COACHES = 20 * SCALE
+NUM_CLIENTS = 100 * SCALE
+CLIENTS_WITHOUT_COMPLETIONS = 15 * SCALE
+
+# Roughly how many clients (with a coach) get at least one questionnaire.
+COMPLETION_PARTICIPATION = 0.82
+
+
+def resolve_bundle_path() -> Path | None:
+    """Use the newest questionnaires_bundle_*.json in the repo root."""
+    candidates = sorted(
+        REPO_ROOT.glob("questionnaires_bundle_*.json"),
+        key=lambda p: p.stat().st_mtime,
+        reverse=True,
+    )
+    return candidates[0] if candidates else None
+
+
+def load_questionnaire_ids(bundle_path: Path | None) -> list[str]:
+    if bundle_path is None or not bundle_path.is_file():
+        return FALLBACK_QUESTIONNAIRE_IDS
+    bundle = json.loads(bundle_path.read_text(encoding="utf-8"))
+    ids = [
+        item["questionnaire"]["questionnaireID"]
+        for item in bundle.get("questionnaires", [])
+        if item.get("questionnaire", {}).get("questionnaireID")
+    ]
+    return ids or FALLBACK_QUESTIONNAIRE_IDS
+
+
+def ts_days_ago(min_days: float, max_days: float) -> int:
+    days = RNG.uniform(min_days, max_days)
+    return int((ANCHOR - timedelta(days=days)).timestamp())
+
+
+def variant_seed(client_code: str, qn_id: str, version_index: int) -> int:
+    raw = f"{client_code}:{qn_id}:v{version_index}"
+    return int.from_bytes(raw.encode(), "big") % 100_000
+
+
+def assign_coaches(clients: list[dict], coaches: list[dict]) -> None:
+    """Uneven coach load: a few busy coaches, several light ones."""
+    coach_names = [c["username"] for c in coaches]
+    weights = []
+    for i, _ in enumerate(coach_names):
+        # Coaches 1–4 and 7, 12 get more clients; tail is lighter.
+        base = RNG.uniform(0.4, 1.0)
+        if (i % 7) in (0, 1, 2):
+            base *= RNG.uniform(2.0, 4.5)
+        if i % 11 == 0:
+            base *= RNG.uniform(0.15, 0.45)
+        weights.append(base)
+
+    for row in clients:
+        row["coach"] = RNG.choices(coach_names, weights=weights, k=1)[0]
+
+
+def build_completions(
+    questionnaire_ids: list[str],
+    clients_with_data: list[str],
+) -> list[dict]:
+    """
+    Natural-ish completion patterns:
+    - Sequential questionnaire funnel (not random qn per slot)
+    - Irregular timestamps and gaps between uploads
+    - ~20% of client/qn pairs have 2–4 archived versions
+    """
+    completions: list[dict] = []
+    qn_count = len(questionnaire_ids)
+
+    # How many questionnaires a client completes (skewed toward mid funnel).
+    qn_count_weights = [8, 14, 22, 24, 18, 10, 4][:qn_count]
+    if len(qn_count_weights) < qn_count:
+        qn_count_weights += [2] * (qn_count - len(qn_count_weights))
+
+    for client_code in clients_with_data:
+        if RNG.random() > COMPLETION_PARTICIPATION:
+            continue
+
+        num_qn = RNG.choices(
+            list(range(1, qn_count + 1)),
+            weights=qn_count_weights[:qn_count],
+            k=1,
+        )[0]
+        selected = questionnaire_ids[:num_qn]
+
+        # First activity somewhere in the last ~6 months; some clients are very recent.
+        cursor = ts_days_ago(3, 185)
+        if RNG.random() < 0.12:
+            cursor = ts_days_ago(0.5, 14)
+
+        for qn_id in selected:
+            version_count = 1
+            roll = RNG.random()
+            if roll < 0.08 and num_qn >= 2:
+                version_count = 4
+            elif roll < 0.20:
+                version_count = 3
+            elif roll < 0.38:
+                version_count = 2
+
+            uploads = []
+            for v in range(version_count):
+                if v > 0:
+                    # Re-upload after days or weeks (sometimes same day correction).
+                    if RNG.random() < 0.18:
+                        gap_sec = RNG.randint(2 * 3600, 36 * 3600)
+                    else:
+                        gap_sec = RNG.randint(2 * 86400, 55 * 86400)
+                    cursor += gap_sec
+                else:
+                    # First upload for this qn: often a few days after previous qn.
+                    cursor += RNG.randint(0, 12 * 86400)
+
+                duration = RNG.randint(4 * 60, 55 * 60)
+                started = cursor - duration
+                completed = cursor + RNG.randint(0, 120)
+
+                uploads.append({
+                    "submittedAt": completed,
+                    "startedAt": started,
+                    "variant": variant_seed(client_code, qn_id, v),
+                })
+
+            completions.append({
+                "clientCode": client_code,
+                "questionnaireID": qn_id,
+                "versions": version_count,
+                "uploads": uploads,
+            })
+
+    RNG.shuffle(completions)
+    return completions
+
+
+def main():
+    bundle_path = resolve_bundle_path()
+    questionnaire_ids = load_questionnaire_ids(bundle_path)
+
+    admins = [
+        {"username": f"{PREFIX}admin_{i}", "location": f"Dev Admin Standort {i}"}
+        for i in range(1, NUM_ADMINS + 1)
+    ]
+    supervisors = [
+        {"username": f"{PREFIX}supervisor_{i}", "location": f"Dev Supervisor Region {i}"}
+        for i in range(1, NUM_SUPERVISORS + 1)
+    ]
+    coaches = []
+    for i in range(1, NUM_COACHES + 1):
+        sv = (i - 1) % NUM_SUPERVISORS + 1
+        coaches.append({
+            "username": f"{PREFIX}coach_{i:02d}",
+            "supervisor": f"{PREFIX}supervisor_{sv}",
+        })
+
+    clients = [
+        {"clientCode": f"{CLIENT_PREFIX}{i:04d}"}
+        for i in range(1, NUM_CLIENTS + 1)
+    ]
+    assign_coaches(clients, coaches)
+
+    all_codes = [c["clientCode"] for c in clients]
+    no_data_codes = set(RNG.sample(all_codes, min(CLIENTS_WITHOUT_COMPLETIONS, len(all_codes))))
+    clients_with_data = [c for c in all_codes if c not in no_data_codes]
+
+    completions = build_completions(questionnaire_ids, clients_with_data)
+    total_uploads = sum(c["versions"] for c in completions)
+    multi_version_pairs = sum(1 for c in completions if c["versions"] > 1)
+
+    bundle_note = bundle_path.name if bundle_path and bundle_path.is_file() else "fallback questionnaire list"
+
+    fixture = {
+        "fixtureVersion": 2,
+        "description": (
+            f"Dev/test users, clients, completions with uneven coach load and upload history "
+            f"({SCALE}x scale). Questionnaires from {bundle_note}. Password: socialvrlab."
+        ),
+        "prefix": PREFIX,
+        "defaultPassword": PASSWORD,
+        "sourceBundle": bundle_note,
+        "questionnaireIDs": questionnaire_ids,
+        "admins": admins,
+        "supervisors": supervisors,
+        "coaches": coaches,
+        "clients": clients,
+        "completions": completions,
+        "stats": {
+            "scale": SCALE,
+            "admins": len(admins),
+            "supervisors": len(supervisors),
+            "coaches": len(coaches),
+            "clients": len(clients),
+            "clientsWithoutCompletions": CLIENTS_WITHOUT_COMPLETIONS,
+            "completionRecords": len(completions),
+            "totalUploads": total_uploads,
+            "multiVersionPairs": multi_version_pairs,
+            "questionnaires": len(questionnaire_ids),
+        },
+    }
+
+    out = REPO_ROOT / "dev-test-fixture.json"
+    out.write_text(json.dumps(fixture, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    print(f"Wrote {out}")
+    print(json.dumps(fixture["stats"], indent=2))
+
+
+if __name__ == "__main__":
+    main()