Skip to content

Instantly share code, notes, and snippets.

@bedekelly bedekelly/maps.py
Created Mar 19, 2018

Embed
What would you like to do?
First stab at an implementation of Record types
"""
A MapKeyValue is a name-value pair retrieved from a map.
The name is a string, and the value can be anything at all.
"""
MapKeyValue = namedtuple("MapKeyValue", "key value")
"""
A MapEntrySpec is a specification for an entry in a Map.
The key is a number, the name is a string exposed to the user,
and the value_type is either a BuiltinType or a UserType.
"""
MapEntrySpec = namedtuple("EntrySpec", "key name value_type")
class Mapping:
"""
A Mapping maps numeric keys onto full Entry types. It doesn't
hold the data itself – that's what a Map does! – but it provides
helper methods to extract the data.
Maps rely on Mappings internally to decode streams of data into
Python dictionaries.
"""
def __init__(self, *args):
self.entry_specs = set(args)
def read_key(self, key, bytestream):
"""
Read a value with `key` from the given bytestream.
Look up the value's type in our `entry_specs`, then
use that information to delegate to something that knows
how to read the value in question.
"""
for entry_spec in self.entry_specs:
if entry_spec.key == key:
break
else:
raise KeyError(f"No type information about key {key}!")
# Assume that the entry specification's type knows how to
# read a value from the bytestream.
value = entry_spec.value_type.read(bytestream)
return MapKeyValue(entry_spec.name, value)
class Map(BuiltinType):
"""
A Map is the equivalent of a Python dictionary, and the building
block for more complex data types.
Unlike a Python dictionary though, it takes a `mapping`, which
specifies exactly what keys it has ahead of time. This makes
it ideal for defining "Record" types, like an Employee who has
a name and age.
"""
def __init__(self, mapping):
self.mapping = mapping
def read(self, bytestream):
# Make sure our bytestream is single-use only!
bytestream = iter(bytestream)
map_data = {}
number_entries = UnsignedInt.read(bytestream)
for _ in range(number_entries):
key = UnsignedInt.read(bytestream)
name, value = self.mapping.read_key(key, bytestream)
map_data[name] = value
return map_data
def to_bytes(self, value):
# Given an entry's name, look up its data type.
specs_by_name = {spec.name: spec
for spec in self.mapping.entry_specs}
seen_keys = set()
# First, send the number of key-value pairs.
items = list(value.items())
yield from UnsignedInt.to_bytes(len(items))
# Next, send the key for each value, then the value itself.
for (name, inner_value) in value.items():
spec = specs_by_name[name]
yield from UnsignedInt.to_bytes(spec.key)
yield from spec.value_type.to_bytes(inner_value)
seen_keys.add(name)
if seen_keys ^ specs_by_name.keys():
raise ValueError(seen_keys ^ specs_by_name.keys())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.