Py_sax2sab/sab.py

"""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)