[coe.git] / coe / bin.py

from . import data

T_END = 0
T_INT = 1
T_STR = 2
T_BIT = 3
T_NIL = 4
T_SYM = 5
T_CON = 6

INT_REF = 1

STR_SYM = 1

BIT_BFLOAT = 1
BIT_DFLOAT = 2

CON_SEQ = 0
CON_SET = 1
CON_MAP = 2
CON_OBJ = 3

NIL_FALSE = 1
NIL_TRUE = 2

class encoder(object):
    def __init__(self, *, backrefs=True):
        self.backrefs = backrefs
        self.reftab = {}
        self.nextref = 0
        self.nstab = {}

    @staticmethod
    def enctag(pri, sec):
        return bytes([(sec << 3) | pri])

    def writetag(self, dst, pri, sec, datum):
        dst.write(self.enctag(pri, sec))
        if self.backrefs:
            ref = self.nextref
            self.nextref += 1
            if datum is not None and id(datum) not in self.reftab:
                self.reftab[id(datum)] = ref
            return ref
        return None

    @staticmethod
    def encint(x):
        ret = bytearray()
        if x >= 0:
            b = x & 0x7f
            x >>= 7
            while (x > 0) or (b & 0x40) != 0:
                ret.append(0x80 | b)
                b = x & 0x7f
                x >>= 7
            ret.append(b)
        elif x < 0:
            b = x & 0x7f
            x >>= 7
            while x < -1 or (b & 0x40) == 0:
                ret.append(0x80 | b)
                b = x & 0x7f
                x >>= 7
            ret.append(b)
        return ret

    @staticmethod
    def writestr(dst, text):
        dst.write(text.encode("utf-8"))
        dst.write(b'\0')

    def dumpseq(self, dst, seq):
        for v in seq:
            self.dump(dst, v)
        dst.write(self.enctag(T_END, 0))

    def dumpmap(self, dst, val):
        for k, v in val.items():
            self.dump(dst, k)
            self.dump(dst, v)
        dst.write(self.enctag(T_END, 0))

    def dump(self, dst, datum):
        ref = self.reftab.get(id(datum))
        if ref is not None:
            dst.write(self.enctag(T_INT, INT_REF))
            dst.write(self.encint(ref))
            return
        if datum == None:
            self.writetag(dst, T_NIL, 0, None)
        elif datum == False:
            self.writetag(dst, T_NIL, NIL_FALSE, None)
        elif datum == True:
            self.writetag(dst, T_NIL, NIL_TRUE, None)
        elif isinstance(datum, int):
            self.writetag(dst, T_INT, 0, None)
            dst.write(self.encint(datum))
        elif isinstance(datum, str):
            self.writetag(dst, T_STR, 0, datum)
            self.writestr(dst, datum)
        elif isinstance(datum, (bytes, bytearray)):
            self.writetag(dst, T_BIT, 0, datum)
            dst.write(self.encint(len(datum)))
            dst.write(datum)
        elif isinstance(datum, data.symbol):
            if datum.ns == "":
                self.writetag(dst, T_STR, STR_SYM, datum)
                self.writestr(dst, datum.name)
            else:
                nsref = self.nstab.get(datum.ns)
                if nsref is None:
                    nsref = self.writetag(dst, T_SYM, 0, datum)
                    dst.write(b'\0')
                    self.writestr(dst, datum.ns)
                    self.writestr(dst, datum.name)
                    if nsref is not None:
                        self.nstab[datum.ns] = nsref
                else:
                    self.writetag(dst, T_SYM, 0, datum)
                    dst.write(b'\x01')
                    dst.write(self.encint(nsref))
                    self.writestr(dst, datum.name)
        elif isinstance(datum, list):
            self.writetag(dst, T_CON, CON_SEQ, datum)
            self.dumpseq(dst, datum)
        elif isinstance(datum, set):
            self.writetag(dst, T_CON, CON_SET, datum)
            self.dumpseq(dst, datum)
        elif isinstance(datum, dict):
            self.writetag(dst, T_CON, CON_MAP, datum)
            self.dumpmap(dst, datum)
        elif isinstance(datum, data.obj):
            self.writetag(dst, T_CON, CON_OBJ, datum)
            self.dump(dst, getattr(type(datum), "typename", None))
            self.dumpmap(dst, datum.__dict__)
        else:
            raise ValueError("unsupported object type: " + repr(datum))

