127 lines
4.4 KiB
Python
127 lines
4.4 KiB
Python
from typing import AsyncIterable, Generic, Iterable, TypeVar
|
|
|
|
from rainbowadn.core import *
|
|
from rainbowadn.nullability import *
|
|
|
|
__all__ = ('Stack', 'StackFactory',)
|
|
|
|
ElementType = TypeVar('ElementType')
|
|
|
|
|
|
class Stack(RecursiveMentionable, Generic[ElementType]):
|
|
def __factory__(self) -> RainbowFactory['Stack[ElementType]']:
|
|
return self.factory(self.element.factory)
|
|
|
|
@classmethod
|
|
def factory(cls, factory: RainbowFactory[ElementType]) -> RainbowFactory['Stack[ElementType]']:
|
|
assert isinstance(factory, RainbowFactory)
|
|
return StackFactory(factory)
|
|
|
|
def __init__(self, previous: NullableReference['Stack[ElementType]'], element: HashPoint[ElementType]):
|
|
assert isinstance(previous, NullableReference)
|
|
assert isinstance(element, HashPoint)
|
|
self.previous = previous
|
|
self.element = element
|
|
|
|
def points(self) -> Iterable[HashPoint]:
|
|
return [*self.previous.points(), self.element]
|
|
|
|
def __bytes__(self):
|
|
return bytes(self.previous) + bytes(self.element)
|
|
|
|
async def str(self, tab: int) -> str:
|
|
assert isinstance(tab, int)
|
|
previous_str, element_str = await gather(
|
|
self.previous.str(tab),
|
|
hash_point_format(self.element, tab),
|
|
)
|
|
assert isinstance(previous_str, str)
|
|
assert isinstance(element_str, str)
|
|
return f'{previous_str}' \
|
|
f'{tabulate(tab)}{element_str}'
|
|
|
|
@classmethod
|
|
def of(
|
|
cls,
|
|
factory: RainbowFactory[ElementType],
|
|
elements: Iterable[HashPoint[ElementType]]
|
|
) -> NullableReference[
|
|
'Stack[ElementType]'
|
|
]:
|
|
assert isinstance(factory, RainbowFactory)
|
|
reference: NullableReference[Stack[ElementType]] = NullableReference(Null(), cls.factory(factory))
|
|
assert isinstance(reference, NullableReference)
|
|
for element in elements:
|
|
assert isinstance(element, HashPoint)
|
|
reference = NullableReference.off(cls(reference, element))
|
|
assert isinstance(reference, NullableReference)
|
|
return reference
|
|
|
|
@classmethod
|
|
def off(
|
|
cls,
|
|
factory: RainbowFactory[ElementType],
|
|
elements: Iterable[ElementType]
|
|
) -> NullableReference[
|
|
'Stack[ElementType]'
|
|
]:
|
|
assert isinstance(factory, RainbowFactory)
|
|
return cls.of(factory, map(HashPoint.of, elements))
|
|
|
|
@classmethod
|
|
async def iter(
|
|
cls,
|
|
reference: NullableReference['Stack[ElementType]']
|
|
) -> AsyncIterable[HashPoint[ElementType]]:
|
|
assert isinstance(reference, NullableReference)
|
|
async for stack in cls.iter_stacks(reference):
|
|
yield stack.element
|
|
|
|
@classmethod
|
|
async def iter_stacks(
|
|
cls,
|
|
reference: NullableReference['Stack[ElementType]']
|
|
) -> AsyncIterable['Stack[ElementType]']:
|
|
assert isinstance(reference, NullableReference)
|
|
if reference.null():
|
|
pass
|
|
else:
|
|
stack: Stack[ElementType] = await reference.resolve()
|
|
assert isinstance(stack, Stack)
|
|
yield stack
|
|
async for stack in cls.iter_stacks(stack.previous):
|
|
yield stack
|
|
|
|
@classmethod
|
|
async def list(
|
|
cls,
|
|
reference: NullableReference['Stack[ElementType]']
|
|
) -> list[ElementType]:
|
|
assert isinstance(reference, NullableReference)
|
|
return list(
|
|
await gather(
|
|
*[element.resolve() async for element in cls.iter(reference)]
|
|
)
|
|
)
|
|
|
|
@classmethod
|
|
def unit(cls, element: HashPoint[ElementType]) -> 'Stack[ElementType]':
|
|
return Stack(NullableReference(Null(), StackFactory(element.factory)), element)
|
|
|
|
|
|
class StackFactory(RainbowFactory[Stack[ElementType]], Generic[ElementType]):
|
|
def __init__(self, factory: RainbowFactory[ElementType]):
|
|
assert isinstance(factory, RainbowFactory)
|
|
self.factory = factory
|
|
|
|
def from_bytes(self, source: bytes, resolver: HashResolver) -> Stack[ElementType]:
|
|
assert isinstance(source, bytes)
|
|
assert isinstance(resolver, HashResolver)
|
|
return Stack(
|
|
NullableReference.f(self).from_bytes(source[:HashPoint.HASH_LENGTH], resolver),
|
|
ResolverOrigin(self.factory, source[HashPoint.HASH_LENGTH:], resolver).hash_point()
|
|
)
|
|
|
|
def loose(self) -> RainbowFactory[Stack[ElementType]]:
|
|
return self
|