# Auto-generated by generate.py. Do NOT edit by hand.
import mmap
import os
import numpy as np
AVAILABLE_LEXER: bool = False
try:
from fuzzy_dl_owl2.fuzzydl.parser.tokenizer._fdl_lexer import ffi, lib
AVAILABLE_LEXER = True
except ImportError: # pragma: no cover
try:
from _fdl_lexer import ffi, lib
AVAILABLE_LEXER = True
except ImportError:
ffi = lib = None
TOK_NAMES = {
1: "LPAREN",
2: "RPAREN",
3: "LBRACK",
4: "RBRACK",
5: "LBRACE",
6: "RBRACE",
7: "COMMA",
125: "GE",
124: "LE",
126: "EQ",
93: "FADD",
94: "FSUB",
95: "FMUL",
96: "FDIV",
121: "PLUS",
122: "MINUS",
123: "STAR",
8: "NUMBER",
9: "IDENT",
10: "ERROR",
57: "define-fuzzy-logic",
118: "lukasiewicz",
119: "zadeh",
120: "classical",
34: "define-truth-constant",
44: "define-modifier",
105: "linear-modifier",
106: "triangular-modifier",
38: "define-fuzzy-concept",
98: "crisp",
99: "left-shoulder",
100: "right-shoulder",
101: "triangular",
102: "trapezoidal",
103: "linear",
104: "modified",
40: "define-fuzzy-number-range",
39: "define-fuzzy-number",
54: "range",
129: "*integer*",
130: "*real*",
127: "*string*",
128: "*boolean*",
56: "constraints",
117: "binary",
116: "free",
111: "show-concrete-fillers-for",
110: "show-concrete-fillers",
112: "show-concrete-instance-for",
109: "show-abstract-fillers-for",
108: "show-abstract-fillers",
114: "show-concepts",
113: "show-instances",
107: "show-variables",
115: "show-language",
58: "crisp-concept",
59: "crisp-role",
41: "define-fuzzy-similarity",
42: "define-fuzzy-equivalence",
75: "*top*",
76: "*bottom*",
61: "g-and",
62: "l-and",
60: "and",
69: "g-or",
70: "l-or",
68: "or",
71: "not",
64: "g-implies",
66: "l-implies",
65: "kd-implies",
67: "z-implies",
49: "implies-role",
63: "implies",
74: "all",
73: "b-some",
72: "some",
91: "lua",
89: "tua",
82: "ua",
92: "lla",
90: "tla",
83: "la",
81: "self",
78: "w-sum-zero",
77: "w-sum",
79: "w-max",
80: "w-min",
85: "q-owa",
84: "owa",
86: "choquet",
88: "q-sugeno",
87: "sugeno",
97: "sigma-count",
35: "define-concept",
36: "define-primitive-concept",
37: "equivalent-concepts",
53: "disjoint-union",
52: "disjoint",
55: "domain",
51: "inverse-functional",
45: "functional",
47: "reflexive",
48: "symmetric",
46: "transitive",
50: "inverse",
11: "max-instance?",
12: "min-instance?",
13: "all-instances?",
14: "max-related?",
15: "min-related?",
33: "instance",
43: "related",
16: "max-subs?",
20: "min-subs?",
17: "max-g-subs?",
21: "min-g-subs?",
18: "max-l-subs?",
22: "min-l-subs?",
19: "max-kd-subs?",
23: "min-kd-subs?",
24: "max-sat?",
25: "min-sat?",
26: "max-var?",
27: "min-var?",
29: "defuzzify-lom?",
31: "defuzzify-mom?",
30: "defuzzify-som?",
32: "bnp?",
28: "sat?",
}
NUMBER = 8
IDENT = 9
ERROR = 10
FIRST_KEYWORD = 57
LPAREN = 1
RPAREN = 2
LBRACK = 3
RBRACK = 4
LBRACE = 5
RBRACE = 6
COMMA = 7
GE = 125
LE = 124
EQ = 126
FADD = 93
FSUB = 94
FMUL = 95
FDIV = 96
PLUS = 121
MINUS = 122
STAR = 123
T_IDENT = 9
T_EOF = 131
_PAGE = mmap.PAGESIZE
[docs]
class _Buffer:
"""Read-only file view with a guaranteed NUL at index `size` (needed by the
re2c backend; harmless for flex). mmap zero-fills the final partial page for
free; only a page-multiple file size needs a one-off copy with a NUL."""
__slots__ = ("size", "obj", "cdata", "view", "_mm")
def __init__(self, path):
"""
Opens the file at ``path`` read-only and exposes it as a NUL-terminated buffer suitable for the C tokenizer. An empty file becomes a single NUL byte; a file whose size is an exact multiple of the page size is copied with an appended NUL (since mmap cannot zero-pad a full final page), while any other file is memory-mapped, relying on mmap's free zero-fill of the trailing partial page. A CFFI pointer (``cdata``) and a ``memoryview`` over the data are prepared for downstream tokenizing.
:param path: Filesystem path of the file to map for tokenization.
:type path: typing.Any
"""
fd = os.open(path, os.O_RDONLY)
try:
self.size = os.fstat(fd).st_size
if self.size == 0:
self.obj, self._mm = b"\x00", None
elif self.size % _PAGE == 0:
self.obj = os.pread(fd, self.size, 0) + b"\x00"
self._mm = None
else:
self._mm = mmap.mmap(fd, self.size, access=mmap.ACCESS_READ)
self.obj = self._mm
finally:
os.close(fd)
self.cdata = ffi.from_buffer(self.obj)
self.view = memoryview(self.obj)
[docs]
def close(self):
"""
Releases the resources held by this buffer. The exported ``memoryview`` is released and the CFFI pointer is dropped before the underlying ``mmap`` (if any) is closed, ensuring no live pointer references the memory map when it is unmapped. Safe to call once after tokenization is complete.
"""
self.view.release()
self.cdata = None # drop CFFI's exported pointer into the mmap
if self._mm is not None:
self._mm.close()
[docs]
class _BytesBuffer:
"""In-memory counterpart of ``_Buffer`` for tokenizing a source string
that is already resident (a parser chunk). Appends the NUL sentinel the
re2c backend needs; exposes the same ``cdata`` / ``view`` / ``size`` /
``close`` surface as ``_Buffer`` so ``Tokens`` is agnostic to the source."""
__slots__ = ("size", "obj", "cdata", "view")
def __init__(self, data: bytes):
"""
Wraps an already-resident byte string as a NUL-terminated tokenizer buffer. A single NUL sentinel is appended (required by the re2c backend, harmless for flex) and the original length is recorded as ``size``. A CFFI pointer and a ``memoryview`` over the padded bytes are prepared, mirroring the ``cdata``/``view``/``size``/``close`` surface of :class:`_Buffer` so callers stay source-agnostic.
:param data: The in-memory source bytes to tokenize.
:type data: bytes
"""
self.size = len(data)
self.obj = data + b"\x00" # NUL sentinel for re2c; harmless for flex
self.cdata = ffi.from_buffer(self.obj)
self.view = memoryview(self.obj)
[docs]
def close(self):
"""
Releases the resources held by this in-memory buffer. The exported ``memoryview`` is released and the CFFI pointer is dropped so no live pointer references the byte string afterwards. Safe to call once after tokenization is complete.
"""
self.view.release()
self.cdata = None # drop CFFI's exported pointer before releasing
[docs]
class Tokens:
"""Flat token stream. Holds the buffer so span offsets stay valid."""
__slots__ = ("types", "starts", "lens", "_buf")
def __init__(self, types, starts, lens, buf):
"""
Builds a flat token stream from three parallel int32 arrays and the source buffer they index into. ``types`` holds the per-token codes, ``starts`` the byte offset of each token, and ``lens`` its byte length; the owning ``buf`` is retained so the span offsets stay valid and the original text can be decoded lazily on demand.
:param types: int32 array of per-token type codes (see ``TOK_NAMES``).
:type types: typing.Any
:param starts: int32 array of per-token byte start offsets into the buffer.
:type starts: typing.Any
:param lens: int32 array of per-token byte lengths.
:type lens: typing.Any
:param buf: The source buffer the offsets refer to, kept alive for span decoding.
:type buf: typing.Any
"""
self.types = types # int32[n] token codes (see TOK_NAMES)
self.starts = starts # int32[n] byte offset of each token
self.lens = lens # int32[n] byte length of each token
self._buf = buf
[docs]
def __len__(self):
"""
Returns the number of tokens in the stream, i.e. the length of the underlying ``types`` array.
:return: The token count.
:rtype: int
"""
return len(self.types)
[docs]
def name(self, i):
"""
Returns the display name of token *i*: the structural token name for
punctuation, or the keyword spelling itself for keyword tokens.
:param i: Index of the token in the stream.
:type i: int
:return: The display name of the token.
:rtype: str
"""
return TOK_NAMES[int(self.types[i])]
[docs]
def text(self, i):
"""
Returns the raw source text of token *i*, lazily decoded from the backing
buffer via the span offset and length. Only the requested token is
decoded, so repeated lookups are cheap.
:param i: Index of the token in the stream.
:type i: int
:return: The UTF-8 decoded source text of the token.
:rtype: str
"""
s = int(self.starts[i])
return bytes(self._buf.view[s : s + int(self.lens[i])]).decode("utf-8")
[docs]
def is_keyword(self, i):
"""
Returns whether token *i* is a keyword. This is a pure integer code
comparison, so no string work is performed.
:param i: Index of the token in the stream.
:type i: int
:return: ``True`` if the token is a keyword, ``False`` otherwise.
:rtype: bool
"""
return int(self.types[i]) >= FIRST_KEYWORD
[docs]
def is_name(self, i):
"""
Returns whether token *i* is a user-defined bare identifier (as opposed
to a keyword or structural token).
:param i: Index of the token in the stream.
:type i: int
:return: ``True`` if the token is a bare identifier, ``False`` otherwise.
:rtype: bool
"""
return int(self.types[i]) == IDENT
[docs]
def close(self):
"""
Releases the source buffer backing this token stream by delegating to its ``close`` method. After this call the lazy ``text`` lookups are no longer valid, since the byte storage the span offsets point into has been released.
"""
self._buf.close()
[docs]
def __enter__(self):
"""
Enters a context manager over the token stream and returns it, so the stream can be used in a ``with`` block that guarantees the backing buffer is released on exit.
:return: This token stream instance.
:rtype: Tokens
"""
return self
[docs]
def __exit__(self, *exc):
"""
Exits the context manager, releasing the backing buffer via :meth:`close` regardless of whether the ``with`` block exited normally or because of an exception. Any exception is allowed to propagate.
:param exc: The exception type, value, and traceback (unused), present to satisfy the context-manager protocol.
:type exc: typing.Any
"""
self.close()
[docs]
def _tokenize_buffer(buf) -> Tokens:
"""
Two-pass tokenization over any buffer exposing ``cdata`` / ``size``.
Pass 1 counts tokens in C so the parallel int32 arrays can be sized
exactly; Pass 2 fills them. Shared by :func:`get_tokens` (mmap) and
:func:`get_tokens_from_bytes` (in-memory) so the count/fill logic lives
in one place.
:param buf: A buffer object (``_Buffer`` or ``_BytesBuffer``) exposing
``cdata`` and ``size`` for the C scanner.
:type buf: typing.Any
:return: A :class:`Tokens` object holding the parallel token arrays.
:rtype: Tokens
"""
n = buf.size
# Pass 1: count, so arrays are sized exactly (memory owned by NumPy).
ntok = lib.fdl_count_tokens(buf.cdata, n)
types = np.empty(ntok, dtype=np.int32)
starts = np.empty(ntok, dtype=np.int32)
lens = np.empty(ntok, dtype=np.int32)
# Pass 2: fill the parallel arrays in C.
lib.fdl_tokenize(
buf.cdata,
n,
ffi.cast("int32_t *", ffi.from_buffer(types)),
ffi.cast("int32_t *", ffi.from_buffer(starts)),
ffi.cast("int32_t *", ffi.from_buffer(lens)),
)
return Tokens(types, starts, lens, buf)
[docs]
def get_tokens(path: str) -> Tokens:
"""
Tokenizes the file at *path* via mmap. No whole-file Python ``str`` is
built, so this is the fast / low-memory entry point for large ontologies.
The returned :class:`Tokens` owns the mapping; call ``.close()`` (or use
it as a context manager) to release it.
:param path: Filesystem path of the file to tokenize.
:type path: str
:return: The token stream backed by a memory-mapped buffer.
:rtype: Tokens
"""
return _tokenize_buffer(_Buffer(path))
[docs]
def get_tokens_from_bytes(data: bytes) -> Tokens:
"""
Tokenizes an in-memory source already decoded to UTF-8 *bytes* (e.g. a
parser chunk). Same :class:`Tokens` output as :func:`get_tokens`, without
a file or mmap.
:param data: The UTF-8 encoded source bytes to tokenize.
:type data: bytes
:return: The token stream backed by an in-memory buffer.
:rtype: Tokens
"""
return _tokenize_buffer(_BytesBuffer(data))