def dump(dst, datum):
    encoder().dump(dst, datum)
    return dst

class fmterror(Exception):
    pass

class eoferror(fmterror):
    def __init__(self):
        super().__init__("unexpected end-of-data")

class referror(fmterror):
    def __init__(self):
        super().__init__("bad backref")

class namedtype(type):
    pass

class decoder(object):
    def __init__(self):
        self.reftab = []
        self.namedtypes = {}

    @staticmethod
    def byte(fp):
        b = fp.read(1)
        if b == b"":
            raise eoferror()
        return b[0]

    @staticmethod
    def loadint(fp):
        ret = 0
        p = 0
        while True:
            b = decoder.byte(fp)
            ret += (b & 0x7f) << p
            p += 7
            if (b & 0x80) == 0:
                break
        if (b & 0x40) != 0:
            ret = ret - (1 << p)
        return ret

    @staticmethod
    def loadstr(fp):
        buf = bytearray()
        while True:
            b = decoder.byte(fp)
            if b == 0:
                break
            buf.append(b)
        return buf.decode("utf-8")

    def loadsym(self, fp):
        h = self.byte(fp)
        if h & 0x1:
            nsref = self.loadint(fp)
            if not 0 <= nsref < len(self.reftab):
                raise fmterror("illegal namespace ref: " + str(nsref))
            nssym = self.reftab[nsref]
            if not isinstance(nssym, data.symbol):
                raise fmterror("illegal namespace ref: " + str(nsref))
            ns = nssym.ns
        else:
            ns = self.loadstr(fp)
        nm = self.loadstr(fp)
        ret = data.symbol.get(ns, nm)
        return ret

    def loadlist(self, fp, buf):
        while True:
            tag = self.byte(fp)
            if tag == T_END:
                return buf
            buf.append(self.loadtagged(fp, tag))

    def loadmap(self, fp, buf):
        while True:
            tag = self.byte(fp)
            if tag == T_END:
                return buf
            key = self.loadtagged(fp, tag)
            tag = self.byte(fp)
            if tag == T_END:
                return buf
            buf[key] = self.loadtagged(fp, tag)

    def loadobj(self, fp, ref=False):
        if ref:
            refid = len(self.reftab)
            self.reftab.append(None)
        nm = self.load(fp)
        typ = self.namedtypes.get(nm)
        if typ is None:
            typ = self.namedtypes[nm] = namedtype(str(nm), (data.obj, object), {})
            typ.typename = nm
        ret = typ()
        if ref:
            self.reftab[refid] = ret
        # st = fp.tell()
        # print(">", nm, hex(st))
        ret.__dict__.update(self.loadmap(fp, {}))
        # print("<", nm, hex(fp.tell()), hex(st))
        return ret

    def addref(self, obj):
        self.reftab.append(obj)
        return obj

    def loadtagged(self, fp, tag):
        pri, sec = (tag & 0x7), (tag & 0xf8) >> 3
        if pri == T_END:
            raise fmterror("unexpected end-tag")
        elif pri == T_INT:
            if sec == INT_REF:
                idx = self.loadint(fp)
                if not 0 <= idx < len(self.reftab):
                    raise referror()
                # print(idx, self.reftab[idx], hex(fp.tell()))
                return self.reftab[idx]
            return self.addref(self.loadint(fp))
        elif pri == T_STR:
            ret = self.addref(self.loadstr(fp))
            if sec == STR_SYM:
                return data.symbol.get("", ret)
            return ret
        elif pri == T_BIT:
            ln = self.loadint(fp)
            ret = self.addref(fp.read(ln))
            if len(ret) < ln:
                raise eoferror()
            return ret
        elif pri == T_NIL:
            if sec == NIL_TRUE:
                return self.addref(True)
            elif sec == NIL_FALSE:
                return self.addref(False)
            return self.addref(None)
        elif pri == T_SYM:
            return self.addref(self.loadsym(fp))
        elif pri == T_CON:
            if sec == CON_MAP:
                return self.loadmap(fp, self.addref({}))
            elif sec == CON_OBJ:
                return self.loadobj(fp, ref=True)
            else:
                return self.loadlist(fp, self.addref([]))
        else:
            raise fmterror("unknown primary: " + str(pri))

    def load(self, fp):
        tag = self.byte(fp)
        return self.loadtagged(fp, tag)

def load(fp):
    decoder().load(fp)
