| """
|
| seb/adapters/rpg_ingest.py
|
| Cherry-picked from RBG-ibm-meta-corpus/ingest/rpg_matrix.py
|
| Extended: maps RPG data-flow IR to SEB event schema.
|
|
|
| Pipeline:
|
| fixed-format RPG source (80-col)
|
| → parse rows (F/I/C/O spec columns)
|
| → extract reads/writes/opcodes
|
| → build data flow graph
|
| → map operations to SEB event types
|
| → emit SEB event schema JSON
|
|
|
| SEB event type mapping (from seb_types.ads EventTypeRegistry):
|
| WRITE → target is an audit/ledger file → FISCAL_SETTLE 0x0100
|
| WRITE → target is any other output file → INFRA_PROVISION 0x0001
|
| CHAIN → DB2 lookup (read by key) → ARCH_DECISION 0x0010 (query)
|
| EXSR → subroutine call → CONFIG_DEPLOY 0x0002
|
| SETON LR → end-of-job → SOVEREIGN_ROOT 0xFFFF (if LR)
|
|
|
| Usage:
|
| python rpg_ingest.py <path-to-fixed-rpg> # parse + emit SEB schema
|
| python rpg_ingest.py <path> --seb-events # show SEB event list only
|
| python rpg_ingest.py <path> --adapter-stub # emit RPGLE adapter stub
|
|
|
| Dependencies: stdlib only (no pip installs)
|
| """
|
|
|
| from __future__ import annotations
|
|
|
| import json
|
| import re
|
| import sys
|
| from collections import Counter
|
| from pathlib import Path
|
|
|
|
|
| COLUMN_MAP: dict = {
|
| "format": "fixed-rpg-80",
|
| "base_columns": {
|
| "sequence": {"start": 1, "end": 5},
|
| "spec": {"start": 6, "end": 6},
|
| "comment_tail": {"start": 54, "end": 80},
|
| },
|
| "layouts": {
|
| "token_window": {"start": 7, "end": 60}
|
| },
|
| "spec_types": {
|
| "H": "header", "F": "file", "E": "extension",
|
| "L": "line_counter", "I": "input", "C": "calculation",
|
| "O": "output", "*": "comment",
|
| },
|
| "opcode_classes": {
|
| "ADD": "arithmetic", "SUB": "arithmetic", "MULT": "arithmetic",
|
| "DIV": "arithmetic", "Z-ADD": "arithmetic", "MOVE": "move",
|
| "MOVEL": "move", "CHAIN": "io", "READ": "io",
|
| "WRITE": "io", "READE": "io", "READP": "io",
|
| "IFEQ": "condition", "IFNE": "condition", "IFLT": "condition",
|
| "IFGT": "condition", "IFLE": "condition", "IFGE": "condition",
|
| "ELSE": "condition", "END": "condition", "BEGSR": "subroutine",
|
| "ENDSR": "subroutine","EXSR": "subroutine", "PLIST": "parameter",
|
| "PARM": "parameter", "SETON": "indicator", "SETOF": "indicator",
|
| "CALL": "call", "RETURN": "call",
|
| }
|
| }
|
|
|
|
|
| SEB_INFRA_PROVISION = 0x0001
|
| SEB_CONFIG_DEPLOY = 0x0002
|
| SEB_ARCH_DECISION = 0x0010
|
| SEB_FISCAL_SETTLE = 0x0100
|
| SEB_SOVEREIGN_ROOT = 0xFFFF
|
|
|
|
|
| FISCAL_PATTERNS = re.compile(
|
| r'(LEDGER|AUDIT|SETTLE|FISCAL|GL_|PAY|JOURNAL|JRN|LEDGE|VAULT)',
|
| re.IGNORECASE
|
| )
|
|
|
| TOKEN_RE = re.compile(r"\S+")
|
|
|
|
|
|
|
|
|
| def pad_line(line: str, width: int = 80) -> str:
|
| return line.rstrip("\n").rstrip("\r")[:width].ljust(width)
|
|
|
| def slice_1(text: str, start: int, end: int) -> str:
|
| return text[start - 1 : end]
|
|
|
| def normalize(s: str) -> str | None:
|
| v = " ".join(s.strip().split())
|
| return v or None
|
|
|
| def classify_opcode(op: str) -> str:
|
| return COLUMN_MAP["opcode_classes"].get(op.strip().upper(), "unknown")
|
|
|
| def extract_tokens(line: str, start: int = 7, end: int = 53) -> list[str]:
|
| return TOKEN_RE.findall(slice_1(line, start, end))
|
|
|
| def find_opcode_idx(tokens: list[str]) -> int | None:
|
| for i, t in enumerate(tokens):
|
| if t.strip().upper() in COLUMN_MAP["opcode_classes"]:
|
| return i
|
| return None
|
|
|
| def base_row(line: str, lineno: int) -> tuple[dict, str]:
|
| cols = COLUMN_MAP["base_columns"]
|
| spec = slice_1(line, cols["spec"]["start"], cols["spec"]["end"]).strip().upper()
|
| if not spec and slice_1(line, 7, 7) == "*":
|
| spec = "*"
|
| row = {
|
| "line_number": lineno,
|
| "raw": line,
|
| "spec_code": spec,
|
| "spec_kind": COLUMN_MAP["spec_types"].get(spec, "unknown"),
|
| "fields": {"sequence": slice_1(line, 1, 5).rstrip(),
|
| "comment": slice_1(line, cols["comment_tail"]["start"],
|
| cols["comment_tail"]["end"]).rstrip()},
|
| "classification": {"domain": COLUMN_MAP["spec_types"].get(spec, "unknown"),
|
| "is_comment": spec == "*",
|
| "is_blank": not line.strip()},
|
| "reads": [], "writes": [], "warnings": [],
|
| }
|
| return row, spec
|
|
|
| def decode_file_row(line: str, row: dict) -> dict:
|
| toks = extract_tokens(line)
|
| name = toks[0] if toks else ""
|
| row["fields"].update({"file_name": name, "access": toks[1] if len(toks) > 1 else "",
|
| "keywords": toks[2:]})
|
| if name: row["writes"].append(name)
|
| else: row["warnings"].append("file row missing file_name")
|
| return row
|
|
|
| def decode_input_row(line: str, row: dict) -> dict:
|
| toks = extract_tokens(line)
|
| if not toks:
|
| row["warnings"].append("input row has no tokens"); return row
|
| if toks[0].isdigit():
|
| row["classification"]["domain"] = "input_field"
|
| fn = toks[2] if len(toks) > 2 else ""
|
| row["fields"].update({"input_shape": "field", "field_from": toks[0],
|
| "field_to": toks[1] if len(toks) > 1 else "",
|
| "field_name": fn})
|
| if fn: row["writes"].append(fn)
|
| else:
|
| row["classification"]["domain"] = "input_record"
|
| row["fields"].update({"input_shape": "record", "record_name": toks[0]})
|
| row["reads"].append(toks[0])
|
| return row
|
|
|
| def decode_calculation_row(line: str, row: dict) -> dict:
|
| toks = extract_tokens(line)
|
| idx = find_opcode_idx(toks)
|
| comment = slice_1(line, 54, 80).rstrip()
|
| if idx is None:
|
| row["fields"].update({"factor1": "", "opcode": "", "factor2": "",
|
| "result": "", "comment": comment, "tokens": toks})
|
| row["classification"]["domain"] = "unknown"
|
| row["warnings"].append("calculation row missing recognized opcode")
|
| return row
|
| op = toks[idx].strip().upper()
|
| before = toks[:idx]; after = toks[idx + 1:]
|
| f1 = before[-1] if before else ""
|
| f2 = after[0] if after else ""
|
| result = after[1] if len(after) > 1 else ""
|
| if op in {"WRITE", "SETON", "SETOF", "END", "ELSE", "BEGSR", "ENDSR", "EXSR"}:
|
| result = ""
|
| row["fields"].update({"factor1": f1, "opcode": op, "factor2": f2,
|
| "result": result, "comment": comment, "tokens": toks})
|
| row["classification"]["domain"] = classify_opcode(op)
|
| reads, writes = [], []
|
| if f1 := normalize(f1): reads.append(f1)
|
| if f2v := normalize(f2):
|
| if op not in {"WRITE", "SETON"}: reads.append(f2v)
|
| if rv := normalize(result):
|
| if op in {"Z-ADD","ADD","SUB","MULT","DIV","MOVEL","MOVE","PARM"}: writes.append(rv)
|
| if op == "CHAIN" and f2v: reads.append(f2v)
|
| if op == "WRITE" and f2v: writes.append(f2v)
|
| if op == "SETON" and f2v: writes.append(f2v)
|
| if op in {"BEGSR","EXSR","ENDSR"} and f1: writes.append(f1)
|
| row["reads"] = list(dict.fromkeys(reads))
|
| row["writes"] = list(dict.fromkeys(writes))
|
| return row
|
|
|
| def decode_output_row(line: str, row: dict) -> dict:
|
| toks = extract_tokens(line)
|
| rec = toks[0] if toks else ""
|
| row["classification"]["domain"] = "output"
|
| row["fields"].update({"output_record": rec,
|
| "operation": toks[1] if len(toks) > 1 else "",
|
| "operands": toks[2:]})
|
| if rec: row["writes"].append(rec)
|
| else: row["warnings"].append("output row missing output_record")
|
| return row
|
|
|
| def decode_row(line: str, lineno: int) -> dict:
|
| row, spec = base_row(line, lineno)
|
| if row["classification"]["is_blank"] or row["classification"]["is_comment"]:
|
| return row
|
| if spec == "F": return decode_file_row(line, row)
|
| if spec == "I": return decode_input_row(line, row)
|
| if spec == "C": return decode_calculation_row(line, row)
|
| if spec == "O": return decode_output_row(line, row)
|
| return row
|
|
|
| def build_matrix(lines: list[str]) -> list[list[int]]:
|
| return [[ord(ch) for ch in line] for line in lines]
|
|
|
| def build_flow(rows: list[dict]) -> dict:
|
| nodes, edges = [], []
|
| for row in rows:
|
| nid = f"line_{row['line_number']}"
|
| nodes.append({"id": nid, "line_number": row["line_number"],
|
| "spec_kind": row["spec_kind"],
|
| "opcode": row["fields"].get("opcode", "").strip(),
|
| "domain": row["classification"]["domain"],
|
| "warnings": row["warnings"]})
|
| for s in row["reads"]: edges.append({"type": "reads", "symbol": s, "target": nid})
|
| for s in row["writes"]: edges.append({"type": "writes", "symbol": s, "source": nid})
|
| return {"nodes": nodes, "edges": edges}
|
|
|
| def build_summary(rows: list[dict]) -> dict:
|
| return {"spec_counts": dict(Counter(r["spec_kind"] for r in rows)),
|
| "domain_counts": dict(Counter(r["classification"]["domain"] for r in rows)),
|
| "warnings": [{"line_number": r["line_number"], "warnings": r["warnings"]}
|
| for r in rows if r["warnings"]]}
|
|
|
|
|
|
|
|
|
| def row_to_seb_event(row: dict) -> dict | None:
|
| """Map a single RPG calculation row to a SEB event descriptor."""
|
| op = row["fields"].get("opcode", "").strip().upper()
|
| if not op:
|
| return None
|
|
|
| writes = row.get("writes", [])
|
| reads = row.get("reads", [])
|
| lineno = row["line_number"]
|
|
|
|
|
| if op == "WRITE":
|
| for w in writes:
|
| if FISCAL_PATTERNS.search(w):
|
| return {"line": lineno, "opcode": op, "target": w,
|
| "seb_event_type": SEB_FISCAL_SETTLE,
|
| "seb_event_name": "FISCAL_SETTLE",
|
| "required_capability": "execute",
|
| "weight": 0xFFFFFFFF,
|
| "human_review_required": True}
|
| for w in writes:
|
| return {"line": lineno, "opcode": op, "target": w,
|
| "seb_event_type": SEB_INFRA_PROVISION,
|
| "seb_event_name": "INFRA_PROVISION",
|
| "required_capability": "execute",
|
| "weight": 1000,
|
| "human_review_required": False}
|
|
|
|
|
| if op == "CHAIN":
|
| target = reads[-1] if reads else "unknown"
|
| return {"line": lineno, "opcode": op, "target": target,
|
| "seb_event_type": SEB_ARCH_DECISION,
|
| "seb_event_name": "ARCH_DECISION",
|
| "required_capability": "verify",
|
| "weight": 100,
|
| "human_review_required": False}
|
|
|
|
|
| if op == "EXSR":
|
| sub = reads[0] if reads else writes[0] if writes else "unknown"
|
| return {"line": lineno, "opcode": op, "target": sub,
|
| "seb_event_type": SEB_CONFIG_DEPLOY,
|
| "seb_event_name": "CONFIG_DEPLOY",
|
| "required_capability": "write",
|
| "weight": 500,
|
| "human_review_required": False}
|
|
|
|
|
| if op == "SETON":
|
| factor2 = row["fields"].get("factor2", "").strip().upper()
|
| if "LR" in factor2 or "LR" in writes:
|
| return {"line": lineno, "opcode": op, "target": "LR",
|
| "seb_event_type": SEB_SOVEREIGN_ROOT,
|
| "seb_event_name": "SOVEREIGN_ROOT",
|
| "required_capability": "vacuum_collapse",
|
| "weight": 0xFFFFFFFF,
|
| "human_review_required": True}
|
| return None
|
|
|
|
|
| def extract_seb_events(rows: list[dict]) -> list[dict]:
|
| events = []
|
| for row in rows:
|
| if row["spec_code"] == "C":
|
| evt = row_to_seb_event(row)
|
| if evt:
|
| events.append(evt)
|
| return events
|
|
|
|
|
| def emit_adapter_stub(source_name: str, events: list[dict]) -> str:
|
| """Emit an RPGLE adapter stub that emits the detected SEB events."""
|
| lines = [
|
| f"** SEB adapter stub generated from {source_name}",
|
| f"** Cherry-picked rpg_ingest.py from RBG-ibm-meta-corpus",
|
| f"** DO NOT EDIT — regenerate with: python rpg_ingest.py {source_name} --adapter-stub",
|
| "**FREE",
|
| "ctl-opt dftactgrp(*no) actgrp(*new) bnddir('SEB_BND');",
|
| "",
|
| "// SEB kernel entry points",
|
| "dcl-pr SEB_Append_Event extproc(*dclcase);",
|
| " Hdr pointer value options(*nopass);",
|
| " Pay pointer value options(*nopass);",
|
| " Ftr pointer value options(*nopass);",
|
| "end-pr;",
|
| "",
|
| "dcl-pr SEB_Worm_Flush extproc(*dclcase);",
|
| "end-pr;",
|
| "",
|
| ]
|
|
|
| for evt in events:
|
| ename = evt["seb_event_name"]
|
| etype = evt["seb_event_type"]
|
| cap = evt["required_capability"]
|
| hr = evt["human_review_required"]
|
| target = evt["target"]
|
| lines += [
|
| f"// Line {evt['line']}: {evt['opcode']} {target}",
|
| f"// SEB event: {ename} (0x{etype:04X}) cap={cap} human_review={hr}",
|
| f"dcl-proc Emit_{ename}_{evt['line']} export;",
|
| f" // TODO: build 68-byte Event_Header with Event_Type = 0x{etype:04X}",
|
| f" // TODO: build payload from {target} record",
|
| f" // TODO: call SEB_Append_Event(hdr_ptr : pay_ptr : ftr_ptr);",
|
| f" // TODO: if human_review_required call SEB_Human_Review_Queue;",
|
| f"end-proc;",
|
| "",
|
| ]
|
|
|
| lines += ["// Flush WORM chain after all events emitted",
|
| "SEB_Worm_Flush();", "*inlr = *on;"]
|
| return "\n".join(lines)
|
|
|
|
|
|
|
|
|
| def main(argv: list[str]) -> int:
|
| if len(argv) < 2:
|
| print("usage: python rpg_ingest.py <path-to-fixed-rpg> [--seb-events] [--adapter-stub]")
|
| return 1
|
|
|
| source_path = Path(argv[1]).resolve()
|
| if not source_path.exists():
|
| print(f"error: not found: {source_path}"); return 1
|
|
|
| mode_events = "--seb-events" in argv
|
| mode_adapter = "--adapter-stub" in argv
|
|
|
| lines = [pad_line(l) for l in source_path.read_text(encoding="utf-8").splitlines()]
|
| rows = [decode_row(l, i + 1) for i, l in enumerate(lines)]
|
| flow = build_flow(rows)
|
| summary = build_summary(rows)
|
| events = extract_seb_events(rows)
|
|
|
| if mode_adapter:
|
| print(emit_adapter_stub(source_path.name, events))
|
| return 0
|
|
|
| if mode_events:
|
| print(json.dumps(events, indent=2))
|
| return 0
|
|
|
|
|
| out = {
|
| "source": str(source_path),
|
| "shape": [len(rows), 80],
|
| "summary": summary,
|
| "seb_events": events,
|
| "flow": flow,
|
| "rows": rows,
|
| "matrix": build_matrix(lines),
|
| }
|
| print(json.dumps(out, indent=2))
|
|
|
|
|
| import sys as _sys
|
| print(f"\nsource: {source_path}", file=_sys.stderr)
|
| print(f"specs: {summary['spec_counts']}", file=_sys.stderr)
|
| print(f"seb_events: {len(events)}", file=_sys.stderr)
|
| for e in events:
|
| print(f" line {e['line']:4d} {e['opcode']:8s} → {e['seb_event_name']} (0x{e['seb_event_type']:04X})",
|
| file=_sys.stderr)
|
| print(f"warnings: {len(summary['warnings'])}", file=_sys.stderr)
|
| return 0
|
|
|
|
|
| if __name__ == "__main__":
|
| raise SystemExit(main(sys.argv))
|
|
|
