238 lines
7.9 KiB
Python
238 lines
7.9 KiB
Python
#!/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()
|