Commit	Line	Data
e8a122ff FT	1	from . import data
	2
	3	T_END = 0
	4	T_INT = 1
	5	T_STR = 2
	6	T_BIT = 3
	7	T_NIL = 4
	8	T_SYM = 5
	9	T_CON = 6
	10
	11	INT_REF = 1
	12
	13	STR_SYM = 1
	14
	15	BIT_BFLOAT = 1
	16	BIT_DFLOAT = 2
	17
82855e29	18	CON_SEQ = 0
e8a122ff FT	19	CON_SET = 1
	20	CON_MAP = 2
	21	CON_OBJ = 3
	22
	23	NIL_FALSE = 1
	24	NIL_TRUE = 2
	25
82855e29 FT	26	class encoder(object):
	27	def __init__(self, *, backrefs=True):
	28	self.backrefs = backrefs
	29	self.reftab = {}
	30	self.nextref = 0
	31	self.nstab = {}
	32
	33	@staticmethod
	34	def enctag(pri, sec):
	35	return bytes([(sec << 3) \| pri])
	36
	37	def writetag(self, dst, pri, sec, datum):
	38	dst.write(self.enctag(pri, sec))
	39	if self.backrefs:
	40	ref = self.nextref
	41	self.nextref += 1
	42	if datum is not None and id(datum) not in self.reftab:
	43	self.reftab[id(datum)] = ref
	44	return ref
	45	return None
	46
	47	@staticmethod
	48	def encint(x):
	49	ret = bytearray()
	50	if x >= 0:
	51	b = x & 0x7f
	52	x >>= 7
	53	while (x > 0) or (b & 0x40) != 0:
	54	ret.append(0x80 \| b)
	55	b = x & 0x7f
	56	x >>= 7
	57	ret.append(b)
	58	elif x < 0:
	59	b = x & 0x7f
	60	x >>= 7
	61	while x < -1 or (b & 0x40) == 0:
	62	ret.append(0x80 \| b)
	63	b = x & 0x7f
	64	x >>= 7
	65	ret.append(b)
	66	return ret
	67
	68	@staticmethod
	69	def writestr(dst, text):
	70	dst.write(text.encode("utf-8"))
	71	dst.write(b'\0')
	72
	73	def dumpseq(self, dst, seq):
	74	for v in seq:
	75	self.dump(dst, v)
	76	dst.write(self.enctag(T_END, 0))
	77
	78	def dumpmap(self, dst, val):
	79	for k, v in val.items():
	80	self.dump(dst, k)
	81	self.dump(dst, v)
	82	dst.write(self.enctag(T_END, 0))
	83
	84	def dump(self, dst, datum):
	85	ref = self.reftab.get(id(datum))
	86	if ref is not None:
	87	dst.write(self.enctag(T_INT, INT_REF))
	88	dst.write(self.encint(ref))
	89	return
90	if datum == None:
91	self.writetag(dst, T_NIL, 0, None)
92	elif datum == False:
93	self.writetag(dst, T_NIL, NIL_FALSE, None)
94	elif datum == True:
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)))
105	dst.write(datum)
106	elif isinstance(datum, data.symbol):
107	if datum.ns == "":
108	self.writetag(dst, T_STR, STR_SYM, datum)
109	self.writestr(dst, datum.name)
110	else:
111	nsref = self.nstab.get(datum.ns)
112	if nsref is None:
113	nsref = self.writetag(dst, T_SYM, 0, datum)
114	dst.write(b'\0')
115	self.writestr(dst, datum.ns)
116	self.writestr(dst, datum.name)
117	if nsref is not None:
118	self.nstab[datum.ns] = nsref
119	else:
120	self.writetag(dst, T_SYM, 0, datum)
121	dst.write(b'\x01')
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__)
137	else:
138	raise ValueError("unsupported object type: " + repr(datum))
139
140	def dump(dst, datum):
141	encoder().dump(dst, datum)
142	return dst
143
e8a122ff FT	144	class fmterror(Exception):
	145	pass
	146
	147	class eoferror(fmterror):
	148	def __init__(self):
	149	super().__init__("unexpected end-of-data")
	150
	151	class referror(fmterror):
	152	def __init__(self):
	153	super().__init__("bad backref")
	154
	155	class namedtype(type):
	156	pass
	157
	158	class decoder(object):
	159	def __init__(self):
	160	self.reftab = []
	161	self.namedtypes = {}
	162
	163	@staticmethod
	164	def byte(fp):
	165	b = fp.read(1)
	166	if b == b"":
	167	raise eoferror()
	168	return b[0]
	169
	170	@staticmethod
	171	def loadint(fp):
	172	ret = 0
	173	p = 0
	174	while True:
	175	b = decoder.byte(fp)
	176	ret += (b & 0x7f) << p
	177	p += 7
	178	if (b & 0x80) == 0:
	179	break
	180	if (b & 0x40) != 0:
	181	ret = ret - (1 << p)
	182	return ret
	183
	184	@staticmethod
	185	def loadstr(fp):
	186	buf = bytearray()
	187	while True:
	188	b = decoder.byte(fp)
	189	if b == 0:
	190	break
	191	buf.append(b)
	192	return buf.decode("utf-8")
	193
	194	def loadsym(self, fp):
	195	h = self.byte(fp)
	196	if h & 0x1:
	197	nsref = self.loadint(fp)
	198	if not 0 <= nsref < len(self.reftab):
	199	raise fmterror("illegal namespace ref: " + str(nsref))
	200	nssym = self.reftab[nsref]
	201	if not isinstance(nssym, data.symbol):
	202	raise fmterror("illegal namespace ref: " + str(nsref))
	203	ns = nssym.ns
	204	else:
	205	ns = self.loadstr(fp)
	206	nm = self.loadstr(fp)
	207	ret = data.symbol.get(ns, nm)
