(value, typ=None)
| 78 | raise Exception("Expecting bytes") |
| 79 | |
| 80 | def serialize_value(value, typ=None): |
| 81 | if typ is None: |
| 82 | typ = infer_type(value) |
| 83 | if isinstance(typ, str) and typ[:4] == 'uint': |
| 84 | length = int(typ[4:]) |
| 85 | assert length in (8, 16, 32, 64, 128, 256) |
| 86 | return value.to_bytes(length // 8, 'little') |
| 87 | elif typ == 'bool': |
| 88 | assert value in (True, False) |
| 89 | return b'\x01' if value is True else b'\x00' |
| 90 | elif (isinstance(typ, list) and len(typ) == 1) or typ == 'bytes': |
| 91 | serialized_bytes = coerce_to_bytes(value) if typ == 'bytes' else b''.join([serialize_value(element, typ[0]) for element in value]) |
| 92 | assert len(serialized_bytes) < 2**(8 * BYTES_PER_LENGTH_PREFIX) |
| 93 | serialized_length = len(serialized_bytes).to_bytes(BYTES_PER_LENGTH_PREFIX, 'little') |
| 94 | return serialized_length + serialized_bytes |
| 95 | elif isinstance(typ, list) and len(typ) == 2: |
| 96 | assert len(value) == typ[1] |
| 97 | return b''.join([serialize_value(element, typ[0]) for element in value]) |
| 98 | elif isinstance(typ, str) and len(typ) > 5 and typ[:5] == 'bytes': |
| 99 | assert len(value) == int(typ[5:]), (value, int(typ[5:])) |
| 100 | return coerce_to_bytes(value) |
| 101 | elif hasattr(typ, 'fields'): |
| 102 | serialized_bytes = b''.join([serialize_value(getattr(value, field), subtype) for field, subtype in typ.fields.items()]) |
| 103 | if is_constant_sized(typ): |
| 104 | return serialized_bytes |
| 105 | else: |
| 106 | assert len(serialized_bytes) < 2**(8 * BYTES_PER_LENGTH_PREFIX) |
| 107 | serialized_length = len(serialized_bytes).to_bytes(BYTES_PER_LENGTH_PREFIX, 'little') |
| 108 | return serialized_length + serialized_bytes |
| 109 | else: |
| 110 | print(value, typ) |
| 111 | raise Exception("Type not recognized") |
| 112 | |
| 113 | def chunkify(bytez): |
| 114 | bytez += b'\x00' * (-len(bytez) % BYTES_PER_CHUNK) |
no test coverage detected