58 lines
2.3 KiB
Python
58 lines
2.3 KiB
Python
from typing import Generic, Iterable, TypeVar
|
|
|
|
from rainbowadn.data.collection.keyed import Keyed
|
|
from rainbowadn.hashing.hash_point_format import hash_point_format, tabulate
|
|
from rainbowadn.hashing.hashpoint import HashPoint
|
|
from rainbowadn.hashing.hashresolver import HashResolver
|
|
from rainbowadn.hashing.rainbow_factory import RainbowFactory
|
|
from rainbowadn.hashing.resolverorigin import ResolverOrigin
|
|
|
|
__all__ = ('KeyMetadata', 'KeyMetadataFactory',)
|
|
|
|
ActiveKeyType = TypeVar('ActiveKeyType')
|
|
MetaDataType = TypeVar('MetaDataType')
|
|
|
|
|
|
class KeyMetadata(Keyed[ActiveKeyType], Generic[ActiveKeyType, MetaDataType]):
|
|
def __init__(self, key: HashPoint[ActiveKeyType], metadata: HashPoint[MetaDataType]):
|
|
assert isinstance(key, HashPoint)
|
|
assert isinstance(metadata, HashPoint)
|
|
super().__init__(key)
|
|
self.metadata = metadata
|
|
|
|
def points(self) -> Iterable[HashPoint]:
|
|
return [self.key, self.metadata]
|
|
|
|
def __bytes__(self):
|
|
return bytes(self.key) + bytes(self.metadata)
|
|
|
|
def __factory__(self) -> RainbowFactory['KeyMetadata[ActiveKeyType, MetaDataType]']:
|
|
return KeyMetadataFactory(self.key.factory, self.metadata.factory)
|
|
|
|
def str(self, tab: int) -> str:
|
|
assert isinstance(tab, int)
|
|
return f'{hash_point_format(self.key, tab)}' \
|
|
f'{tabulate(tab)}{hash_point_format(self.metadata, tab)}'
|
|
|
|
|
|
class KeyMetadataFactory(
|
|
RainbowFactory[KeyMetadata[ActiveKeyType, MetaDataType]],
|
|
Generic[ActiveKeyType, MetaDataType]
|
|
):
|
|
def __init__(self, key_factory: RainbowFactory[ActiveKeyType], metadata_factory: RainbowFactory[MetaDataType]):
|
|
assert isinstance(key_factory, RainbowFactory)
|
|
assert isinstance(metadata_factory, RainbowFactory)
|
|
self.key_factory = key_factory
|
|
self.metadata_factory = metadata_factory
|
|
|
|
def from_bytes(self, source: bytes, resolver: HashResolver) -> KeyMetadata[ActiveKeyType, MetaDataType]:
|
|
assert isinstance(source, bytes)
|
|
assert isinstance(resolver, HashResolver)
|
|
return KeyMetadata(
|
|
ResolverOrigin(self.key_factory, source[:HashPoint.HASH_LENGTH], resolver).hash_point(),
|
|
ResolverOrigin(self.metadata_factory, source[HashPoint.HASH_LENGTH:], resolver).hash_point(),
|
|
)
|
|
|
|
def loose(self) -> RainbowFactory[KeyMetadata[ActiveKeyType, MetaDataType]]:
|
|
return self
|