26 class encoder(object):
27 def __init__(self, *, backrefs=True):
28 self.backrefs = backrefs
35 return bytes([(sec << 3) | pri])
37 def writetag(self, dst, pri, sec, datum):
38 dst.write(self.enctag(pri, sec))
42 if datum is not None and id(datum) not in self.reftab:
43 self.reftab[id(datum)] = ref
53 while (x > 0) or (b & 0x40) != 0:
61 while x < -1 or (b & 0x40) == 0:
69 def writestr(dst, text):
70 dst.write(text.encode("utf-8"))
73 def dumpseq(self, dst, seq):
76 dst.write(self.enctag(T_END, 0))
78 def dumpmap(self, dst, val):
79 for k, v in val.items():
82 dst.write(self.enctag(T_END, 0))
84 def dump(self, dst, datum):
85 ref = self.reftab.get(id(datum))
87 dst.write(self.enctag(T_INT, INT_REF))
88 dst.write(self.encint(ref))
91 self.writetag(dst, T_NIL, 0, None)
93 self.writetag(dst, T_NIL, NIL_FALSE, None)
95 self.writetag(dst, T_NIL, NIL_TRUE, None)
96 elif isinstance(datum, int):
97 self.writetag(dst, T_INT, 0, None)
98 dst.write(self.encint(datum))
99 elif isinstance(datum, str):
100 self.writetag(dst, T_STR, 0, datum)
101 self.writestr(dst, datum)
102 elif isinstance(datum, (bytes, bytearray)):
103 self.writetag(dst, T_BIT, 0, datum)
104 dst.write(self.encint(len(datum)))
106 elif isinstance(datum, data.symbol):
108 self.writetag(dst, T_STR, STR_SYM, datum)
109 self.writestr(dst, datum.name)
111 nsref = self.nstab.get(datum.ns)
113 nsref = self.writetag(dst, T_SYM, 0, datum)
115 self.writestr(dst, datum.ns)
116 self.writestr(dst, datum.name)
117 if nsref is not None:
118 self.nstab[datum.ns] = nsref
120 self.writetag(dst, T_SYM, 0, datum)
122 dst.write(self.encint(nsref))
123 self.writestr(dst, datum.name)
124 elif isinstance(datum, list):
125 self.writetag(dst, T_CON, CON_SEQ, datum)
126 self.dumpseq(dst, datum)
127 elif isinstance(datum, set):
128 self.writetag(dst, T_CON, CON_SET, datum)
129 self.dumpseq(dst, datum)
130 elif isinstance(datum, dict):
131 self.writetag(dst, T_CON, CON_MAP, datum)
132 self.dumpmap(dst, datum)
133 elif isinstance(datum, data.obj):
134 self.writetag(dst, T_CON, CON_OBJ, datum)
135 self.dump(dst, getattr(type(datum), "typename", None))
136 self.dumpmap(dst, datum.__dict__)
138 raise ValueError("unsupported object type: " + repr(datum))
140 def dump(dst, datum):
141 encoder().dump(dst, datum)
144 class fmterror(Exception):
147 class eoferror(fmterror):
149 super().__init__("unexpected end-of-data")
151 class referror(fmterror):
153 super().__init__("bad backref")
155 class namedtype(type):
156 def __new__(cls, *args, typename=None, **kwargs):
157 self = super().__new__(cls, *args, **kwargs)
158 self.typename = typename
161 class decoder(object):
179 ret += (b & 0x7f) << p
195 return buf.decode("utf-8")
197 def loadsym(self, fp):
200 nsref = self.loadint(fp)
201 if not 0 <= nsref < len(self.reftab):
202 raise fmterror("illegal namespace ref: " + str(nsref))
203 nssym = self.reftab[nsref]
204 if not isinstance(nssym, data.symbol):
205 raise fmterror("illegal namespace ref: " + str(nsref))
208 ns = self.loadstr(fp)
209 nm = self.loadstr(fp)
210 ret = data.symbol.get(ns, nm)
213 def loadlist(self, fp, buf):
218 buf.append(self.loadtagged(fp, tag))
220 def loadmap(self, fp, buf):
225 key = self.loadtagged(fp, tag)
229 buf[key] = self.loadtagged(fp, tag)
231 def makeobjtype(self, nm):
232 return namedtype(str(nm), (data.obj, object), {}, typename=nm)
234 def loadobj(self, fp, ref=False):
236 refid = len(self.reftab)
237 self.reftab.append(None)
239 typ = self.namedtypes.get(nm)
241 typ = self.namedtypes[nm] = self.makeobjtype(nm)
244 self.reftab[refid] = ret
246 # print(">", nm, hex(st))
247 ret.__dict__.update(self.loadmap(fp, {}))
248 # print("<", nm, hex(fp.tell()), hex(st))
251 def addref(self, obj):
252 self.reftab.append(obj)
255 def loadtagged(self, fp, tag):
256 pri, sec = (tag & 0x7), (tag & 0xf8) >> 3
258 raise fmterror("unexpected end-tag")
261 idx = self.loadint(fp)
262 if not 0 <= idx < len(self.reftab):
264 # print(idx, self.reftab[idx], hex(fp.tell()))
265 return self.reftab[idx]
266 return self.addref(self.loadint(fp))
268 ret = self.addref(self.loadstr(fp))
270 return data.symbol.get("", ret)
273 ln = self.loadint(fp)
274 ret = self.addref(fp.read(ln))
280 return self.addref(True)
281 elif sec == NIL_FALSE:
282 return self.addref(False)
283 return self.addref(None)
285 return self.addref(self.loadsym(fp))
288 return self.loadmap(fp, self.addref({}))
290 return self.loadobj(fp, ref=True)
292 return self.loadlist(fp, self.addref([]))
294 raise fmterror("unknown primary: " + str(pri))
298 return self.loadtagged(fp, tag)