#!/usr/bin/env python3 """Deterministic scorer for the NGO Blockchain Readiness Framework. Python 3 standard library only. Same behaviour as scorer.js in this folder. python3 scorer.py --answers answers.json --data ../../site/css/data/versions/v1.0-ngo > result.json It ports the site's scoring engine (ngo-p0p1p2.js, ngo-p4.js, ngo-evaluation-engine.js) without a browser: the Phase 3 ranking, the fit and readiness gates, the Phase 4 dimension scores, thresholds and outcome, and the optional detailed evaluation. For every verdict it also says which rule produced it, in a plain sentence. Nothing is sent anywhere. The script reads two things, the answers file and the data folder, and prints JSON. """ import argparse import json import math import os import re import sys from decimal import Decimal, ROUND_HALF_UP SCORER_VERSION = "1.0.0" # --------------------------------------------------------------------------- # JavaScript semantics the engine relies on # --------------------------------------------------------------------------- UNDEF = object() # stands for JavaScript undefined (a missing key) NAN = float("nan") def is_num(v): return isinstance(v, (int, float)) and not isinstance(v, bool) def is_nan(v): return isinstance(v, float) and math.isnan(v) def truthy(v): if v is UNDEF or v is None or v is False: return False if is_num(v): return not (v == 0 or is_nan(v)) if isinstance(v, str): return len(v) > 0 return True # arrays and objects are truthy in JavaScript, even when empty def or0(v): """JavaScript `v || 0` for a number that may be NaN.""" return 0 if (is_nan(v) or v == 0) else v def fmt_num(v): if is_nan(v): return "NaN" if v == math.inf: return "Infinity" if v == -math.inf: return "-Infinity" if v == int(v) and abs(v) < 1e21: return str(int(v)) return repr(v) def js_str(v): if v is UNDEF: return "undefined" if v is None: return "null" if v is True: return "true" if v is False: return "false" if isinstance(v, str): return v if isinstance(v, int): return str(v) if isinstance(v, float): return fmt_num(v) if isinstance(v, list): return ",".join("" if (x is None or x is UNDEF) else js_str(x) for x in v) return "[object Object]" _DEC = re.compile(r"^[+-]?(\d+\.?\d*|\.\d+)([eE][+-]?\d+)?$") _HEX = re.compile(r"^0[xX][0-9a-fA-F]+$") def js_number(v): """JavaScript Number(v).""" if v is UNDEF: return NAN if v is None or v is False: return 0 if v is True: return 1 if is_num(v): return v if isinstance(v, list): v = js_str(v) if isinstance(v, str): s = v.strip() if s == "": return 0 if _DEC.match(s): return float(s) if _HEX.match(s): return int(s, 16) if s in ("Infinity", "+Infinity"): return math.inf if s == "-Infinity": return -math.inf return NAN return NAN _INT = re.compile(r"^[+-]?\d+") _HEXP = re.compile(r"^[+-]?0[xX][0-9a-fA-F]+") def js_parseint(v): """JavaScript parseInt(v) with no radix. NaN when no leading integer.""" s = js_str(v).lstrip() m = _HEXP.match(s) if m: t = m.group(0) neg = t.startswith("-") return -int(t.lstrip("+-"), 16) if neg else int(t.lstrip("+-"), 16) m = _INT.match(s) if m: return int(m.group(0)) return NAN def js_round(x): return math.floor(x + 0.5) def round1(x): return js_round(x * 10) / 10 def round2(x): return js_round(x * 100) / 100 def to_fixed(x, n): q = Decimal(1).scaleb(-n) return str(Decimal(x).quantize(q, rounding=ROUND_HALF_UP)) def is_finite_num(v): return is_num(v) and math.isfinite(v) def js_keys(d): """Object.keys order: integer-like keys ascending, then the rest in insertion order.""" ints, rest = [], [] for k in d.keys(): if isinstance(k, str) and re.match(r"^(0|[1-9]\d{0,9})$", k) and int(k) < 4294967295: ints.append(k) else: rest.append(k) ints.sort(key=int) return ints + rest def lower(s): return js_str(s).lower() def get(d, k, default=UNDEF): if isinstance(d, dict): return d.get(k, default) return default # --------------------------------------------------------------------------- # Constants copied from the engine # --------------------------------------------------------------------------- SECTOR_KEYWORDS = { "humanitarian": ["humanitarian", "relief", "aid", "refugee", "displacement", "emergency response", "crisis"], "environmental": ["environmental", "conservation", "climate", "sustainability", "wildlife", "ecosystem"], "healthcare": ["healthcare", "medical", "clinical", "health", "disease", "hospital"], "microfinance": ["microfinance", "financial inclusion", "lending", "micro-credit", "savings group"], "education": ["education", "school", "literacy", "learning", "teaching", "academic"], "agriculture": ["agriculture", "farming", "food security", "crop", "livestock", "agri"], "human_rights": ["human rights", "advocacy", "justice", "legal aid", "civil liberties", "rights"], "wash": ["water", "sanitation", "hygiene", "wash", "clean water", "latrine"], "community": ["community development", "community-based", "grassroots", "local development"], } SECTOR_UC_FLOORS = { "humanitarian": {"UC-B-024": 2.5, "UC-B-025": 2.0, "UC-C-039": 2.0, "UC-A-001": 2.0, "UC-D-053": 2.0}, "education": {"UC-B-024": 1.5, "UC-C-039": 2.0, "UC-A-001": 1.5}, "agriculture": {"UC-D-053": 2.5, "UC-A-001": 2.0, "UC-B-024": 1.5}, "human_rights": {"UC-C-039": 2.5, "UC-B-024": 2.0}, "wash": {"UC-D-053": 2.5, "UC-C-039": 2.0}, "community": {"UC-B-024": 1.5, "UC-A-001": 2.0}, } DIMENSION_NAMES = { "D1": "Governance and authority", "D2": "Financial resources", "D3": "Operational readiness", "D4": "Technical capability", "D5": "Data readiness", } ACTIVE_DIMS = ["D1", "D2", "D3", "D4", "D5"] DIM_QUESTION_COUNTS = {"D1": 3, "D2": 1, "D3": 3, "D4": 3, "D5": 4} MULTISELECT_SEP = "|~|" FIT_SCORES = {"strong": 4, "moderate": 3, "weak": 2, "negative": 1, "neutral": None, "flag_for_review": None} READINESS_SCORES = {"ready": 4, "moderate": 3, "low": 2, "not_ready": 1, "neutral": None, "flag_for_review": None} EVAL_DIM_MAPPING = { "technicalCapacity": "D4", "stakeholderBuyIn": "D1", "dataReadiness": "D5", "financialCapacity": "D2", "operationalReadiness": "D3", "changeManagement": "D4", "lastMileReadiness": "D3", "deviceConnectivity": "D4", "hardwareConnectivity": "D4", "systemVisibility": "D4", "monitoringCapability": "D4", "systemIntegrationReadiness": "D4", "dataIntegrationReadiness": "D5", "dataCapabilityReadiness": "D5", "recordingTimeliness": "D3", "ruleUpdateSpeed": "D4", "costTolerance": "D2", } EVAL_DIM_LABELS = { "technicalCapacity": "Technical capacity", "stakeholderBuyIn": "Stakeholder buy-in", "dataReadiness": "Data readiness", "financialCapacity": "Financial capacity", "operationalReadiness": "Operational readiness", "changeManagement": "Change management", "lastMileReadiness": "Last-mile readiness", "deviceConnectivity": "Devices and connectivity", "hardwareConnectivity": "Hardware and connectivity", "systemVisibility": "System visibility", "monitoringCapability": "Monitoring capability", "systemIntegrationReadiness": "System integration", "dataIntegrationReadiness": "Data integration", "dataCapabilityReadiness": "Data capability", "recordingTimeliness": "Recording timeliness", "ruleUpdateSpeed": "Rule update speed", "costTolerance": "Cost tolerance", } B_NAMES = { "B1": "a shared record that several parties rely on", "B2": "verification without having to trust the other parties", "B3": "tamper-proof, auditable records", "B4": "removing unnecessary middlemen", "B5": "settlement that cannot be reversed", "B6": "control over who sees which data", "B7": "automated rule enforcement", "B8": "representing assets as digital units", } # --------------------------------------------------------------------------- # Data folder # --------------------------------------------------------------------------- def read_json(path): with open(path, "r", encoding="utf-8") as fh: return json.load(fh) def extract_questions(obj, target, text_keys): """Mirror of the engine's recursive question extractor.""" if obj is None or not isinstance(obj, (dict, list)): return if isinstance(obj, list): for x in obj: extract_questions(x, target, text_keys) return if truthy(obj.get("id", UNDEF)) and any(truthy(obj.get(k, UNDEF)) for k in text_keys): sm = obj.get("severityMapping") sc = obj.get("severityCount") sf = obj.get("severityByField") target[obj["id"]] = { "severityMapping": sm if truthy(sm) else None, "severityCount": sc if truthy(sc) else None, "severityByField": sf if truthy(sf) else None, "type": obj.get("type") if truthy(obj.get("type", UNDEF)) else "freetext", } return for k in js_keys(obj): v = obj[k] if isinstance(v, (dict, list)): extract_questions(v, target, text_keys) class Data: def __init__(self, folder): self.folder = folder p = lambda f: os.path.join(folder, f) self.catalogue = read_json(p("ngo-use-case-catalogue.json")) self.use_cases = self.catalogue.get("useCases", []) self.uc_by_id = {} for uc in self.use_cases: self.uc_by_id.setdefault(uc["id"], uc) # Array.find returns the first match self.scoring_map = read_json(p("uc-scoring-map.json")) self.p0q, self.p1q, self.p2q = {}, {}, {} extract_questions(read_json(p("questions-phase0.json")), self.p0q, ["text", "ngoText", "label"]) extract_questions(read_json(p("questions-phase1.json")), self.p1q, ["text", "ngoText", "label"]) for f in ("questions-phase2-cats-ae.json", "questions-phase2-cats-fj.json"): extract_questions(read_json(p(f)), self.p2q, ["text", "ngoText"]) self.programmes = [] self._companions = {} try: comp = read_json(p("companion-tadamon-cso.json")) for q in comp.get("additionalQuestions", []) or []: if truthy(q.get("id", UNDEF)): sm = q.get("severityMapping") sc = q.get("severityCount") sf = q.get("severityByField") self.p1q[q["id"]] = { "severityMapping": sm if truthy(sm) else None, "severityCount": sc if truthy(sc) else None, "severityByField": sf if truthy(sf) else None, "type": q.get("type") if truthy(q.get("type", UNDEF)) else "freetext", } except (OSError, ValueError): pass try: ov = read_json(p("companion-overlays.json")) prog = (ov.get("indexedCompanions") or {}).get("programme") or {} for name, meta in prog.items(): entry = {"name": name, "meta": meta, "boosts": {}} if meta.get("file"): try: c = read_json(p(meta["file"])) entry["boosts"] = c.get("ucRelevanceBoosts") or {} except (OSError, ValueError): pass self.programmes.append(entry) except (OSError, ValueError): pass self._eval = None def eval_questions(self): if self._eval is None: try: self._eval = read_json(os.path.join(self.folder, "uc-evaluation-questions.json")) except (OSError, ValueError): self._eval = {} return self._eval # --------------------------------------------------------------------------- # Input: the site's own storage shape # --------------------------------------------------------------------------- _UUID_PREFIX = re.compile(r"^[0-9a-fA-F-]{36}:") def parse_json_value(v): if isinstance(v, str): try: return json.loads(v) except ValueError: return v return v def demo_to_storage(raw, notes): """Convert a demo file (answers.phase0 ... phase4) the way the site's demo loader does.""" out = {} answers = raw.get("answers") or {} for n in range(0, 5): layer = answers.get("phase%d" % n) if not layer: continue flat = {} for k, e in layer.items(): if not truthy(e): continue if isinstance(e, dict): flat[e.get("questionId") if truthy(e.get("questionId", UNDEF)) else k] = ( e.get("answer") if truthy(e.get("answer", UNDEF)) else (e.get("content") if truthy(e.get("content", UNDEF)) else "")) elif isinstance(e, str): flat[k] = e out["ngo_assessment_%d" % n] = {"layerId": str(n), "answers": flat} p1 = (out.get("ngo_assessment_1") or {}).get("answers") or {} if "ngo.interagencyContext" in p1: out["ngo.interagencyContext"] = p1["ngo.interagencyContext"] if isinstance(raw.get("selectedCategories"), list): out["irf_ngo_selected_problem_areas"] = raw["selectedCategories"] notes.append("Demo file: selected categories taken from selectedCategories.") p4 = (out.get("ngo_assessment_4") or {}).get("answers") or {} with_part_b = [] for k in p4: m = re.match(r"^NGO-P4-[A-Z]-Q\d+__(UC-[A-Z]-\d+)$", k) if m and m.group(1) not in with_part_b: with_part_b.append(m.group(1)) if with_part_b: out["irf_p3_starred_usecases"] = with_part_b notes.append("Demo file: starred use cases taken from the use cases that have answers about the problem.") elif isinstance(raw.get("selectedUseCases"), list): out["irf_p3_starred_usecases"] = raw["selectedUseCases"] notes.append("Demo file: starred use cases taken from selectedUseCases.") return out class Store: """Read side of the browser's localStorage, for one assessment.""" def __init__(self, raw, notes): if isinstance(raw, dict) and isinstance(raw.get("answers"), dict) and any( str(k).startswith("phase") for k in raw["answers"].keys()): raw = demo_to_storage(raw, notes) self.items = {} if not isinstance(raw, dict): raise ValueError("The answers file must be a JSON object.") for k, v in raw.items(): self.items[_UUID_PREFIX.sub("", k)] = v def get(self, key): v = self.items.get(key, UNDEF) return v def raw_text_or_obj(self, key): """asGet: undefined or empty means 'not stored'.""" v = self.get(key) if v is UNDEF or v is None or v == "": return None return v def blob(self, n): """Parsed assessment blob for phase n, or None. Mirrors asGet('ngo_assessment_n') || asGet('assessment_n').""" raw = self.raw_text_or_obj("ngo_assessment_%d" % n) if raw is None: raw = self.raw_text_or_obj("assessment_%d" % n) if raw is None: return None if isinstance(raw, str): try: return json.loads(raw) except ValueError: return "__bad__" return raw def json_key(self, key, default): v = self.raw_text_or_obj(key) if v is None: return default if isinstance(v, str): try: return json.loads(v) except ValueError: return default return v def text_key(self, key): v = self.raw_text_or_obj(key) return "" if v is None else js_str(v) def answers_of_blob(blob): """`d.answers || d || {}` (the loaders used by Phase 3).""" if blob is None or blob == "__bad__": return {} if isinstance(blob, dict): a = blob.get("answers", UNDEF) if truthy(a): return a if isinstance(a, dict) else {} return blob return {} def split_multi(v): if v is None or v is UNDEF or v == "": return [] if isinstance(v, list): return [x for x in (js_str(s).strip() for s in v) if x] if isinstance(v, dict) and v.get("value", UNDEF) is not UNDEF: return split_multi(v["value"]) s = js_str(v) if MULTISELECT_SEP in s: return [x for x in (t.strip() for t in s.split(MULTISELECT_SEP)) if x] if s[:1] == "[": try: arr = json.loads(s) if isinstance(arr, list): return [x for x in (js_str(t).strip() for t in arr) if x] except ValueError: pass return [x for x in (t.strip() for t in s.split(",")) if x] def answer_value(v): """getAnswerValue: unwrap {value: ...}.""" if v is UNDEF or v is None or v == "": return v if isinstance(v, dict) and v.get("value", UNDEF) is not UNDEF: return v["value"] return v # --------------------------------------------------------------------------- # Phase 3: ranking, gates, tiers # --------------------------------------------------------------------------- class Engine: def __init__(self, data, store): self.d = data self.s = store self.p0 = answers_of_blob(store.blob(0)) self.p1 = answers_of_blob(store.blob(1)) self.p2 = answers_of_blob(store.blob(2)) p4 = store.blob(4) self.p4 = (p4.get("answers") if isinstance(p4, dict) and isinstance(p4.get("answers"), dict) else {}) if p4 else {} self.selected = store.json_key("irf_ngo_selected_problem_areas", None) self.interagency = store.text_key("ngo.interagencyContext") self._warned_unmapped = set() # -- helpers ---------------------------------------------------------- def interagency_boost(self): c = self.interagency if c in ("central", "partial"): return 0.3 if c == "unsure": return 0.1 return 0 @staticmethod def picked_values(val): if isinstance(val, list): return [js_str(x) for x in val] return [x.strip() for x in js_str(val).split("|~|") if x.strip() != ""] def severity_from_shape(self, qinfo, val): """severityCount (multiselect: how many chosen) or severityByField (dual-range). None when the question has neither.""" if qinfo and qinfo.get("severityCount"): n = len([x for x in self.picked_values(val) if not x.startswith("Other")]) for band in qinfo["severityCount"]: if "upTo" not in band or n <= band["upTo"]: return band["severity"] if n > 0 else 0 return 0 if qinfo and qinfo.get("severityByField"): parsed = val if isinstance(val, str): try: parsed = json.loads(val) except ValueError: return 0 if not isinstance(parsed, dict): return 0 best = 0 for f, table in qinfo["severityByField"].items(): got = table.get(js_str(parsed.get(f, UNDEF)), 0) if got > best: best = got return best return None def severity(self, sig): qid = sig["q"] if qid.startswith("NGO-P0-") or qid.startswith("FI-P0-"): val = get(self.p0, qid) qinfo = self.d.p0q.get(qid) elif qid.startswith(("NGO-P1-", "FI-P1-", "TADAMON-P1-", "TADAMON-P0-")): val = get(self.p1, qid) qinfo = self.d.p1q.get(qid) if val is UNDEF: val = get(self.p0, qid) qinfo = qinfo or self.d.p0q.get(qid) else: val = get(self.p2, qid) qinfo = self.d.p2q.get(qid) if val is UNDEF or val is None or val == "": return 0 if isinstance(sig.get("when"), list): picked = self.picked_values(val) for w in sig["when"]: if w in picked: return 4 return 0 if qinfo and qinfo["severityMapping"]: m = qinfo["severityMapping"].get(js_str(val)) if isinstance(qinfo["severityMapping"], dict) else None if truthy(m if m is not None else UNDEF): return m if qinfo and qinfo["type"] == "scale": n = js_number(val) if not is_nan(n): return n shaped = self.severity_from_shape(qinfo, val) if shaped is not None: return shaped if qinfo and qinfo["severityMapping"] is not None: return 0 # a mapped question never falls back if qid.startswith(("NGO-P0-", "NGO-P1-", "TADAMON-")): if qid not in self._warned_unmapped: self._warned_unmapped.add(qid) sys.stderr.write("warning: %s has a scoring signal with no `when` list and no severity rule; it scores 4 for any answer.\n" % qid) return 4 if truthy(val) else 0 return 0 def score_by_role(self, signals, role): weight_sum = 0 value_sum = 0 top = [] for sig in signals: if sig.get("role") != role: continue sev = self.severity(sig) if sev > 0: value_sum += sev * sig["w"] weight_sum += sig["w"] top.append({"q": sig["q"], "role": role, "severity": sev, "weight": sig["w"], "why": sig.get("why")}) score = value_sum / weight_sum if weight_sum > 0 else 0 return {"score": score, "signalCount": len(top), "topSignals": top} def p2_summary(self, cat): prefix = "NGO-P2-Cat" + cat skip = ["_notes", "_priority", "_rag", "_na", "_override", "_override_reason", "_severity", "_rootcause", "_volume"] out = [] for key, val in self.p2.items(): if not key.startswith(prefix): continue if any(key.endswith(s) for s in skip): continue if "SUMMARY" in key or "-followup" in key: continue if not truthy(val): continue qinfo = self.d.p2q.get(key) qtype = qinfo["type"] if qinfo else "unknown" sev = None if qinfo and qinfo["severityMapping"] and truthy(qinfo["severityMapping"].get(js_str(val), UNDEF)): sev = qinfo["severityMapping"][js_str(val)] if qtype in ("select", "multiselect", "scale"): out.append({"questionId": key, "severity": sev}) return out def match_use_cases(self, cat): cases = [u for u in self.d.use_cases if u.get("category") == cat] hits = [q for q in self.p2_summary(cat) if q["severity"] and q["severity"] >= 3] scored = [] for uc in cases: score = 0 trig = uc.get("p2TriggerConditions") sc = len(trig["rangeSignals"]) if trig and truthy(trig.get("rangeSignals", UNDEF)) else 0 for q in hits: score += 2 if q["severity"] >= 4 else 1 if sc > 0 and len(hits) > 0: score += min(sc, len(hits)) scored.append({"id": uc["id"], "score": score}) scored.sort(key=lambda x: -x["score"]) return scored def detect_sector(self): # The Conversation asks the sector outright and keeps it in irf_act0. It wins over the mission text. act0 = self.s.json_key("irf_act0", None) if isinstance(act0, dict) and isinstance(act0.get("sector"), str) and act0["sector"] in SECTOR_KEYWORDS: return act0["sector"] p0 = self.s.blob(0) if p0 is None or p0 == "__bad__" or not isinstance(p0, dict) or not truthy(p0.get("answers", UNDEF)): return None a = p0["answers"] def part(k): v = get(a, k) return js_str(v) if truthy(v) else "" text = " ".join([part("NGO-P0-S1-Q5"), part("NGO-P0-S1-Q5_notes"), part("NGO-P0-S1-Q4"), part("NGO-P0-S1-Q4_notes")]).lower() if not text.strip(): return None best, best_hits = None, 0 for sector, kws in SECTOR_KEYWORDS.items(): hits = sum(1 for kw in kws if kw in text) if hits > best_hits: best_hits, best = hits, sector return best def programme_floors(self): p0 = self.s.blob(0) answers = p0.get("answers") if isinstance(p0, dict) and isinstance(p0.get("answers"), dict) else {} for pr in self.d.programmes: meta = pr["meta"] field = meta.get("activationField") vals = meta.get("activationValues") if not field or not isinstance(vals, list): continue a = get(answers, field) if a is UNDEF or a is None: continue arr = a if isinstance(a, list) else [a] if any(x in vals for x in arr): floors = {} for uid, b in pr["boosts"].items(): if isinstance(b, dict) and is_num(b.get("floor")): floors[uid] = b["floor"] return pr["name"], floors return None, {} # -- computeGlobalUseCaseScores ------------------------------------------ def compute_global_scores(self, selected_areas): smap = self.d.scoring_map scored = [] for cat in selected_areas: cases = [u for u in self.d.use_cases if u.get("category") == cat] quant = self.p2_summary(cat) sev_sum, sev_count = 0, 0 for q in quant: if truthy(q["severity"] if q["severity"] is not None else UNDEF): sev_sum += q["severity"] sev_count += 1 max_cat = max([q["severity"] for q in quant if truthy(q["severity"] if q["severity"] is not None else UNDEF)] or [0]) avg_cat = sev_sum / sev_count if sev_count > 0 else 0 for uc in cases: uc_map = smap.get(uc["id"]) if isinstance(uc_map, dict) and truthy(uc_map.get("signals", UNDEF)) and len(uc_map["signals"]) > 0: need = self.score_by_role(uc_map["signals"], "need") scale = self.score_by_role(uc_map["signals"], "scale") ready = self.score_by_role(uc_map["signals"], "readiness") boost = self.score_by_role(uc_map["signals"], "fit_boost") need_s, scale_s = need["score"], scale["score"] raw_ready = ready["score"] ready_s = (5 - raw_ready) if raw_ready > 0 else 2.5 ready_source = "signals" if ready["signalCount"] > 0 else "default" base_fit = uc_map.get("blockchainFit") if truthy(uc_map.get("blockchainFit", UNDEF)) else 2.5 fit_boost = boost["score"] fit = base_fit + (5 - base_fit) * (fit_boost / 5) * 0.4 if fit_boost > 0 else base_fit fit = min(5, fit) ia_tagged = uc_map.get("interagency") is True if ia_tagged: b = self.interagency_boost() if b > 0: fit = min(5, fit + b) if ready_s < 2.0 and need_s > 3.0 and scale_s > 2.5: ready_s = max(ready_s, 2.0 + (need_s - 3.0) * 0.3 + (scale_s - 2.5) * 0.2) ready_s = min(ready_s, 3.5) if fit >= 4.0 and need_s >= 3.5 and ready_s < 2.5: ready_s = max(ready_s, 2.5) blended = need_s * 0.30 + fit * 0.25 + scale_s * 0.25 + ready_s * 0.20 total = need["signalCount"] + scale["signalCount"] + ready["signalCount"] + boost["signalCount"] if total == 0: blended = base_fit * 0.3 final = max(0.1, min(5.0, round1(blended))) allsig = need["topSignals"] + scale["topSignals"] + ready["topSignals"] + boost["topSignals"] allsig = sorted(allsig, key=lambda s: -(s["severity"] * s["weight"])) evidence = [] for sg in allsig[:5]: ans = get(self.p2, sg["q"], UNDEF) evidence.append({"questionId": sg["q"], "role": sg["role"], "severity": sg["severity"], "weight": sg["weight"], "answer": ans if truthy(ans) else ""}) pm = 0 if total == 0 else round1(need_s * 0.6 + scale_s * 0.4) pm_pre = None ia_mag = 0 if ia_tagged and total > 0: b = self.interagency_boost() if b > 0: pm_pre = pm pm = min(5, round1(pm + b)) ia_mag = b fit_gate = uc_map.get("fitGate") if truthy(uc_map.get("fitGate", UNDEF)) else 2.5 rd_gate = uc_map.get("readinessGate") if truthy(uc_map.get("readinessGate", UNDEF)) else 2.5 if total == 0: fg = rg = "no-data" else: fg = "pass" if fit >= fit_gate else ("borderline" if fit >= fit_gate - 1.0 else "fail") rg = "pass" if ready_s >= rd_gate else ("borderline" if ready_s >= rd_gate - 1.0 else "fail") scored.append({ "ucId": uc["id"], "name": uc.get("name"), "categoryCode": cat, "finalScore": final, "problemMagnitude": pm, "needScore": round1(need_s), "scaleScore": round1(scale_s), "readinessScore": round1(ready_s), "readinessSource": ready_source, "blockchainFit": round1(fit), "fitGate": {"threshold": fit_gate, "status": fg}, "readinessGate": {"threshold": rd_gate, "status": rg}, "evidence": evidence, "scoringMethod": "per-uc", "signalsFired": total, "signalsAvailable": len(uc_map["signals"]), "interagencyTagged": ia_tagged, "interagencyBoostApplied": ia_mag, }) else: match = self.match_use_cases(cat) item = next((m for m in match if m["id"] == uc["id"]), None) ms = item["score"] if item else 0 cmax = 1 for m in match: if m["score"] > cmax: cmax = m["score"] prop = ms / cmax if cmax > 0 else 0 on5 = 1 + prop * 4 blend = max_cat * 0.65 + avg_cat * 0.15 + on5 * 0.2 if max_cat == 0: blend = on5 * 0.3 fb = max(0.1, min(5.0, round1(blend))) scored.append({ "ucId": uc["id"], "name": uc.get("name"), "categoryCode": cat, "finalScore": fb, "problemMagnitude": round1(avg_cat), "needScore": None, "scaleScore": None, "readinessScore": None, "readinessSource": "none", "blockchainFit": None, "fitGate": {"threshold": 2.5, "status": "no-data"}, "readinessGate": {"threshold": 2.5, "status": "no-data"}, "evidence": [], "scoringMethod": "category-fallback", "signalsFired": 0, "signalsAvailable": 0, "interagencyTagged": False, "interagencyBoostApplied": 0, }) sector = self.detect_sector() sector_floors = SECTOR_UC_FLOORS.get(sector, {}) if sector else {} prog_name, prog_floors = self.programme_floors() for it in scored: sf = sector_floors.get(it["ucId"], 0) pf = prog_floors.get(it["ucId"], 0) floor = max(sf, pf) if floor > 0: if it["finalScore"] < floor: it["finalScore"] = floor if (it["problemMagnitude"] or 0) < floor: it["problemMagnitudePreFloor"] = it["problemMagnitude"] or 0 it["problemMagnitude"] = floor it["floorApplied"] = True if pf >= sf and pf > 0: it["floorReason"] = "programme:" + (prog_name or "programme") it["floorAmount"] = pf elif sf > 0: it["floorReason"] = "sector:" + sector it["floorAmount"] = sf scored.sort(key=lambda x: -(x["problemMagnitude"] or 0)) self.detected_sector = sector self.programme = prog_name return scored # -- tierUseCases --------------------------------------------------------- def tier_use_cases(self, scored): by_need = sorted(scored, key=lambda x: -(x["problemMagnitude"] or 0)) CAP, FLOOR = 4, 2 top10, ids, count = [], set(), {} sel = self.selected if isinstance(self.selected, list) else [] for cat in sel: per = [it for it in by_need if it["categoryCode"] == cat] i = 0 while i < len(per) and count.get(cat, 0) < FLOOR and len(top10) < 10: it = per[i] i += 1 if it["ucId"] in ids: continue top10.append(it) ids.add(it["ucId"]) count[cat] = count.get(cat, 0) + 1 for it in by_need: if len(top10) >= 10: break if it["ucId"] in ids: continue c = it["categoryCode"] count[c] = count.get(c, 0) if count[c] >= CAP: continue top10.append(it) ids.add(it["ucId"]) count[c] += 1 top10 = sorted(top10, key=lambda x: -(x["problemMagnitude"] or 0)) for it in scored: it["isTop10"] = it["ucId"] in ids fg, rg = it["fitGate"]["status"], it["readinessGate"]["status"] if fg == "fail" or rg == "fail": it["gateStatus"] = "fail" elif fg == "borderline" or rg == "borderline": it["gateStatus"] = "borderline" elif fg == "no-data" or rg == "no-data": it["gateStatus"] = "no-data" else: it["gateStatus"] = "pass" tiers = {"top10": top10, "highNeed": [], "mediumNeed": [], "lowNeed": [], "minimal": []} for it in by_need: pm = it["problemMagnitude"] or 0 if pm >= 3.5: tiers["highNeed"].append(it) elif pm >= 2.0: tiers["mediumNeed"].append(it) elif pm >= 1.0: tiers["lowNeed"].append(it) else: tiers["minimal"].append(it) return tiers # -- Phase 2 severity for a category ------------------------------------ def p2_severity(self, cat): """Severity of the Phase 2 answers in one category, from each question's severityMapping. Returns avg and max on the 1 to 5 scale, and how many answers could be rated. """ res = {"avgSeverity": 0, "maxSeverity": 0, "count": 0} total, count, mx = 0, 0, 0 for key, val in self.p2.items(): up = key.upper() if not ("Cat" + cat in key or "CAT" + cat.upper() in key or "CAT-" + cat.upper() in up or "CAT" + cat.upper() in up): continue if isinstance(val, dict) and truthy(val.get("value", UNDEF)): val = val["value"] qinfo = self.d.p2q.get(key) if not qinfo or not isinstance(qinfo["severityMapping"], dict): continue m = qinfo["severityMapping"].get(js_str(val)) if is_num(m) and 1 <= m <= 5: count += 1 total += m if m > mx: mx = m if count > 0: res = {"avgSeverity": total / count, "maxSeverity": mx, "count": count} return res # --------------------------------------------------------------------------- # Phase 4 # --------------------------------------------------------------------------- def get_val(p0, key): v = get(p0, key) if not truthy(v): return "" if isinstance(v, dict) and v.get("value", UNDEF) is not UNDEF: return js_str(v["value"]) return js_str(v) def compute_org_agility(p0): def get_num(key): n = js_parseint(get_val(p0, key)) return 0 if is_nan(n) else n def get_arr(key): v = get(p0, key) if not truthy(v): return [] if isinstance(v, list): return v if isinstance(v, dict) and v.get("value", UNDEF) is not UNDEF: if isinstance(v["value"], list): return v["value"] return split_multi(v["value"]) return split_multi(v) staff_val = get_val(p0, "NGO-P0-S2-Q7") or get_val(p0, "NGO-P0-S1-Q3") staff_count = or0(js_parseint(staff_val)) sl = staff_val.lower() if any(t in sl for t in ("fewer than 20", "under 20", "1-19", "< 20")): staff_count = 15 elif any(t in sl for t in ("20-", "21-", "50-")): staff_count = 50 elif any(t in sl for t in ("100", "500", "1000")): staff_count = 200 bucket = "small" if staff_count < 20 else ("medium" if staff_count <= 100 else "large") staff_score = 3 if staff_count < 20 else (1 if staff_count <= 100 else 0) d1q1 = get_num("NGO-P4-D1-Q1") if d1q1 >= 4: decision = 2 elif d1q1 >= 3: decision = 1 elif bucket == "small": decision = 2 else: decision = 0 d4q1 = get_num("NGO-P4-D4-Q1") tech = 2 if d4q1 >= 3 else (1 if d4q1 >= 2 else 0) partners_val = get_val(p0, "NGO-P0-S5-Q17a") or get_val(p0, "NGO-P0-S5-Q17b") partners_num = or0(js_parseint(partners_val)) partners_score = 1 if partners_num >= 10 else 0 coord_val = get_val(p0, "NGO-P0-S5-Q18") coord_score = 1 if (coord_val and len(coord_val) > 2) else 0 ds_val = get_val(p0, "NGO-P0-S5-Q19") ds_score = 1 if (ds_val and "yes" in ds_val.lower()) else 0 d3q3 = get_num("NGO-P4-D3-Q3") pilot = 1 if d3q3 >= 3 else 0 sys_val = get_val(p0, "NGO-P0-S3-Q10") sys_score = 1 if (sys_val and len(sys_val) > 5) else 0 general = staff_score + decision + tech + partners_score + coord_score + ds_score + pilot + sys_score co_val = get_val(p0, "NGO-P0-S7-Q2") co_score = 0 co_num = or0(js_parseint(co_val)) if co_num >= 3 or (co_val and len(co_val) > 10): co_score = 1 it_val = get_val(p0, "NGO-P0-S2-Q8") cs_val = get_val(p0, "NGO-P0-S2-Q9") has_cs = False if cs_val and "yes" in cs_val.lower(): has_cs = True if it_val and ("compliance" in it_val.lower() or "legal" in it_val.lower()): has_cs = True compliance = general + co_score + (1 if has_cs else 0) countries = get_arr("NGO-P0-S1-Q7") funding = get_arr("NGO-P0-S4-Q15") aml = get_arr("NGO-P0-S6-Q21") voluntary = get_arr("NGO-P0-S6-Q20") risk = get_arr("NGO-P0-S7-Q1") registration = get_val(p0, "NGO-P0-S3-Q11a") sys_count_val = get_val(p0, "NGO-P0-S3-Q11b") connectivity = get_val(p0, "NGO-P0-S3-Q13") governance = get_val(p0, "NGO-P0-S2-Q9") tech_approval = get_val(p0, "NGO-P0-S2-Q9b") political = get_val(p0, "NGO-P0-S6-Q23") ben_scale = get_val(p0, "NGO-P0-S1-Q8") budget = get_val(p0, "NGO-P0-S4-Q14a") budget_trend = get_val(p0, "NGO-P0-S4-Q14b") rl = registration.lower() has_dig = bool(registration) and ("platform" in rl or "database" in rl or "crm" in rl) cl = coord_val.lower() has_coord = bool(coord_val) and ("cluster" in cl or "consortium" in cl or "working group" in cl or "coordination" in cl) shares = bool(ds_val) and "regularly" in ds_val.lower() gl = governance.lower() has_gov = bool(governance) and ("board" in gl or "federated" in gl) has_aml = any(("aml" in js_str(a).lower() or "ofac" in js_str(a).lower() or "sanctions" in js_str(a).lower()) for a in aml) donor = any(any(t in js_str(f).lower() for t in ("institutional", "usaid", "dfid", "eu", "un ")) for f in funding) sc_num = 1 if "2-3" in sys_count_val: sc_num = 3 elif "4-6" in sys_count_val: sc_num = 5 elif "More" in sys_count_val or "6" in sys_count_val: sc_num = 7 conn_level = 3 cn = connectivity.lower() if "reliable high" in cn: conn_level = 5 elif "reliable at hq" in cn: conn_level = 4 elif "intermittent" in cn and "both" in cn: conn_level = 2 elif "significant" in cn or "gaps" in cn: conn_level = 1 elif "offline" in cn or "no regular" in cn: conn_level = 0 pol_risk = 0 pl = political.lower() if "hostile" in pl: pol_risk = 3 elif "challenging" in pl: pol_risk = 2 elif "mixed" in pl: pol_risk = 1 elif "varies" in pl: pol_risk = 1 return { "general": general, "compliance": compliance, "staffCount": staff_count, "hasComplianceStaff": has_cs, "orgSizeBucket": bucket, "ctx": { "operatingContext": get_val(p0, "NGO-P0-S1-Q3"), "beneficiaryScale": ben_scale, "operatingCountryCount": len(countries), "hasDigitalRegistration": has_dig, "systemCount": sc_num, "systemCountRaw": sys_count_val, "connectivityLevel": conn_level, "budgetRange": budget, "budgetTrend": budget_trend, "isDonorFunded": donor, "hasFormalGovernance": has_gov, "techApprovalProcess": tech_approval, "politicalRisk": pol_risk, "hasAML": has_aml, "riskManagement": risk, "hasFormalCoordination": has_coord, "sharesDataSystematically": shares, "voluntaryFrameworks": voluntary, }, } def org_adjustment_caps(bucket): if bucket == "small": return {"f1": -4, "f2": -3, "f3": -2} if bucket == "medium": return {"f1": -3, "f2": -2, "f3": -2} return {"f1": -2, "f2": -2, "f3": -2} def compute_uc_complexity(data, uc_id, agility, p4): neutral = {"f1": 3, "f2": 3, "f3": 3, "total": 9, "problemMaturity": 0, "baseTotal": 9} uc = data.uc_by_id.get(uc_id) if not uc or not truthy(uc.get("complexityFactors", UNDEF)): return dict(neutral) base = uc["complexityFactors"] r, o, c = base.get("relationships"), base.get("operationalChange"), base.get("dataCompliance") if not all(is_finite_num(x) for x in (r, o, c)): return dict(neutral) caps = org_adjustment_caps(agility["orgSizeBucket"]) f1a = max(min(0, -math.floor(agility["general"] * 0.8)), caps["f1"]) f2a = max(min(0, -math.floor(agility["general"] * 0.6)), caps["f2"]) f3a = max(min(0, -math.floor(agility["compliance"] * 0.5)), caps["f3"]) f1_floor = max(1, math.ceil(r / 2)) f2_floor = max(1, math.ceil(o / 2)) f3_floor = max(1, math.ceil(c / 2)) e1 = max(f1_floor, r + f1a) e2 = max(f2_floor, o + f2a) e3 = max(f3_floor, c + f3a) bs = (agility["ctx"].get("beneficiaryScale") or "").lower() if "over 1,000,000" in bs or "1,000,000" in bs: e2 = min(5, e2 + 1) e3 = min(5, e3 + 1) elif "250,000" in bs: e2 = min(5, e2 + 1) pm_val = get(p4, "P4B-UC-" + uc_id + "-problemMaturity") pm_val = pm_val if truthy(pm_val) else "" pmat = 0 if pm_val == "already_doing": pmat = -2 elif pm_val == "tried_before": pmat = -1 elif pm_val == "required_not_started": pmat = -1 base_total = r + o + c total = max(3, e1 + e2 + e3 + pmat) return {"f1": e1, "f2": e2, "f3": e3, "total": total, "problemMaturity": pmat, "baseTotal": base_total} def weighted_thresholds(f1, f2, f3, has_cs, base_total): default = {"importance": {d: 0.2 for d in ACTIVE_DIMS}, "thresholds": {d: 2.5 for d in ACTIVE_DIMS}} if not all(is_finite_num(x) for x in (f1, f2, f3, base_total)): return default matrix = { "factor1": [0.45, 0.05, 0.35, 0.05, 0.10], "factor2": [0.10, 0.15, 0.30, 0.35, 0.10], "factor3": [0.05, 0.05, 0.05, 0.10, 0.75] if has_cs else [0.05, 0.20, 0.05, 0.10, 0.60], } total = f1 + f2 + f3 if total == 0: total = 1 w1, w2, w3 = f1 / total, f2 / total, f3 / total importance, thresholds = {}, {} for i, d in enumerate(ACTIVE_DIMS): importance[d] = w1 * matrix["factor1"][i] + w2 * matrix["factor2"][i] + w3 * matrix["factor3"][i] cbase = 2.5 + (base_total - 9) / 6 spread = 1.5 for d in ACTIVE_DIMS: t = cbase + (importance[d] - 0.2) * spread thresholds[d] = max(1.0, min(4.5, t)) if any(not math.isfinite(thresholds[d]) or not math.isfinite(importance[d]) for d in ACTIVE_DIMS): return default return {"importance": importance, "thresholds": thresholds} def determine_outcome(dim_scores, cat_scores, p2sev, weighted, uc_name_hint=None): """determinePhase4Outcome, returning the outcome and the rule that fired.""" p2sev = p2sev or {"avgSeverity": 0, "maxSeverity": 0, "count": 0} mx = p2sev["maxSeverity"] sev_text = ("The deep-dive answers for this area did not show a severe problem (highest rating %s of 5)." % fmt_num(mx) if p2sev["count"] > 0 else "No deep-dive answer for this area could be rated for severity.") if not cat_scores: return "NOT_ASSESSED", { "id": "no-part-b", "sentence": "The questions about this problem were not answered, so no decision was reached.", "inputs": {}} complexity = cat_scores[0] if len(cat_scores) >= 1 else 3 frequency = cat_scores[1] if len(cat_scores) >= 2 else 3 impact = cat_scores[2] if len(cat_scores) >= 3 else 3 if len(cat_scores) >= 2: if complexity <= 1 and frequency <= 1 and impact <= 1: return "EXPLORE_ALTERNATIVES", { "id": "low-parties-frequency-impact", "sentence": "Few parties are involved, the problem is rare and the impact is small, so a shared ledger adds little.", "inputs": {"partiesAnswer": complexity, "frequencyAnswer": frequency, "impactAnswer": impact}} if complexity <= 1 and mx < 3: return "EXPLORE_ALTERNATIVES", { "id": "single-party", "sentence": "Only your organisation is involved, so a shared ledger adds little. " + sev_text, "inputs": {"partiesAnswer": complexity, "p2MaxSeverity": mx}} weighted_gap, max_gap, max_dim, answered = 0, 0, None, 0 for d in ACTIVE_DIMS: sc = dim_scores[d] if sc <= 0: continue answered += 1 gap = weighted["thresholds"][d] - sc if gap > 0: weighted_gap += gap * (weighted["importance"].get(d) or 0.2) if gap > max_gap: max_gap = gap max_dim = d wg_text = to_fixed(weighted_gap, 2) base_in = {"weightedGap": round2(weighted_gap), "largestGap": round2(max_gap), "largestGapDimension": max_dim, "answeredDimensions": answered} if answered == 0: return "PROCEED_WITH_CONDITIONS", { "id": "no-part-a", "sentence": "The questions about your organisation were not answered, so readiness could not be judged.", "inputs": base_in} if max_gap > 1.5: if mx >= 4.5: return "PROCEED_WITH_CONDITIONS", { "id": "large-gap-severity-override", "sentence": "%s is %s points below what this use case needs, which would normally mean waiting. The problem is severe (rated %s of 5), so the answer is yes, with conditions." % ( DIMENSION_NAMES[max_dim], to_fixed(max_gap, 1), fmt_num(mx)), "inputs": dict(base_in, p2MaxSeverity=mx)} return "NOT_NOW", { "id": "large-single-gap", "sentence": "%s is %s points below what this use case needs. One gap this large means your organisation should build that first." % ( DIMENSION_NAMES[max_dim], to_fixed(max_gap, 1)), "inputs": base_in} if weighted_gap <= 0.2: return "PROCEED", { "id": "gaps-cleared", "sentence": "Your readiness scores meet or come close to the bar for this use case (combined gap %s, limit 0.2)." % wg_text, "inputs": base_in} if weighted_gap <= 0.5: return "PROCEED_WITH_CONDITIONS", { "id": "light-gaps", "sentence": "Small readiness gaps remain (combined gap %s). They can be closed alongside a pilot." % wg_text, "inputs": base_in} if weighted_gap <= 1.5: if mx >= 4 and weighted_gap <= 0.8: return "PROCEED", { "id": "modest-gaps-severe-problem", "sentence": "Readiness gaps are modest (combined gap %s) and the problem is severe (rated %s of 5), so the answer is yes." % (wg_text, fmt_num(mx)), "inputs": dict(base_in, p2MaxSeverity=mx)} return "PROCEED_WITH_CONDITIONS", { "id": "heavier-gaps", "sentence": "Several readiness gaps remain (combined gap %s). Close them before or during a pilot." % wg_text, "inputs": base_in} if mx >= 4.5: return "PROCEED_WITH_CONDITIONS", { "id": "total-gap-severity-override", "sentence": "Combined readiness gaps are large (%s), which would normally mean waiting. The problem is severe (rated %s of 5), so the answer is yes, with conditions." % (wg_text, fmt_num(mx)), "inputs": dict(base_in, p2MaxSeverity=mx)} return "NOT_NOW", { "id": "total-gap-too-large", "sentence": "Combined readiness gaps are too large to start now (combined gap %s, limit 1.5)." % wg_text, "inputs": base_in} # --------------------------------------------------------------------------- # Detailed evaluation (optional) # --------------------------------------------------------------------------- def resolve_logic_key(option_text, options, logic): if js_str(option_text) in logic: return js_str(option_text) if not isinstance(options, list) or option_text not in options: return None idx = options.index(option_text) keys = js_keys(logic) return keys[idx] if idx < len(keys) else None def score_fit(ud, answers): if not ud: return {"overallFit": "insufficient_data", "honestNo": False, "flaggedQuestions": []} questions = ud.get("blockchainFitQuestions") or [] required = ud.get("bCriteria") or [] groups, alt, flagged, answered = {}, None, [], 0 for q in questions: a = answers.get(q["id"]) if not isinstance(a, dict) or not truthy(a.get("value", UNDEF)): continue answered += 1 if a.get("flagged"): flagged.append({"questionId": q["id"]}) continue logic = q.get("fitLogic") label = None if isinstance(logic, dict): lk = resolve_logic_key(a["value"], q.get("options"), logic) label = logic[lk] if lk is not None else None if not label or label in ("neutral", "flag_for_review"): continue num = FIT_SCORES.get(label) if num is None: continue maps_to = q.get("mapsTo") if maps_to == "alternatives": alt = label continue groups.setdefault(js_str(maps_to), []).append(num) scores = {} for k, g in groups.items(): avg = sum(g) / len(g) lab = "strong" if avg >= 3.5 else ("moderate" if avg >= 2.5 else ("weak" if avg >= 1.5 else "negative")) scores[k] = {"score": round2(avg), "label": lab, "questionCount": len(g)} honest, reasons = False, [] if alt == "negative": honest = True reasons.append("Your answers suggest a single trusted system already exists that could address this problem without a shared ledger.") for b in required: s = scores.get(b) if s and s["label"] == "negative": honest = True reasons.append("Your situation does not appear to need %s, which this use case depends on." % B_NAMES.get(b, b)) tws, tw = 0, 0 for b in required: if b in scores: tws += scores[b]["score"] * 2 tw += 2 for b in scores: if b not in required: tws += scores[b]["score"] tw += 1 if tw == 0 and answered > 0: return {"overallFit": "insufficient_data", "overallScore": 0, "bCriteriaScores": scores, "requiredBCriteria": required, "alternativesResult": alt, "honestNo": False, "honestNoReasons": [], "flaggedQuestions": flagged, "answeredCount": answered, "totalCount": len(questions), "allFlagged": True, "confidence": "none"} overall = tws / tw if tw > 0 else 0 if answered == 0: fit = "insufficient_data" elif overall >= 3.5: fit = "strong" elif overall >= 2.5: fit = "moderate" elif overall >= 1.5: fit = "weak" else: fit = "negative" conf = "high" if tw >= 3 else ("medium" if tw == 2 else ("low" if tw == 1 else "none")) return {"overallFit": fit, "overallScore": round2(overall), "bCriteriaScores": scores, "requiredBCriteria": required, "alternativesResult": alt, "honestNo": honest, "honestNoReasons": reasons, "flaggedQuestions": flagged, "answeredCount": answered, "totalCount": len(questions), "allFlagged": False, "confidence": conf} def score_readiness(ud, answers): if not ud: return {"overallReadiness": "insufficient_data", "blockers": [], "conditions": []} questions = ud.get("readinessQuestions") or [] groups, flagged, answered = {}, [], 0 for q in questions: a = answers.get(q["id"]) if not isinstance(a, dict) or not truthy(a.get("value", UNDEF)): continue answered += 1 if a.get("flagged"): flagged.append({"questionId": q["id"]}) continue logic = q.get("readinessLogic") label = None if isinstance(logic, dict): lk = resolve_logic_key(a["value"], q.get("options"), logic) label = logic[lk] if lk is not None else None if not label or label in ("neutral", "flag_for_review"): continue num = READINESS_SCORES.get(label) if num is None: continue dim = q.get("mapsTo") if truthy(q.get("mapsTo", UNDEF)) else "general" groups.setdefault(dim, []).append({"questionId": q["id"], "label": label, "score": num}) dims, blockers, conds = {}, [], [] for dk, g in groups.items(): s = 0 for item in g: s += item["score"] if item["label"] == "not_ready": blockers.append({"dimension": dk, "dimensionLabel": EVAL_DIM_LABELS.get(dk, dk), "score": item["score"], "questionId": item["questionId"]}) avg = s / len(g) lab = "ready" if avg >= 3.5 else ("moderate" if avg >= 2.5 else ("low" if avg >= 1.5 else "not_ready")) dims[dk] = {"score": round2(avg), "label": lab, "questionCount": len(g)} if lab == "low": conds.append({"dimension": dk, "dimensionLabel": EVAL_DIM_LABELS.get(dk, dk), "score": round2(avg), "recommendation": "This area needs strengthening. Make a plan to close the gap before full implementation."}) elif lab == "not_ready": conds.append({"dimension": dk, "dimensionLabel": EVAL_DIM_LABELS.get(dk, dk), "score": round2(avg), "recommendation": "This area needs significant work before implementation."}) allscores = [v["score"] for v in dims.values()] overall = sum(allscores) / len(allscores) if allscores else 0 if answered == 0: rd = "insufficient_data" elif blockers: rd = "not_ready" elif overall >= 3.5: rd = "ready" elif overall >= 2.5: rd = "moderate" elif overall >= 1.5: rd = "low" else: rd = "not_ready" n = len(allscores) return {"overallReadiness": rd, "overallScore": round2(overall), "dimensionScores": dims, "blockers": blockers, "conditions": conds, "flaggedQuestions": flagged, "answeredCount": answered, "totalCount": len(questions), "allFlagged": answered > 0 and n == 0, "confidence": "high" if n >= 3 else ("medium" if n == 2 else ("low" if n == 1 else "none"))} def evaluate_uc(data, uc_id, answers): ud = data.eval_questions().get(uc_id) ud = ud if isinstance(ud, dict) else None fit = score_fit(ud, answers) rd = score_readiness(ud, answers) if fit["overallFit"] == "insufficient_data": outcome, reason = "INSUFFICIENT_DATA", "Not enough fit questions were answered to reach a verdict." elif fit["honestNo"]: outcome, reason = "EXPLORE_ALTERNATIVES", " ".join(fit["honestNoReasons"]) elif fit["overallFit"] == "negative": outcome, reason = "EXPLORE_ALTERNATIVES", "Overall fit for a shared ledger is low, so a conventional solution may suit better." elif fit["overallFit"] == "weak": outcome, reason = "PROCEED_WITH_CONDITIONS", "Fit for a shared ledger is marginal. Weigh it carefully against simpler alternatives." elif rd["overallReadiness"] == "insufficient_data": outcome, reason = "FIT_PASSED_AWAITING_READINESS", "Fit is %s. The readiness questions have not been answered yet." % fit["overallFit"] elif rd["blockers"]: outcome, reason = "NOT_NOW", "A shared ledger fits this problem, but critical readiness gaps need closing first." elif rd["overallReadiness"] in ("not_ready", "low"): outcome, reason = "NOT_NOW", "A shared ledger fits, but organisational readiness is too low to start now." elif rd["conditions"]: outcome, reason = "PROCEED_WITH_CONDITIONS", "A shared ledger fits and basic readiness is there, but some areas need strengthening." else: outcome, reason = "PROCEED", "A shared ledger is a good fit for this problem and the organisation is ready to move forward." return {"ucId": uc_id, "outcome": outcome, "outcomeReason": reason, "fit": fit, "readiness": rd} def map_readiness_to_p4(rd): dims = {d: [] for d in ACTIVE_DIMS} for dk, v in (rd.get("dimensionScores") or {}).items(): dims[EVAL_DIM_MAPPING.get(dk) or "D3"].append(v["score"]) out = {} for d in dims: if dims[d]: s = sum(dims[d]) / len(dims[d]) out[d] = js_round((s / 4) * 5 * 100) / 100 return out def eval_result_for_p4(ev): return { "outcome": ev["outcome"], "outcomeReason": ev["outcomeReason"], "fit": {"overallFit": ev["fit"].get("overallFit"), "overallScore": ev["fit"].get("overallScore"), "honestNo": ev["fit"].get("honestNo")}, "readiness": {"overallReadiness": ev["readiness"].get("overallReadiness"), "overallScore": ev["readiness"].get("overallScore"), "blockers": ev["readiness"].get("blockers") or [], "conditions": ev["readiness"].get("conditions") or []}, "p4DimensionScores": map_readiness_to_p4(ev["readiness"]), } # --------------------------------------------------------------------------- # Starred use cases and Phase 4 # --------------------------------------------------------------------------- def flatten_selected(engine): s, d = engine.s, engine.d order = s.json_key("irf_p3_starred_priority_order", []) starred = s.json_key("irf_p3_starred_usecases", []) order = order if isinstance(order, list) else [] starred = starred if isinstance(starred, list) else [] ids = [] if starred: for u in order: if u in starred and u not in ids: ids.append(u) for u in starred: if u not in ids: ids.append(u) source = "starred" else: prio = s.json_key("irf_p3_priority_order", []) cats = prio if isinstance(prio, list) and prio else (engine.selected if isinstance(engine.selected, list) else []) for c in cats: stored = s.json_key("irf_p3_selected_usecases_" + c, []) for u in (stored if isinstance(stored, list) else []): ids.append(u) source = "per-category" if ids else "none" out = [] for u in ids: ud = d.uc_by_id.get(u) out.append({"id": u, "category": (ud.get("category") or "") if ud else "", "name": ud.get("name") if ud else u, "bCriteria": (ud.get("bCriteria") or []) if ud else []}) return out, source def compute_phase4(engine, flat, eval_results): d = engine.d p0b = engine.s.blob(0) p0 = {} if isinstance(p0b, dict) and truthy(p0b.get("answers", UNDEF)) and isinstance(p0b["answers"], dict): p0 = dict(p0b["answers"]) p4 = engine.p4 for dim in ACTIVE_DIMS: for q in range(1, DIM_QUESTION_COUNTS[dim] + 1): key = "NGO-P4-%s-Q%d" % (dim, q) v = get(p4, key) if v is not UNDEF and v != "": p0[key] = v agility = compute_org_agility(p0) ctx = agility["ctx"] outcomes = {} complexities = {} for uc in flat: dim_scores = {} for dim in ACTIVE_DIMS: sc = [] for q in range(1, DIM_QUESTION_COUNTS[dim] + 1): v = answer_value(get(p4, "NGO-P4-%s-Q%d" % (dim, q))) if v is not UNDEF and v is not None and v != "": n = js_parseint(v) if not is_nan(n): sc.append(n) dim_scores[dim] = sum(sc) / len(sc) if sc else 0 raw_dims = dict(dim_scores) adjustments = [] def add(dim, amt, why): before = dim_scores[dim] dim_scores[dim] = min(5, dim_scores[dim] + amt) adjustments.append({"dimension": dim, "amount": amt, "reason": why}) if ctx["hasDigitalRegistration"] and dim_scores["D2"] > 0: add("D2", 0.3, "digital registration system") if ctx["hasFormalCoordination"] and dim_scores["D3"] > 0: add("D3", 0.3, "formal coordination mechanisms") if ctx["hasFormalGovernance"] and dim_scores["D3"] > 0: add("D3", 0.2, "board or federated governance") if ctx["sharesDataSystematically"] and dim_scores["D4"] > 0: add("D4", 0.3, "systematic data sharing") if len(ctx["riskManagement"]) >= 3 and dim_scores["D3"] > 0: add("D3", 0.2, "mature risk management") oc = js_str(ctx["operatingContext"]) is_fcv = any(t in oc for t in ("Active conflict", "complex emergency", "Post-conflict", "transitional", "Mixed", "hostile-state", "Restrictive", "Economic crisis", "hyperinflation", "Protracted displacement")) if is_fcv: if not ctx["hasDigitalRegistration"] and dim_scores["D2"] > 0: add("D2", 0.3, "conflict or fragile context") if not ctx["hasFormalCoordination"] and dim_scores["D3"] > 0: add("D3", 0.3, "conflict or fragile context") if not ctx["hasFormalGovernance"] and dim_scores["D3"] > 0: add("D3", 0.2, "conflict or fragile context") if not ctx["sharesDataSystematically"] and dim_scores["D4"] > 0: add("D4", 0.3, "conflict or fragile context") if not (len(ctx["riskManagement"]) >= 3) and dim_scores["D3"] > 0: add("D3", 0.2, "conflict or fragile context") answered_vals = [dim_scores[x] for x in ACTIVE_DIMS if dim_scores[x] > 0] baseline = sum(answered_vals) / len(answered_vals) if answered_vals else 0 cat = uc["category"] cat_scores = [] for q in range(1, 6): v = get(p4, "NGO-P4-%s-Q%d__%s" % (cat, q, uc["id"])) if v is not UNDEF and v != "": n = js_parseint(v) if not is_nan(n): cat_scores.append(n) cat_score = sum(cat_scores) / len(cat_scores) if cat_scores else 0 bespoke = eval_results.get(uc["id"]) if bespoke and isinstance(bespoke.get("p4DimensionScores"), dict): for bd, bs in bespoke["p4DimensionScores"].items(): cur = dim_scores.get(bd) if is_num(cur) and not is_nan(cur) and cur >= 0: blended = dim_scores[bd] if (not is_num(bs) or is_nan(bs)) else (dim_scores[bd] * 0.4 + bs * 0.6) dim_scores[bd] = max(0, min(5, blended)) elif is_num(bs) and not is_nan(bs): dim_scores[bd] = max(0, min(5, bs)) p2s = engine.p2_severity(cat) blended_cat = cat_score if cat_scores and p2s["count"] > 0: blended_cat = max(cat_score, p2s["avgSeverity"] * 0.6) cx = compute_uc_complexity(d, uc["id"], agility, p4) weighted = weighted_thresholds(cx["f1"], cx["f2"], cx["f3"], agility["hasComplianceStaff"], cx["baseTotal"]) complexities[uc["id"]] = {"complexity": cx, "category": cat, "name": uc["name"]} conditions = [] for dim in ACTIVE_DIMS: thr = weighted["thresholds"][dim] sc = dim_scores[dim] if sc > 0 and sc < thr - 1.0: conditions.append({"dim": dim, "level": "significant", "text": "%s scored %s of 5, which is well below the %s this use case needs. Build this capability before implementation." % (DIMENSION_NAMES[dim], to_fixed(sc, 1), to_fixed(thr, 1))}) elif sc > 0 and sc < thr: conditions.append({"dim": dim, "level": "moderate", "text": "%s scored %s of 5 against the %s this use case needs. Address it before or during the pilot." % (DIMENSION_NAMES[dim], to_fixed(sc, 1), to_fixed(thr, 1))}) elif sc > 0 and sc < thr + 0.5: conditions.append({"dim": dim, "level": "minor", "text": "%s scored %s of 5, only just above the %s needed. Keep strengthening it during implementation." % (DIMENSION_NAMES[dim], to_fixed(sc, 1), to_fixed(thr, 1))}) def p0c(level, text): conditions.append({"dim": "P0", "level": level, "text": text}) if ctx["operatingCountryCount"] > 3: p0c("minor", "You operate in %d countries. Rules and rollout timing will differ between them." % ctx["operatingCountryCount"]) if 0 < ctx["connectivityLevel"] <= 1: p0c("moderate", "Connectivity gaps were reported. Any solution needs to work with little or no connection.") elif ctx["connectivityLevel"] == 0: p0c("moderate", "You mostly work offline. Any solution must work without regular internet access.") if ctx["politicalRisk"] >= 2: p0c("minor", "The political environment is challenging. Plan a phased rollout.") if ctx["isDonorFunded"]: p0c("minor", "Funding is mainly institutional donors. Allow for their procurement rules in the timeline.") if ctx["budgetRange"] and "under" in ctx["budgetRange"].lower(): p0c("minor", "A smaller budget may call for shared implementation costs with a partner.") if ctx["systemCount"] >= 5: p0c("minor", "Several disconnected systems were reported (%s). Plan a data consolidation step." % ctx["systemCountRaw"]) oc2 = js_str(ctx["operatingContext"]) if "hostile-state" in oc2 or "Restrictive" in oc2: p0c("significant", "Hostile-state setting. Security comes first, so avoid designs that expose beneficiary identities.") if "Economic crisis" in oc2 or "hyperinflation" in oc2: p0c("significant", "Economic collapse or hyperinflation. Local currency instability is a first-order design concern.") if "Active conflict" in oc2 or "complex emergency" in oc2: p0c("moderate", "Active conflict. The solution must keep working through outages and staff turnover.") if "Protracted displacement" in oc2: p0c("moderate", "Protracted displacement. Plan for identity portability and credential recovery.") ta = ctx["techApprovalProcess"] if ta and ("board" in ta.lower() or "senior management" in ta.lower()): p0c("minor", "New technology needs senior or board approval. Allow for it in the pilot timeline.") if ctx["hasAML"]: if weighted["importance"] and truthy(weighted["importance"].get("D5", UNDEF)): weighted["importance"]["D5"] = min(1, weighted["importance"]["D5"] + 0.1) outcome, rule = determine_outcome(dim_scores, cat_scores, p2s, weighted) rule["inputs"]["p2Count"] = p2s["count"] if bespoke: bo = bespoke.get("outcome") if bo == "EXPLORE_ALTERNATIVES": outcome = "EXPLORE_ALTERNATIVES" rule = {"id": "evaluation-explore-alternatives", "sentence": "The detailed evaluation found a simpler system would do the job. " + (bespoke.get("outcomeReason") or ""), "inputs": {"evaluationOutcome": bo}} elif bo == "NOT_NOW" and outcome != "EXPLORE_ALTERNATIVES": outcome = "NOT_NOW" rule = {"id": "evaluation-not-now", "sentence": "The detailed evaluation found readiness gaps that need closing first. " + (bespoke.get("outcomeReason") or ""), "inputs": {"evaluationOutcome": bo}} if bo == "PROCEED" and outcome == "NOT_NOW": outcome = "PROCEED_WITH_CONDITIONS" rule = {"id": "evaluation-softens-not-now", "sentence": "The general readiness check said to wait, but the detailed questions were positive, so the answer is yes, with conditions.", "inputs": {"evaluationOutcome": bo}} if outcome == "NOT_ASSESSED" and truthy(bo if bo is not None else UNDEF): outcome = bo rule = {"id": "evaluation-only", "sentence": "The use-case questions were not answered, so this rests on the detailed questions alone.", "inputs": {"evaluationOutcome": bo}} for b in (bespoke.get("readiness") or {}).get("blockers") or []: conditions.append({"dim": "EVAL", "level": "significant", "text": "%s was a blocker in the detailed evaluation." % b["dimensionLabel"]}) for c in (bespoke.get("readiness") or {}).get("conditions") or []: conditions.append({"dim": "EVAL", "level": "moderate", "text": "%s: %s" % (c["dimensionLabel"], c["recommendation"])}) sig_count = len([c for c in conditions if c["level"] in ("significant", "moderate")]) outcomes[uc["id"]] = { "outcome": outcome, "rule": rule, "dimensionScores": dict(dim_scores), "rawDimensionScores": raw_dims, "adjustments": adjustments, "baselineScore": float(to_fixed(baseline, 2)), "categoryScore": float(to_fixed(blended_cat, 2)), "partB": {"answers": cat_scores, "parties": cat_scores[0] if len(cat_scores) >= 1 else None, "frequency": cat_scores[1] if len(cat_scores) >= 2 else None, "impact": cat_scores[2] if len(cat_scores) >= 3 else None}, "p2Severity": p2s, "thresholds": weighted["thresholds"], "importance": weighted["importance"], "complexity": cx, "conditions": [{"dimension": c["dim"], "level": c["level"], "text": c["text"]} for c in conditions], "conditionCount": sig_count, "evaluation": ({"outcome": bespoke.get("outcome")} if bespoke else None), "steppingStone": None, } for uid, od in outcomes.items(): comp = complexities.get(uid) if od["outcome"] == "PROCEED_WITH_CONDITIONS" and comp and comp["complexity"]["total"] > 8: cur = comp["complexity"]["total"] best, best_total = None, 999 for oid, other in complexities.items(): if oid == uid: continue if other["category"] == comp["category"] and other["complexity"]["total"] < cur - 3: oo = outcomes[oid]["outcome"] if oid in outcomes else "" if oo != "NOT_NOW" and oo != "NOT_ASSESSED": if other["complexity"]["total"] < best_total: best_total = other["complexity"]["total"] best = (oid, other["name"]) if best: od["steppingStone"] = {"id": best[0], "name": best[1]} return outcomes, agility # --------------------------------------------------------------------------- # Top level # --------------------------------------------------------------------------- def tier_name(it): pm = it["problemMagnitude"] or 0 if it["signalsFired"] == 0 and it["scoringMethod"] == "per-uc": return "not-enough-information" if pm >= 3.5: return "high" if pm >= 2.0: return "medium" if pm >= 1.0: return "low" return "minimal" def score(data, raw, internals=None): notes = [] store = Store(raw, notes) eng = Engine(data, store) selected = eng.selected if isinstance(eng.selected, list) else [] if not selected: notes.append("No selected categories (irf_ngo_selected_problem_areas) in the answers file, so no use case was ranked.") scored = eng.compute_global_scores(selected) tiers = eng.tier_use_cases(scored) has_data = [it for it in scored if not (it["signalsFired"] == 0)] tiers_known = eng.tier_use_cases([dict(it) for it in has_data]) if len(has_data) != len(scored) else tiers top_ids = set(x["ucId"] for x in tiers_known["top10"]) # Evaluation results: stored results win, answers fill the rest. eval_results = store.json_key("ngo_eval_results", {}) eval_results = eval_results if isinstance(eval_results, dict) else {} eval_answers = store.json_key("ngo_evaluation_answers", {}) if isinstance(eval_answers, dict): for uid, ans in eval_answers.items(): if uid in eval_results or not isinstance(ans, dict): continue ev = evaluate_uc(data, uid, ans) eval_results[uid] = eval_result_for_p4(ev) flat, flat_source = flatten_selected(eng) if not flat: part_b = [] for k in eng.p4: m = re.match(r"^NGO-P4-[A-Z]-Q\d+__(UC-[A-Z]-\d+)$", k) if m and m.group(1) not in part_b: part_b.append(m.group(1)) if part_b: for u in part_b: ud = data.uc_by_id.get(u) flat.append({"id": u, "category": (ud.get("category") or "") if ud else "", "name": ud.get("name") if ud else u, "bCriteria": []}) notes.append("No starred use cases in the answers file. Phase 4 was run for the use cases that have answers about the problem.") flat_source = "part-b-answers" outcomes, agility = compute_phase4(eng, flat, eval_results) if flat else ({}, compute_org_agility({})) if internals is not None: internals.update({"scored": scored, "tiers": tiers, "flat": flat, "outcomes": outcomes, "evalResults": eval_results}) rank = 0 use_cases, not_enough = [], [] by_id = {} for it in scored: known = it["signalsFired"] > 0 if known: rank += 1 entry = { "id": it["ucId"], "name": it["name"], "category": it["categoryCode"], "problemMagnitude": it["problemMagnitude"], "need": it["needScore"], "scale": it["scaleScore"], "blockchainFit": it["blockchainFit"], "readiness": it["readinessScore"], "readinessSource": it["readinessSource"], "gates": {"fit": it["fitGate"], "readiness": it["readinessGate"], "overall": it["gateStatus"]}, "tier": tier_name(it), "rank": rank if known else None, "inTop10": bool(known and it["ucId"] in top_ids), "signalsFired": it["signalsFired"], "signalsAvailable": it["signalsAvailable"], "scoringMethod": it["scoringMethod"], "evidence": it["evidence"], } for k in ("floorApplied", "floorReason", "floorAmount", "problemMagnitudePreFloor"): if k in it: entry[k] = it[k] if it["interagencyTagged"]: entry["interagencyBoostApplied"] = it["interagencyBoostApplied"] entry["phase4"] = None use_cases.append(entry) by_id[it["ucId"]] = entry if not known: ne = {"id": it["ucId"], "name": it["name"], "category": it["categoryCode"], "stage": "ranking", "reason": "None of the questions that feed this use case's score were answered, so it was not ranked."} if it.get("floorApplied"): ne["note"] = "Its score of %s comes only from a sector or programme minimum, not from your answers." % fmt_num(it["problemMagnitude"]) not_enough.append(ne) for uc in flat: od = outcomes[uc["id"]] entry = by_id.get(uc["id"]) if entry is None: entry = {"id": uc["id"], "name": uc["name"], "category": uc["category"], "problemMagnitude": None, "need": None, "scale": None, "blockchainFit": None, "readiness": None, "readinessSource": "none", "gates": None, "tier": "not-ranked", "rank": None, "inTop10": False, "signalsFired": 0, "signalsAvailable": 0, "scoringMethod": "none", "evidence": [], "phase4": None} use_cases.append(entry) not_enough.append({"id": uc["id"], "name": uc["name"], "category": uc["category"], "stage": "ranking", "reason": "This use case's category was not in the selected categories, so it was not ranked."}) entry["phase4"] = od if od["outcome"] == "NOT_ASSESSED": not_enough.append({"id": uc["id"], "name": uc["name"], "category": uc["category"], "stage": "readiness", "reason": od["rule"]["sentence"]}) elif od["rule"]["id"] == "no-part-a": not_enough.append({"id": uc["id"], "name": uc["name"], "category": uc["category"], "stage": "readiness", "reason": od["rule"]["sentence"]}) ctx = agility["ctx"] result = { "scorerVersion": SCORER_VERSION, "inputNotes": notes, "context": { "selectedCategories": selected, "starredUseCases": [u["id"] for u in flat], "starredSource": flat_source, "interagencyContext": eng.interagency or None, "detectedSector": eng.detected_sector, "programme": eng.programme, "organisation": {"sizeBucket": agility["orgSizeBucket"], "staffCount": agility["staffCount"], "generalAgility": agility["general"], "complianceAgility": agility["compliance"], "hasComplianceStaff": agility["hasComplianceStaff"], "operatingContext": ctx["operatingContext"] or None}, }, "tiers": {k: [x["ucId"] for x in v] for k, v in tiers_known.items()}, "useCases": use_cases, "notEnoughInformation": not_enough, } return result def normalise_numbers(o, places=None): if isinstance(o, float): if math.isnan(o) or math.isinf(o): return None if places is not None: o = float(to_fixed(o, places)) if o == int(o) and abs(o) < 1e15: return int(o) return o if isinstance(o, dict): return {k: normalise_numbers(v, places) for k, v in o.items()} if isinstance(o, list): return [normalise_numbers(v, places) for v in o] return o def main(argv=None): ap = argparse.ArgumentParser(description="Deterministic scorer for the NGO Blockchain Readiness Framework.") ap.add_argument("--answers", required=True, help="JSON file of answers in the site's storage shape") ap.add_argument("--data", required=True, help="the v1.0-ngo data folder") ap.add_argument("--out", help="write the result here instead of printing it") ap.add_argument("--compact", action="store_true", help="no indentation") ap.add_argument("--precise", action="store_true", help="do not round numbers to 3 decimal places (used by the parity test)") args = ap.parse_args(argv) data = Data(args.data) with open(args.answers, "r", encoding="utf-8") as fh: raw = json.load(fh) result = normalise_numbers(score(data, raw), None if args.precise else 3) if args.compact: text = json.dumps(result, separators=(",", ":"), ensure_ascii=False) else: text = json.dumps(result, indent=2, ensure_ascii=False) if args.out: with open(args.out, "w", encoding="utf-8") as fh: fh.write(text + "\n") else: sys.stdout.write(text + "\n") return 0 if __name__ == "__main__": sys.exit(main())