|
|
@@ -64,111 +64,108 @@ class Buffer(object):
|
|
|
if pfx == '_':
|
|
|
continue
|
|
|
val = msg[r]
|
|
|
- match pfx:
|
|
|
- case 'b':
|
|
|
- self.write_id(self.FT_BOOL | (
|
|
|
- self.FS_SHORT if val else 0), r)
|
|
|
- case 'u':
|
|
|
- val = val if val is not None else -1
|
|
|
- if 0 <= val < 256:
|
|
|
- self.write_id(self.FT_U32 | self.FS_SHORT, r)
|
|
|
- self.arr[self.pos] = val
|
|
|
- self.pos += 1
|
|
|
- else:
|
|
|
- self.write_id(self.FT_U32, r)
|
|
|
- self.write_32(val)
|
|
|
- case 'q':
|
|
|
- self.write_id(self.FT_U64, r)
|
|
|
- self.write_64(val)
|
|
|
- case 'a':
|
|
|
- self.write_id(self.FT_ADDR6, r)
|
|
|
- for i in range(16):
|
|
|
+ if pfx == 'b':
|
|
|
+ self.write_id(self.FT_BOOL | (
|
|
|
+ self.FS_SHORT if val else 0), r)
|
|
|
+ elif pfx == 'u':
|
|
|
+ val = val if val is not None else -1
|
|
|
+ if 0 <= val < 256:
|
|
|
+ self.write_id(self.FT_U32 | self.FS_SHORT, r)
|
|
|
+ self.arr[self.pos] = val
|
|
|
+ self.pos += 1
|
|
|
+ else:
|
|
|
+ self.write_id(self.FT_U32, r)
|
|
|
+ self.write_32(val)
|
|
|
+ elif pfx == 'q':
|
|
|
+ self.write_id(self.FT_U64, r)
|
|
|
+ self.write_64(val)
|
|
|
+ elif pfx == 'a':
|
|
|
+ self.write_id(self.FT_ADDR6, r)
|
|
|
+ for i in range(16):
|
|
|
+ self.arr[self.pos] = val[i]
|
|
|
+ self.pos += 1
|
|
|
+ elif pfx == 's':
|
|
|
+ if len(val) < 256:
|
|
|
+ self.write_id(self.FT_STRING | self.FS_SHORT, r)
|
|
|
+ self.arr[self.pos] = len(val)
|
|
|
+ self.pos += 1
|
|
|
+ else:
|
|
|
+ self.write_id(self.FT_STRING, r)
|
|
|
+ self.write_16(len(val))
|
|
|
+ for i in range(len(val)):
|
|
|
+ self.arr[self.pos] = ord(val[i])
|
|
|
+ self.pos += 1
|
|
|
+ elif pfx == 'r':
|
|
|
+ if len(val) < 256:
|
|
|
+ self.write_id(self.FT_RAW | self.FS_SHORT, r)
|
|
|
+ self.arr[self.pos] = len(val)
|
|
|
+ self.pos += 1
|
|
|
+ else:
|
|
|
+ self.write_id(self.FT_RAW, r)
|
|
|
+ self.write_16(len(val))
|
|
|
+ for i in range(len(val)):
|
|
|
self.arr[self.pos] = val[i]
|
|
|
self.pos += 1
|
|
|
- case 's':
|
|
|
- if len(val) < 256:
|
|
|
- self.write_id(self.FT_STRING | self.FS_SHORT, r)
|
|
|
- self.arr[self.pos] = len(val)
|
|
|
- self.pos += 1
|
|
|
- else:
|
|
|
- self.write_id(self.FT_STRING, r)
|
|
|
- self.write_16(len(val))
|
|
|
- for i in range(len(val)):
|
|
|
- self.arr[self.pos] = ord(val[i])
|
|
|
+ elif pfx == 'm':
|
|
|
+ x = self.msg2buffer(val)
|
|
|
+ if len(x) < 256:
|
|
|
+ self.write_id(self.FT_MESSAGE | self.FS_SHORT, r)
|
|
|
+ self.arr[self.pos] = len(x)
|
|
|
+ self.pos += 1
|
|
|
+ else:
|
|
|
+ self.write_id(self.FT_MESSAGE, r)
|
|
|
+ self.write_16(len(x))
|
|
|
+ for item in x[::-1]:
|
|
|
+ self.arr[self.pos] = item
|
|
|
+ self.pos += len(x)
|
|
|
+ elif pfx == 'B':
|
|
|
+ self.write_id(self.FT_BOOL_ARRAY, r)
|
|
|
+ self.write_16(len(val))
|
|
|
+ for i in range(len(val)):
|
|
|
+ self.arr[self.pos] = val[i]
|
|
|
+ self.pos += 1
|
|
|
+ elif pfx == 'U':
|
|
|
+ self.write_id(self.FT_U32_ARRAY, r)
|
|
|
+ self.write_16(len(val))
|
|
|
+ for i in range(len(val)):
|
|
|
+ self.write_32(val[i])
|
|
|
+ elif pfx == 'Q':
|
|
|
+ self.write_id(self.FT_U64_ARRAY, r)
|
|
|
+ self.write_16(len(val))
|
|
|
+ for i in range(len(val)):
|
|
|
+ self.write_64(val[i])
|
|
|
+ elif pfx == 'A':
|
|
|
+ self.write_id(self.FT_ADDR6_ARRAY, r)
|
|
|
+ self.write_16(len(val))
|
|
|
+ for i in range(len(val)):
|
|
|
+ for k in range(16):
|
|
|
+ self.arr[self.pos] = val[i][k]
|
|
|
self.pos += 1
|
|
|
- case 'r':
|
|
|
- if len(val) < 256:
|
|
|
- self.write_id(self.FT_RAW | self.FS_SHORT, r)
|
|
|
- self.arr[self.pos] = len(val)
|
|
|
+ elif pfx == 'S':
|
|
|
+ self.write_id(self.FT_STRING_ARRAY, r)
|
|
|
+ self.write_16(len(val))
|
|
|
+ for i in range(len(val)):
|
|
|
+ self.write_16(len(val[i]))
|
|
|
+ for k in range(len(val[i])):
|
|
|
+ self.arr[self.pos] = ord(val[i][k])
|
|
|
self.pos += 1
|
|
|
- else:
|
|
|
- self.write_id(self.FT_RAW, r)
|
|
|
- self.write_16(len(val))
|
|
|
- for i in range(len(val)):
|
|
|
- self.arr[self.pos] = val[i]
|
|
|
- self.pos += 1
|
|
|
- case 'm':
|
|
|
- x = self.msg2buffer(val)
|
|
|
- if len(x) < 256:
|
|
|
- self.write_id(self.FT_MESSAGE | self.FS_SHORT, r)
|
|
|
- self.arr[self.pos] = len(x)
|
|
|
+ elif pfx == 'R':
|
|
|
+ self.write_id(self.FT_RAW_ARRAY, r)
|
|
|
+ self.write_16(len(val))
|
|
|
+ for i in range(len(val)):
|
|
|
+ self.write_16(len(val[i]))
|
|
|
+ for k in range(len(val[i])):
|
|
|
+ self.arr[self.pos] = val[i][k]
|
|
|
self.pos += 1
|
|
|
- else:
|
|
|
- self.write_id(self.FT_MESSAGE, r)
|
|
|
- self.write_16(len(x))
|
|
|
+ elif pfx == 'M':
|
|
|
+ self.write_id(self.FT_MESSAGE_ARRAY, r)
|
|
|
+ self.write_16(len(val))
|
|
|
+ for i in range(len(val)):
|
|
|
+ x = self.msg2buffer(val[i])
|
|
|
+ self.write_16(len(x))
|
|
|
for item in x[::-1]:
|
|
|
self.arr[self.pos] = item
|
|
|
self.pos += len(x)
|
|
|
- case 'B':
|
|
|
- self.write_id(self.FT_BOOL_ARRAY, r)
|
|
|
- self.write_16(len(val))
|
|
|
- for i in range(len(val)):
|
|
|
- self.arr[self.pos] = val[i]
|
|
|
- self.pos += 1
|
|
|
- case 'U':
|
|
|
- self.write_id(self.FT_U32_ARRAY, r)
|
|
|
- self.write_16(len(val))
|
|
|
- for i in range(len(val)):
|
|
|
- self.write_32(val[i])
|
|
|
- case 'Q':
|
|
|
- self.write_id(self.FT_U64_ARRAY, r)
|
|
|
- self.write_16(len(val))
|
|
|
- for i in range(len(val)):
|
|
|
- self.write_64(val[i])
|
|
|
- case 'A':
|
|
|
- self.write_id(self.FT_ADDR6_ARRAY, r)
|
|
|
- self.write_16(len(val))
|
|
|
- for i in range(len(val)):
|
|
|
- for k in range(16):
|
|
|
- self.arr[self.pos] = val[i][k]
|
|
|
- self.pos += 1
|
|
|
- case 'S':
|
|
|
- self.write_id(self.FT_STRING_ARRAY, r)
|
|
|
- self.write_16(len(val))
|
|
|
- for i in range(len(val)):
|
|
|
- self.write_16(len(val[i]))
|
|
|
- for k in range(len(val[i])):
|
|
|
- self.arr[self.pos] = ord(val[i][k])
|
|
|
- self.pos += 1
|
|
|
- case 'R':
|
|
|
- self.write_id(self.FT_RAW_ARRAY, r)
|
|
|
- self.write_16(len(val))
|
|
|
- for i in range(len(val)):
|
|
|
- self.write_16(len(val[i]))
|
|
|
- for k in range(len(val[i])):
|
|
|
- self.arr[self.pos] = val[i][k]
|
|
|
- self.pos += 1
|
|
|
- case 'M':
|
|
|
- self.write_id(self.FT_MESSAGE_ARRAY, r)
|
|
|
- self.write_16(len(val))
|
|
|
- for i in range(len(val)):
|
|
|
- x = self.msg2buffer(val[i])
|
|
|
- self.write_16(len(x))
|
|
|
- for item in x[::-1]:
|
|
|
- self.arr[self.pos] = item
|
|
|
- self.pos += len(x)
|
|
|
- case _:
|
|
|
- return None
|
|
|
return self.arr[:self.pos]
|
|
|
|
|
|
def buffer2msg(self, arr, offset: int = 0):
|
|
|
@@ -177,116 +174,115 @@ class Buffer(object):
|
|
|
return ret
|
|
|
while self.pos < len(self.arr):
|
|
|
_id = self.read_32()
|
|
|
- match _id & self.MASK_FTYPE:
|
|
|
- case self.FT_BOOL:
|
|
|
- ret['b' + self.idnum2hex(_id)] = 1 if (_id & self.FS_SHORT) else 0
|
|
|
- case self.FT_U32:
|
|
|
- if _id & self.FS_SHORT:
|
|
|
- ret['u' + self.idnum2hex(_id)] = self.arr[self.pos]
|
|
|
- self.pos += 1
|
|
|
- else:
|
|
|
- ret['u' + self.idnum2hex(_id)
|
|
|
- ] = self.int2num(self.read_32())
|
|
|
- case self.FT_U64:
|
|
|
- ret['q' + self.idnum2hex(_id)] = self.read_64()
|
|
|
- case self.FT_ADDR6:
|
|
|
- a = []
|
|
|
- for i in range(16):
|
|
|
- a[i] = self.arr[self.pos]
|
|
|
- self.pos += 1
|
|
|
- ret['a' + self.idnum2hex(_id)] = a
|
|
|
- case self.FT_STRING:
|
|
|
- length = self.arr[self.pos]
|
|
|
+ if _id & self.MASK_FTYPE == self.FT_BOOL:
|
|
|
+ ret['b' + self.idnum2hex(_id)] = 1 if (_id & self.FS_SHORT) else 0
|
|
|
+ elif _id & self.MASK_FTYPE == self.FT_U32:
|
|
|
+ if _id & self.FS_SHORT:
|
|
|
+ ret['u' + self.idnum2hex(_id)] = self.arr[self.pos]
|
|
|
self.pos += 1
|
|
|
- if not (_id & self.FS_SHORT):
|
|
|
- length |= self.arr[self.pos] << 8
|
|
|
- self.pos += 1
|
|
|
- s = ''
|
|
|
- for i in range(length):
|
|
|
- s = s + chr(self.arr[self.pos])
|
|
|
- self.pos += 1
|
|
|
- ret['s' + self.idnum2hex(_id)] = s
|
|
|
- case self.FT_RAW:
|
|
|
- length = self.arr[self.pos]
|
|
|
+ else:
|
|
|
+ ret['u' + self.idnum2hex(_id)
|
|
|
+ ] = self.int2num(self.read_32())
|
|
|
+ elif _id & self.MASK_FTYPE == self.FT_U64:
|
|
|
+ ret['q' + self.idnum2hex(_id)] = self.read_64()
|
|
|
+ elif _id & self.MASK_FTYPE == self.FT_ADDR6:
|
|
|
+ a = []
|
|
|
+ for i in range(16):
|
|
|
+ a[i] = self.arr[self.pos]
|
|
|
+ self.pos += 1
|
|
|
+ ret['a' + self.idnum2hex(_id)] = a
|
|
|
+ elif _id & self.MASK_FTYPE == self.FT_STRING:
|
|
|
+ length = self.arr[self.pos]
|
|
|
+ self.pos += 1
|
|
|
+ if not (_id & self.FS_SHORT):
|
|
|
+ length |= self.arr[self.pos] << 8
|
|
|
+ self.pos += 1
|
|
|
+ s = ''
|
|
|
+ for i in range(length):
|
|
|
+ s = s + chr(self.arr[self.pos])
|
|
|
+ self.pos += 1
|
|
|
+ ret['s' + self.idnum2hex(_id)] = s
|
|
|
+ elif _id & self.MASK_FTYPE == self.FT_RAW:
|
|
|
+ length = self.arr[self.pos]
|
|
|
+ self.pos += 1
|
|
|
+ if not (_id & self.FS_SHORT):
|
|
|
+ length |= self.arr[self.pos] << 8
|
|
|
self.pos += 1
|
|
|
- if not (_id & self.FS_SHORT):
|
|
|
- length |= self.arr[self.pos] << 8
|
|
|
+ a = [0] * length
|
|
|
+ for i in range(length):
|
|
|
+ a[i] = self.arr[self.pos]
|
|
|
+ self.pos += 1
|
|
|
+ ret['r' + self.idnum2hex(_id)] = a
|
|
|
+ elif _id & self.MASK_FTYPE == self.FT_MESSAGE:
|
|
|
+ length = self.arr[self.pos]
|
|
|
+ self.pos += 1
|
|
|
+ if not (_id & self.FS_SHORT):
|
|
|
+ length |= self.arr[self.pos] << 8
|
|
|
+ self.pos += 1
|
|
|
+ ret['m' + self.idnum2hex(_id)] = self.buffer2msg(
|
|
|
+ self.arr[offset + self.pos:offset + self.pos + length])
|
|
|
+ self.pos += length
|
|
|
+ offset += self.pos
|
|
|
+ elif _id & self.MASK_FTYPE == self.FT_BOOL_ARRAY:
|
|
|
+ length = self.read_16()
|
|
|
+ a = [0] * length
|
|
|
+ for i in range(length):
|
|
|
+ a[i] = not (not self.arr[self.pos])
|
|
|
+ self.pos += 1
|
|
|
+ ret['B' + self.idnum2hex(_id)] = a
|
|
|
+ elif _id & self.MASK_FTYPE == self.FT_U32_ARRAY:
|
|
|
+ length = self.read_16()
|
|
|
+ a = [0] * length
|
|
|
+ for i in range(length):
|
|
|
+ a[i] = self.int2num(self.read_32())
|
|
|
+ ret['U' + self.idnum2hex(_id)] = a
|
|
|
+ elif _id & self.MASK_FTYPE == self.FT_U64_ARRAY:
|
|
|
+ length = self.read_16()
|
|
|
+ a = [0] * length
|
|
|
+ for i in range(length):
|
|
|
+ a[i] = self.read_64()
|
|
|
+ ret['Q' + self.idnum2hex(_id)] = a
|
|
|
+ elif _id & self.MASK_FTYPE == self.FT_ADDR6_ARRAY:
|
|
|
+ length = self.read_16()
|
|
|
+ a = [0] * length
|
|
|
+ for i in range(length):
|
|
|
+ x = [0] * 16
|
|
|
+ for k in range(16):
|
|
|
+ x[k] = self.arr[self.pos]
|
|
|
self.pos += 1
|
|
|
- a = [0] * length
|
|
|
- for i in range(length):
|
|
|
- a[i] = self.arr[self.pos]
|
|
|
+ a[i] = x
|
|
|
+ ret['A' + self.idnum2hex(_id)] = a
|
|
|
+ elif _id & self.MASK_FTYPE == self.FT_STRING_ARRAY:
|
|
|
+ length = self.read_16()
|
|
|
+ a = [0] * length
|
|
|
+ for i in range(length):
|
|
|
+ x = ''
|
|
|
+ x_len = self.read_16()
|
|
|
+ for k in range(x_len):
|
|
|
+ x = x + chr(self.arr[self.pos])
|
|
|
self.pos += 1
|
|
|
- ret['r' + self.idnum2hex(_id)] = a
|
|
|
- case self.FT_MESSAGE:
|
|
|
- length = self.arr[self.pos]
|
|
|
- self.pos += 1
|
|
|
- if not (_id & self.FS_SHORT):
|
|
|
- length |= self.arr[self.pos] << 8
|
|
|
+ a[i] = x
|
|
|
+ ret['S' + self.idnum2hex(_id)] = a
|
|
|
+ elif _id & self.MASK_FTYPE == self.FT_RAW_ARRAY:
|
|
|
+ length = self.read_16()
|
|
|
+ a = [0] * length
|
|
|
+ for i in range(length):
|
|
|
+ x_len = self.read_16()
|
|
|
+ x = [0] * x_len
|
|
|
+ for k in range(x_len):
|
|
|
+ x[k] = self.arr[self.pos]
|
|
|
self.pos += 1
|
|
|
- ret['m' + self.idnum2hex(_id)] = self.buffer2msg(
|
|
|
- self.arr[offset + self.pos:offset + self.pos + length])
|
|
|
- self.pos += length
|
|
|
+ a[i] = x
|
|
|
+ ret['R' + self.idnum2hex(_id)] = a
|
|
|
+ elif _id & self.MASK_FTYPE == self.FT_MESSAGE_ARRAY:
|
|
|
+ length = self.read_16()
|
|
|
+ a = [0] * length
|
|
|
+ for i in range(length):
|
|
|
+ x_len = self.read_16()
|
|
|
+ a[i] = self.buffer2msg(self.arr[offset + self.pos:offset + self.pos + x_len], offset + self.pos)
|
|
|
+ self.pos += x_len
|
|
|
offset += self.pos
|
|
|
- case self.FT_BOOL_ARRAY:
|
|
|
- length = self.read_16()
|
|
|
- a = [0] * length
|
|
|
- for i in range(length):
|
|
|
- a[i] = not (not self.arr[self.pos])
|
|
|
- self.pos += 1
|
|
|
- ret['B' + self.idnum2hex(_id)] = a
|
|
|
- case self.FT_U32_ARRAY:
|
|
|
- length = self.read_16()
|
|
|
- a = [0] * length
|
|
|
- for i in range(length):
|
|
|
- a[i] = self.int2num(self.read_32())
|
|
|
- ret['U' + self.idnum2hex(_id)] = a
|
|
|
- case self.FT_U64_ARRAY:
|
|
|
- length = self.read_16()
|
|
|
- a = [0] * length
|
|
|
- for i in range(length):
|
|
|
- a[i] = self.read_64()
|
|
|
- ret['Q' + self.idnum2hex(_id)] = a
|
|
|
- case self.FT_ADDR6_ARRAY:
|
|
|
- length = self.read_16()
|
|
|
- a = [0] * length
|
|
|
- for i in range(length):
|
|
|
- x = [0] * 16
|
|
|
- for k in range(16):
|
|
|
- x[k] = self.arr[self.pos]
|
|
|
- self.pos += 1
|
|
|
- a[i] = x
|
|
|
- ret['A' + self.idnum2hex(_id)] = a
|
|
|
- case self.FT_STRING_ARRAY:
|
|
|
- length = self.read_16()
|
|
|
- a = [0] * length
|
|
|
- for i in range(length):
|
|
|
- x = ''
|
|
|
- x_len = self.read_16()
|
|
|
- for k in range(x_len):
|
|
|
- x = x + chr(self.arr[self.pos])
|
|
|
- self.pos += 1
|
|
|
- a[i] = x
|
|
|
- ret['S' + self.idnum2hex(_id)] = a
|
|
|
- case self.FT_RAW_ARRAY:
|
|
|
- length = self.read_16()
|
|
|
- a = [0] * length
|
|
|
- for i in range(length):
|
|
|
- x_len = self.read_16()
|
|
|
- x = [0] * x_len
|
|
|
- for k in range(x_len):
|
|
|
- x[k] = self.arr[self.pos]
|
|
|
- self.pos += 1
|
|
|
- a[i] = x
|
|
|
- ret['R' + self.idnum2hex(_id)] = a
|
|
|
- case self.FT_MESSAGE_ARRAY:
|
|
|
- length = self.read_16()
|
|
|
- a = [0] * length
|
|
|
- for i in range(length):
|
|
|
- x_len = self.read_16()
|
|
|
- a[i] = self.buffer2msg(self.arr[offset + self.pos:offset + self.pos + x_len], offset + self.pos)
|
|
|
- self.pos += x_len
|
|
|
- offset += self.pos
|
|
|
- ret['M' + self.idnum2hex(_id)] = a
|
|
|
+ ret['M' + self.idnum2hex(_id)] = a
|
|
|
return ret
|
|
|
|
|
|
def buffer2msgs(self, arr, offset: int = 0):
|
