Source code for bibble.util.pair_stack

  1#!/usr/bin/env python3
  2"""
  3
  4"""
  5# Imports:
  6from __future__ import annotations
  7
  8# ##-- stdlib imports
  9import datetime
 10import enum
 11import functools as ftz
 12import itertools as itz
 13import logging as logmod
 14import pathlib as pl
 15import re
 16import time
 17import types
 18import collections
 19import contextlib
 20import hashlib
 21from copy import deepcopy
 22from uuid import UUID, uuid1
 23from weakref import ref
 24import atexit # for @atexit.register
 25import faulthandler
 26# ##-- end stdlib imports
 27
 28import bibble._interface as API
 29
 30# ##-- types
 31# isort: off
 32import abc
 33import collections.abc
 34from typing import TYPE_CHECKING, cast, assert_type, assert_never
 35from typing import Generic, NewType
 36# Protocols:
 37from typing import Protocol, runtime_checkable
 38# Typing Decorators:
 39from typing import no_type_check, final, override, overload
 40
 41if TYPE_CHECKING:
 42    from jgdv import Maybe, Fifo, Lifo
 43    from typing import Final
 44    from typing import ClassVar, Any, LiteralString
 45    from typing import Never, Self, Literal
 46    from typing import TypeGuard
 47    from collections.abc import Iterable, Iterator, Callable, Generator
 48    from collections.abc import Sequence, Mapping, MutableMapping, Hashable
 49
 50    type UniMiddleware   = API.UniMiddleware
 51    type BidiMiddleware  = API.BidiMiddleware
 52    type Middleware      = API.Middleware
 53
 54##--|
 55
 56# isort: on
 57# ##-- end types
 58
 59##-- logging
 60logging = logmod.getLogger(__name__)
 61##-- end logging
 62
 63# Vars:
 64
 65# Body:
 66

[docs]
 67class PairStack:
 68    """ A pair of middleware stacks,
 69    allowing reader/writer pairs to be added at the same time
 70
 71    The read stack is Fifo, the write stack is Lifo.
 72    So any transforms are undone in the correct order before writing.
 73    Parse[m1, m2, m3] -> Write[m3, m2, m1]
 74
 75    eg: Parse[LatexToUnicode, SplitName] -> Write[MergeNames, UnicodeToLatex]
 76
 77    """
 78    _read_time  : Fifo[Middleware]
 79    _write_time : Lifo[Middleware]
 80
 81    def __init__(self):
 82        self._read_time  = []
 83        self._write_time = []
 84

[docs]
 85    def add(self, *bidis:BidiMiddleware, read:Maybe[list[Middleware]]=None, write:Maybe[list[Middleware]]=None) -> Self:
 86        """
 87        Add middlewares to the read/write stacks.
 88        Args *must* be bidirection, and will be added to both.
 89        kwargs 'read' and 'write' just add to specific stacks
 90        """
 91        read  = read or []
 92        write = write or []
 93
 94        for mid in bidis:
 95            match mid:
 96                case None:
 97                     pass
 98                case API.BidirectionalMiddleware_p():
 99                    self._read_time.append(mid)
100                    self._write_time.append(mid)
101                case x:
102                    raise TypeError(type(x))
103
104        for x in read:
105            match x:
106                case None:
107                    pass
108                case API.Middleware_p():
109                    self._read_time.append(x)
110                case API.BidirectionalMiddleware_p():
111                    self._read_time.append(x)
112                case _:
113                    raise TypeError(type(x))
114
115        for x in write:
116            match x:
117                case None:
118                    pass
119                case API.Middleware_p():
120                    self._write_time.append(x)
121                case API.BidirectionalMiddleware_p():
122                    self._write_time.append(x)
123                case _:
124                    raise TypeError(type(x))
125
126        return self

127

[docs]
128    def read_stack(self) -> list[Middleware]:
129        """ Return the read stack """
130        return self._read_time[:]

131

[docs]
132    def write_stack(self) -> list[Middleware]:
133        """ Return the write stack """
134        return self._write_time[::-1]

135

[docs]
136    def has_read_transform(self, mid:type[Middleware]|Middleware) -> bool:
137        match mid:
138            case type():
139                return any(isinstance(x, mid) for x in self._read_time)
140            case API.Middleware_p():
141                return mid in self._read_time
142            case _:
143                return False

144

[docs]
145    def has_write_transform(self, mid:type[Middleware]|Middleware) -> bool:
146        match mid:
147            case type():
148                return any(isinstance(x, mid) for x in self._write_time)
149            case API.Middleware_p():
150                return mid in self._write_time
151            case _:
152                return False

153
154    def __contains__(self, other:type[Middleware]|Middleware) -> bool:
155        return self.has_read_transform(other) or self.has_write_transform(other)