"""Minimal pure-Python SAX -> SAB encoder, ported from sedona.offline.OfflineApp.""" import os, struct, sys, base64, datetime import xml.etree.ElementTree as ET import config class Buf: def __init__(self): self.b = bytearray() def u1(self, v): self.b.append(v & 0xFF) def u2(self, v): self.b += struct.pack(">H", v & 0xFFFF) def i4(self, v): self.b += struct.pack(">i", v if -2**31 <= v < 2**31 else v - 2**32) def i8(self, v): self.b += struct.pack(">q", v) def f4(self, v): self.b += struct.pack(">f", v) def f8(self, v): self.b += struct.pack(">d", v) def str_(self, s): self.b += s.encode("ascii"); self.u1(0) # ---------- schema ---------- def load_kits(parts, home): """[(kitName, checksumHexOrNone)] in kit-id order -> (kits, types_by_qname).""" kits = [] # [(name, checksum, [typeElem, ...])] types_by_q = {} for kit_id, (name, cks) in enumerate(parts): d = os.path.join(home, "manifests", name) f = os.path.join(d, "%s-%s.xml" % (name, cks)) if cks else None if not f or not os.path.exists(f): cands = sorted(x for x in os.listdir(d) if x.startswith(name + "-") and x.endswith(".xml")) f = os.path.join(d, cands[-1]) mroot = ET.parse(f).getroot() tlist = sorted(mroot.findall("type"), key=lambda t: int(t.get("id"))) for t in tlist: types_by_q["%s::%s" % (name, t.get("name"))] = t kits.append((name, int(mroot.get("checksum"), 16), tlist)) return kits, types_by_q def load_schema(sax_root, home): parts = [(k.get("name"), k.get("checksum")) for k in sax_root.find("schema").findall("kit")] # sortKits: sys first, rest alphabetical parts.sort(key=lambda p: ("" if p[0] == "sys" else "\x01" + p[0])) return load_kits(parts, home) def resolve_slots(qname, types_by_q, cache): """Flatten base chain -> [(name, type, flags, default)] indexed by slot id.""" if qname in cache: return cache[qname] t = types_by_q[qname] base = t.get("base") out = list(resolve_slots(base, types_by_q, cache)) if base else [] byname = {s[0]: i for i, s in enumerate(out)} for s in sorted(t.findall("slot"), key=lambda s: int(s.get("id"))): rec = (s.get("name"), s.get("type"), s.get("flags", ""), s.get("default")) if rec[0] in byname: # action override keeps the inherited id out[byname[rec[0]]] = rec else: byname[rec[0]] = len(out); out.append(rec) cache[qname] = out return out # ---------- values ---------- def _int(raw): # Integer.decode: supports 0x / # / leading 0 octal if raw in (None, "", "null"): return 0 return int(raw, 0) if raw.lower().startswith(("0x", "-0x")) else int(raw) def _flt(raw): if raw == "null": return float("nan") # explicit null only if raw in (None, ""): return 0.0 # Value.defaultForType -> ZERO return float(raw) def encode_value(out, vtype, flags, raw): if "s" in flags: # asStr Buf s = raw or "" out.u2(len(s) + 1) out.b += s.encode("ascii"); out.u1(0) elif vtype == "bool": out.u1(2 if raw == "null" else (1 if raw in ("true", "1") else 0)) elif vtype == "byte": out.u1(_int(raw)) elif vtype == "short": out.u2(_int(raw)) elif vtype == "int": out.i4(_int(raw)) elif vtype == "long": out.i8(_int(raw)) elif vtype == "float": out.f4(_flt(raw)) elif vtype == "double": out.f8(_flt(raw)) elif vtype == "sys::Buf" or vtype == "buf": data = base64.b64decode(raw) if raw else b"" out.u2(len(data)); out.b += data else: raise Exception("unhandled slot type " + str(vtype)) # ---------- main ---------- def default_sab_name(sax_path): """`app.sax` -> `app-sab2sax-20260728-140609.sab` -- never clobbers an existing .sab.""" stem = os.path.splitext(os.path.basename(sax_path))[0] stamp = datetime.datetime.now().strftime("%Y%m%d-%H%M%S") return "%s-sab2sax-%s.sab" % (stem, stamp) def encode(sax_path, home=None, sab_path=None): if home is None: home = config.get_sedona_home() if sab_path is None: sab_path = os.path.join(os.path.dirname(os.path.abspath(sax_path)), default_sab_name(sax_path)) root = ET.parse(sax_path).getroot() kits, types_by_q = load_schema(root, home) kit_id = {k[0]: i for i, k in enumerate(kits)} type_id = {} for ki, (kname, _, tlist) in enumerate(kits): for t in tlist: type_id["%s::%s" % (kname, t.get("name"))] = (ki, int(t.get("id"))) cache = {} # walk tree -> comps with (elem, type, parent, kids) comps = [] def walk(elem, qtype, parent): rec = {"elem": elem, "type": qtype, "parent": parent, "kids": [], "id": int(elem.get("id")) if elem.get("id") else None} comps.append(rec) if parent is not None: parent["kids"].append(rec) for c in elem.findall("comp"): walk(c, c.get("type"), rec) return rec app_elem = root.find("app") app = walk(app_elem, "sys::App", None) app["id"] = 0 # assign missing ids used = {c["id"] for c in comps if c["id"] is not None} nxt = 1 for c in comps: if c["id"] is None: while nxt in used: nxt += 1 c["id"] = nxt; used.add(nxt) by_id = {c["id"]: c for c in comps} out = Buf() out.i4(0x73617070) # "sapp" out.i4(0x0003) # version 0.3 out.u1(len(kits)) # schema for name, cks, _ in kits: out.str_(name); out.i4(cks) out.u2(max(by_id)) # maxId for cid in sorted(by_id): # components, ascending id c = by_id[cid] slots = resolve_slots(c["type"], types_by_q, cache) ki, ti = type_id[c["type"]] out.u2(cid); out.u1(ki); out.u1(ti) out.str_(c["elem"].get("name", "app") if cid else "app") out.u2(c["parent"]["id"] if c["parent"] else 0xFFFF) out.u2(c["kids"][0]["id"] if c["kids"] else 0xFFFF) sib = 0xFFFF if c["parent"]: k = c["parent"]["kids"]; i = k.index(c) if i + 1 < len(k): sib = k[i + 1]["id"] out.u2(sib) props = {p.get("name"): p.get("val") for p in c["elem"].findall("prop")} for name, vtype, flags, dflt in slots: if "a" in flags or "c" not in flags: continue # config props only encode_value(out, vtype, flags, props.get(name, dflt)) out.u1(ord(';')) out.u2(0xFFFF) links = root.find("links") if links is not None: for ln in links.findall("link"): def ref(s): path, slot = s.rsplit(".", 1) cur = app for part in [p for p in path.split("/") if p]: cur = next(k for k in cur["kids"] if k["elem"].get("name") == part) sl = resolve_slots(cur["type"], types_by_q, cache) return cur["id"], [x[0] for x in sl].index(slot) fc, fs = ref(ln.get("from")); tc, ts = ref(ln.get("to")) out.u2(fc); out.u1(fs); out.u2(tc); out.u1(ts) out.u2(0xFFFF) out.u1(ord('.')) open(sab_path, "wb").write(bytes(out.b)) return len(out.b) if __name__ == "__main__": import sab2sax # the CLI converts either direction; this is just one half sab2sax.main(sys.argv)