transformers: rename Transform to Proc, add Preproc

This commit is contained in:
Shiz 2022-05-10 13:28:51 +02:00
parent b8cdfaf5dd
commit b737cb17b1
2 changed files with 76 additions and 7 deletions

View File

@ -19,12 +19,13 @@ from .types.num import (
from .types.str import Str, CStr, cstr, wcstr, utf8cstr, utf16cstr
from .types.seq import Arr, Tuple
from .types.struct import StructType, Struct, Union
from .types.transforms import Default, Transform, Mapped, Enum, Check, Fixed
from .types.transforms import Default, Preproc, Proc, Mapped, Enum, Check, Fixed
from .types.control import Switch, If
from .types.io import Sized, Terminated, Ref, AlignTo, AlignedTo, Lazy
del types
# Compatibility
Static = Calc = Implied
Transform = Proc
__all__ = [k for k in globals() if not k.startswith('_')]

View File

@ -1,9 +1,10 @@
import enum
import io
from typing import Optional as O, Generic as G, Union as U, TypeVar, Callable, Sequence, Mapping, Type as Ty
from ..core.base import Type, Context, PathElement
from ..core.io import Stream, Pos
from ..core.io import Stream, Pos, to_stream
from ..core.meta import Wrapper
from ..core import to_type
from ..core import to_type, dump
from .data import Data
@ -19,7 +20,74 @@ class Default(G[T], Wrapper[T]):
def default(self, context: Context) -> T:
return self._default
class Transform(G[T, V], Type[V]):
Preprocessor = U[Callable[[bytes], bytes], Callable[[bytes, Context], bytes]]
class Preproc(G[T], Wrapper[T]):
def __init__(self, child: Type[T], parse: Preprocessor, dump: O[Preprocessor] = None, sizeof: O[U[Callable[[O[T]], Pos], Callable[[O[T], Context], Pos]]] = None, context: bool = False, bits: bool = False, str: O[str] = None, repr: O[str] = None) -> None:
super().__init__(child)
self.on_parse = parse
self.on_dump = dump or self.on_parse
self.on_sizeof = sizeof
self.context = context
self.bits = bits
self.on_str = str or repr
self.on_repr = repr or str
def parse(self, context: Context, stream: Stream) -> T:
size = self.sizeof(context, None)
if size is None:
raise ValueError(f'{self.__class__.__name__} needs a sized child')
bits = context.get(self.bits)
if bits:
raw = stream.read(int(size * 8), bits=True)
else:
raw = stream.read(size)
ctx = context.get(self.context)
proc = self.on_parse(raw, context) if ctx else self.on_parse(raw)
return super().parse(context, to_stream(proc))
def dump(self, context: Context, stream: Stream, value: T) -> None:
raw_stream = io.BytesIO()
super().on_dump(context, raw_stream, value)
raw = raw_stream.getvalue()
ctx = context.get(self.context)
proc = self.on_dump(raw, context) if ctx else self.on_dump(raw)
size = self.sizeof(context, value)
bits = context.get(self.bits)
if bits:
stream.write(proc, bits=int(size * 8))
else:
stream.write(proc)
def sizeof(self, context: Context, value: O[T]) -> O[Pos]:
if value is not None:
raw_stream = io.BytesIO()
dump(context.peek(self.child), value, raw_stream)
value = raw_stream.getvalue()
ctx = context.peek(self.context)
if self.on_sizeof:
size = self.on_sizeof(value, context) if ctx else self.on_sizeof(value)
if size is None:
size = super().sizeof(context, value)
return size
def __str__(self) -> str:
if self.on_str is not None:
return self.on_str
return f'λ({self.child})'
def __repr__(self) -> str:
if self.on_repr is not None:
return self.on_repr
return f'{__name__}.Preproc({self.child!r}, parse={self.on_parse!r}, dump={self.on_dump!r}, sizeof={self.on_sizeof!r}, context={self.context!r}, bits={self.bits!r}, str={self.on_str!r}, repr={self.on_repr!r})'
class Proc(G[T, V], Type[V]):
def __init__(self, child: Type[T], parse: U[Callable[[T], V], Callable[[T, Context], V]], dump: U[Callable[[V], T], Callable[[V, Context], T]], context: bool = False, str: O[str] = None, repr: O[str] = None) -> None:
self.child = child
self.on_parse = parse
@ -68,10 +136,10 @@ class Transform(G[T, V], Type[V]):
def __repr__(self) -> str:
if self.on_repr is not None:
return self.on_repr
return f'{__name__}.Transform({self.child!r}, parse={self.on_parse!r}, dump={self.on_dump!r}, context={self.context!r}, str={self.on_str!r}, repr={self.on_repr!r})'
return f'{__name__}.Proc({self.child!r}, parse={self.on_parse!r}, dump={self.on_dump!r}, context={self.context!r}, str={self.on_str!r}, repr={self.on_repr!r})'
class Mapped(G[T, V], Transform[T, V]):
class Mapped(G[T, V], Proc[T, V]):
def __init__(self, child: Type[T], mapping: Mapping[T, V], str: O[str] = None, repr: O[str] = None) -> None:
reverse = {v: k for k, v in mapping.items()}
super().__init__(child,
@ -85,7 +153,7 @@ class Mapped(G[T, V], Transform[T, V]):
E = TypeVar('E', bound=enum.Enum)
class Enum(G[T, E], Transform[T, E]):
class Enum(G[T, E], Proc[T, E]):
def __init__(self, enum: Ty[E], child: Type[T]) -> None:
super().__init__(child,
parse=enum,