223 lines
7.4 KiB
Python
223 lines
7.4 KiB
Python
"""Shared types and helpers for library implementations."""
|
|
from __future__ import annotations
|
|
|
|
from abc import ABC
|
|
from typing import TYPE_CHECKING, Literal, Protocol, TypeAlias
|
|
from collections.abc import Callable, Collection, Iterator, Mapping, MutableMapping, Sequence
|
|
import logging
|
|
import re
|
|
|
|
from ..error import LibraryError
|
|
|
|
if TYPE_CHECKING:
|
|
import numpy
|
|
from numpy.typing import NDArray
|
|
|
|
from ..pattern import Pattern
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
SINGLE_USE_PREFIX = '_'
|
|
"""
|
|
Names starting with this prefix are assumed to refer to single-use patterns,
|
|
which may be renamed automatically by `ILibrary.add()` (via
|
|
`rename_theirs=_rename_patterns()` )
|
|
"""
|
|
# TODO what are the consequences of making '_' special? maybe we can make this decision everywhere?
|
|
|
|
|
|
class INameView(Collection[str], ABC):
|
|
"""
|
|
Read-only collection of reserved names with a shared name allocator.
|
|
|
|
Name views support membership, iteration, `len()`, and `get_name()`. They
|
|
do not provide pattern lookup or the other operations of a library mapping.
|
|
"""
|
|
|
|
def get_name(
|
|
self,
|
|
name: str = SINGLE_USE_PREFIX * 2,
|
|
sanitize: bool = True,
|
|
max_length: int = 32,
|
|
quiet: bool | None = None,
|
|
) -> str:
|
|
"""
|
|
Find a unique name.
|
|
|
|
This function may be overridden in a subclass or monkey-patched to fit
|
|
the caller's requirements.
|
|
|
|
Args:
|
|
name: Preferred name. Default is `SINGLE_USE_PREFIX * 2`.
|
|
sanitize: Allow only alphanumeric characters and _?$, replacing
|
|
invalid characters with underscores.
|
|
max_length: Truncate names longer than this.
|
|
quiet: Suppress log messages when `True`. The default suppresses
|
|
messages only when `name` starts with `SINGLE_USE_PREFIX`.
|
|
|
|
Returns:
|
|
A name unique within this view.
|
|
"""
|
|
if quiet is None:
|
|
quiet = name.startswith(SINGLE_USE_PREFIX)
|
|
|
|
if sanitize:
|
|
sanitized_name = re.compile(r'[^A-Za-z0-9_\?\$]').sub('_', name)
|
|
else:
|
|
sanitized_name = name
|
|
|
|
suffixed_name = sanitized_name
|
|
if sanitized_name in self:
|
|
ii = sum(1 for nn in self if nn.startswith(sanitized_name))
|
|
else:
|
|
ii = 0
|
|
while suffixed_name in self or suffixed_name == '':
|
|
suffixed_name = sanitized_name + b64suffix(ii)
|
|
ii += 1
|
|
|
|
if len(suffixed_name) > max_length:
|
|
if name == '':
|
|
raise LibraryError(f'No valid pattern names remaining within the specified {max_length=}')
|
|
|
|
cropped_name = self.get_name(sanitized_name[:-1], sanitize=sanitize, max_length=max_length, quiet=True)
|
|
else:
|
|
cropped_name = suffixed_name
|
|
|
|
if not quiet:
|
|
logger.info(f'Requested name "{name}" changed to "{cropped_name}"')
|
|
|
|
return cropped_name
|
|
|
|
|
|
class visitor_function_t(Protocol):
|
|
""" Signature for `Library.dfs()` visitor functions. """
|
|
def __call__(
|
|
self,
|
|
pattern: Pattern,
|
|
hierarchy: tuple[str | None, ...],
|
|
memo: dict,
|
|
transform: NDArray[numpy.float64] | Literal[False],
|
|
) -> Pattern:
|
|
...
|
|
|
|
|
|
TreeView: TypeAlias = Mapping[str, 'Pattern']
|
|
""" A name-to-`Pattern` mapping which is expected to have only one top-level cell """
|
|
|
|
Tree: TypeAlias = MutableMapping[str, 'Pattern']
|
|
""" A mutable name-to-`Pattern` mapping which is expected to have only one top-level cell """
|
|
|
|
dangling_mode_t: TypeAlias = Literal['error', 'ignore', 'include']
|
|
""" How helpers should handle refs whose targets are not present in the library. """
|
|
|
|
|
|
def _rename_patterns(lib: INameView, name: str) -> str:
|
|
"""
|
|
The default `rename_theirs` function for `ILibrary.add`.
|
|
|
|
Treats names starting with `SINGLE_USE_PREFIX` (default: one underscore) as
|
|
"one-offs" for which name conflicts should be automatically resolved.
|
|
Conflicts are resolved by calling `lib.get_name(SINGLE_USE_PREFIX + stem)`
|
|
where `stem = name.removeprefix(SINGLE_USE_PREFIX).split('$')[0]`.
|
|
Names lacking the prefix are directly returned (not renamed).
|
|
|
|
Args:
|
|
lib: The library into which `name` is to be added (but is presumed to conflict)
|
|
name: The original name, to be modified
|
|
|
|
Returns:
|
|
The new name, not guaranteed to be conflict-free!
|
|
"""
|
|
if not name.startswith(SINGLE_USE_PREFIX):
|
|
return name
|
|
|
|
stem = name.removeprefix(SINGLE_USE_PREFIX).split('$')[0]
|
|
return lib.get_name(SINGLE_USE_PREFIX + stem)
|
|
|
|
|
|
def _validate_dangling_mode(dangling: dangling_mode_t) -> None:
|
|
if dangling not in ('error', 'ignore', 'include'):
|
|
raise ValueError(
|
|
f'Unknown dangling-reference mode {dangling!r}; '
|
|
'expected one of "error", "ignore", or "include"'
|
|
)
|
|
|
|
|
|
class _ProspectiveNames(INameView):
|
|
"""Target names plus names reserved earlier in an addition plan."""
|
|
|
|
def __init__(
|
|
self,
|
|
target: INameView,
|
|
reserved: set[str],
|
|
) -> None:
|
|
self._target = target
|
|
self._reserved = reserved
|
|
|
|
def __iter__(self) -> Iterator[str]:
|
|
yield from self._target
|
|
yield from self._reserved
|
|
|
|
def __len__(self) -> int:
|
|
return len(self._target) + len(self._reserved)
|
|
|
|
def __contains__(self, key: object) -> bool:
|
|
return key in self._reserved or key in self._target
|
|
|
|
|
|
def _plan_source_names(
|
|
target: INameView,
|
|
source_order: Sequence[str],
|
|
*,
|
|
rename_theirs: Callable[[INameView, str], str] | None = None,
|
|
rename_when: Literal['conflict', 'always'] = 'conflict',
|
|
) -> dict[str, str]:
|
|
if rename_when not in ('conflict', 'always'):
|
|
raise ValueError(f'Unknown source rename mode: {rename_when!r}')
|
|
if rename_when == 'always' and rename_theirs is None:
|
|
raise TypeError('rename_theirs is required when rename_when="always"')
|
|
|
|
source_to_visible: dict[str, str] = {}
|
|
reserved: set[str] = set()
|
|
prospective = _ProspectiveNames(target, reserved)
|
|
|
|
for name in source_order:
|
|
visible = name
|
|
if rename_when == 'always':
|
|
assert rename_theirs is not None
|
|
visible = rename_theirs(prospective, name)
|
|
elif visible in prospective:
|
|
if rename_theirs is None:
|
|
raise LibraryError(f'Conflicting name while adding source: {name!r}')
|
|
visible = rename_theirs(prospective, name)
|
|
if visible in prospective:
|
|
raise LibraryError(f'Unresolved duplicate key encountered while adding source: {name!r} -> {visible!r}')
|
|
source_to_visible[name] = visible
|
|
reserved.add(visible)
|
|
|
|
return source_to_visible
|
|
|
|
|
|
def _source_rename_map(source_to_visible: Mapping[str, str]) -> dict[str, str]:
|
|
return {
|
|
source_name: visible_name
|
|
for source_name, visible_name in source_to_visible.items()
|
|
if source_name != visible_name
|
|
}
|
|
|
|
def b64suffix(ii: int) -> str:
|
|
"""
|
|
Turn an integer into a base64-equivalent suffix.
|
|
|
|
This could be done with base64.b64encode, but this way is faster for many small `ii`.
|
|
"""
|
|
def i2a(nn: int) -> str:
|
|
return 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789$?'[nn]
|
|
|
|
parts = ['$', i2a(ii % 64)]
|
|
ii >>= 6
|
|
while ii:
|
|
parts.append(i2a(ii % 64))
|
|
ii >>= 6
|
|
return ''.join(parts)
|