167 lines
6.7 KiB
Python
167 lines
6.7 KiB
Python
"""Minimal pure-Python SAX -> SAB encoder, ported from sedona.offline.OfflineApp."""
|
|
import os, struct, sys, base64
|
|
import xml.etree.ElementTree as ET
|
|
|
|
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_schema(sax_root, home):
|
|
parts = []
|
|
for k in sax_root.find("schema").findall("kit"):
|
|
parts.append((k.get("name"), k.get("checksum")))
|
|
# sortKits: sys first, rest alphabetical
|
|
parts.sort(key=lambda p: ("" if p[0] == "sys" else "\x01" + p[0]))
|
|
|
|
kits = [] # [(name, checksum, {typeName: 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 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 encode(sax_path, home, sab_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__":
|
|
n = encode(sys.argv[1], sys.argv[2], sys.argv[3])
|
|
print("wrote", n, "bytes")
|