208	return ret
209
210	def loadlist(self, fp, buf):
211	while True:
212	tag = self.byte(fp)
213	if tag == T_END:
214	return buf
215	buf.append(self.loadtagged(fp, tag))
216
217	def loadmap(self, fp, buf):
218	while True:
219	tag = self.byte(fp)
220	if tag == T_END:
221	return buf
222	key = self.loadtagged(fp, tag)
223	tag = self.byte(fp)
224	if tag == T_END:
225	return buf
226	buf[key] = self.loadtagged(fp, tag)
227
228	def loadobj(self, fp, ref=False):
229	if ref:
230	refid = len(self.reftab)
231	self.reftab.append(None)
232	nm = self.load(fp)
233	typ = self.namedtypes.get(nm)
234	if typ is None:
235	typ = self.namedtypes[nm] = namedtype(str(nm), (data.obj, object), {})
236	typ.typename = nm
237	ret = typ()
238	if ref:
239	self.reftab[refid] = ret
240	# st = fp.tell()
241	# print(">", nm, hex(st))
242	ret.__dict__.update(self.loadmap(fp, {}))
243	# print("<", nm, hex(fp.tell()), hex(st))
244	return ret
245
246	def addref(self, obj):
247	self.reftab.append(obj)
248	return obj
249
250	def loadtagged(self, fp, tag):
251	pri, sec = (tag & 0x7), (tag & 0xf8) >> 3
252	if pri == T_END:
253	raise fmterror("unexpected end-tag")
254	elif pri == T_INT:
255	if sec == INT_REF:
256	idx = self.loadint(fp)
257	if not 0 <= idx < len(self.reftab):
258	raise referror()
259	# print(idx, self.reftab[idx], hex(fp.tell()))
260	return self.reftab[idx]
261	return self.addref(self.loadint(fp))
262	elif pri == T_STR:
263	ret = self.addref(self.loadstr(fp))
264	if sec == STR_SYM:
265	return data.symbol.get("", ret)
266	return ret
267	elif pri == T_BIT:
268	ln = self.loadint(fp)
269	ret = self.addref(fp.read(ln))
270	if len(ret) < ln:
271	raise eoferror()
272	return ret
273	elif pri == T_NIL:
274	if sec == NIL_TRUE:
275	return self.addref(True)
276	elif sec == NIL_FALSE:
277	return self.addref(False)
278	return self.addref(None)
279	elif pri == T_SYM:
280	return self.addref(self.loadsym(fp))
281	elif pri == T_CON:
282	if sec == CON_MAP:
283	return self.loadmap(fp, self.addref({}))
284	elif sec == CON_OBJ:
285	return self.loadobj(fp, ref=True)
286	else:
287	return self.loadlist(fp, self.addref([]))
288	else:
289	raise fmterror("unknown primary: " + str(pri))
290
291	def load(self, fp):
292	tag = self.byte(fp)
293	return self.loadtagged(fp, tag)
294
295	def load(fp):
296	decoder().load(fp)