from __future__ import annotations
from collections.abc import Sequence
from typing import ClassVar, Self
from uuid import uuid4
import pyparsing
from rdflib import Graph, URIRef
type DataType = str
type TreeList = list[str | TreeList]
class TreeOld:
"""A labeled ordered tree with search and RDF conversion methods."""
RDF_TYPES: ClassVar[dict[str, URIRef]] = {}
RDF_EDGES: ClassVar[dict[str, URIRef]] = {
"is": URIRef("is-a"),
"parent": URIRef("is-the-parent-of"),
"child": URIRef("is-a-child-of"),
"sister": URIRef("is-a-sister-of"),
}
LPAR = pyparsing.Suppress("(")
RPAR = pyparsing.Suppress(")")
DATA = pyparsing.Regex(r"[^()\s]+")
PARSER = pyparsing.Forward()
PARSER_LIST = pyparsing.Group(LPAR + DATA + pyparsing.ZeroOrMore(PARSER) + RPAR)
PARSER <<= DATA | PARSER_LIST
SOURCE_PARSER = (
LPAR + pyparsing.OneOrMore(PARSER_LIST) + RPAR
) | PARSER_LIST
def __init__(
self,
data: DataType,
children: Sequence[Self] | None = None,
) -> None:
if children is not None and (
isinstance(children, (str, bytes))
or not isinstance(children, Sequence)
):
raise TypeError("children must be a finite sequence of trees")
self._data = data
self._children: tuple[Self, ...] = tuple(
() if children is None else children
)
self._validate()
def _validate(self) -> None:
"""Reject non-trees, cycles, and shared subtree objects."""
seen = {id(self)}
def visit(node: Self) -> None:
if not isinstance(node, type(self)):
raise TypeError("all children must use the same tree class")
if id(node) in seen:
raise ValueError("trees cannot contain cycles or shared subtrees")
seen.add(id(node))
for child in node._children:
visit(child)
for child in self._children:
visit(child)
def __str__(self) -> str:
if self._children:
return " ".join(str(child) for child in self._children)
return self._data
def __repr__(self) -> str:
return self.to_string()
def to_string(self, depth: int = 0) -> str:
"""Render this tree as an indented string."""
prefix = " " * max(depth - 1, 0) + ("--" if depth else "")
return (
prefix
+ self._data
+ "\n"
+ "".join(child.to_string(depth + 1) for child in self._children)
)
def __contains__(self, data: DataType) -> bool:
return self._data == data or any(data in child for child in self._children)
def __getitem__(self, idx: int | tuple[int, ...]) -> TreeOld:
if isinstance(idx, int):
if idx < 0:
raise IndexError("tree indices must be nonnegative")
return self._children[idx]
if not isinstance(idx, tuple) or not all(
isinstance(i, int) and i >= 0 for i in idx
):
raise IndexError("index must be an int or tuple of nonnegative integers")
if not idx:
return self
return self._children[idx[0]][idx[1:]]
@property
def data(self) -> DataType:
"""The data stored at this node."""
return self._data
@property
def children(self) -> tuple[Self, ...]:
"""The children of this node."""
return self._children
def index(
self,
data: DataType,
index_path: tuple[int, ...] = (),
) -> list[tuple[int, ...]]:
"""Return every root-relative path whose node stores ``data``."""
indices = [index_path] if self._data == data else []
indices.extend(
match
for i, child in enumerate(self._children)
for match in child.index(data, index_path + (i,))
)
return indices
def to_rdf(self, graph: Graph | None = None) -> Graph:
"""Encode the tree's labels and structural relations as RDF."""
graph = Graph() if graph is None else graph
namespace = f"urn:tree:{uuid4().hex}:"
nodes: dict[tuple[int, ...], URIRef] = {}
def encode(tree: Self, idx: tuple[int, ...]) -> None:
path_string = "root" if not idx else ".".join(map(str, idx))
nodes[idx] = URIRef(namespace + path_string)
type_uri = self.RDF_TYPES.setdefault(tree.data, URIRef(tree.data))
graph.add((nodes[idx], self.RDF_EDGES["is"], type_uri))
for i, child in enumerate(tree.children):
child_idx = idx + (i,)
encode(child, child_idx)
graph.add((nodes[idx], self.RDF_EDGES["parent"], nodes[child_idx]))
graph.add((nodes[child_idx], self.RDF_EDGES["child"], nodes[idx]))
for i, _ in enumerate(tree.children):
for j, _ in enumerate(tree.children):
if i != j:
graph.add(
(
nodes[idx + (i,)],
self.RDF_EDGES["sister"],
nodes[idx + (j,)],
)
)
encode(self, ())
self._rdf_paths = {node: path for path, node in nodes.items()}
return graph
@property
def rdf(self) -> Graph:
"""A fresh RDF graph for this tree."""
return self.to_rdf()
def find(self, query: str) -> list[tuple[int, ...]]:
"""Return paths selected in the first column of a SPARQL query."""
graph = self.rdf
paths: list[tuple[int, ...]] = []
seen: set[tuple[int, ...]] = set()
for result in graph.query(query):
if isinstance(result, bool):
raise ValueError("find requires a SPARQL SELECT query")
row = tuple(result)
if (
not row
or not isinstance(row[0], URIRef)
or row[0] not in self._rdf_paths
):
raise ValueError("the first selected column must be a tree node")
path = self._rdf_paths[row[0]]
if path not in seen:
paths.append(path)
seen.add(path)
return paths
@classmethod
def from_string(cls, tree_string: str) -> Self:
"""Parse one Penn-style bracketed tree."""
source = tree_string.strip()
parsed = cls.SOURCE_PARSER.parse_string(source, parse_all=True)
if len(parsed) == 1:
return cls.from_list(parsed[0])
return cls(
"ROOT",
[cls.from_list(tree_list) for tree_list in parsed],
)
@classmethod
def from_list(
cls,
tree_list: str | TreeList | pyparsing.ParseResults,
) -> Self:
"""Build a tree from a nested parser result."""
if isinstance(tree_list, str):
return cls(tree_list)
if not tree_list:
raise ValueError("a tree list must contain a node label")
return cls(
str(tree_list[0]),
[cls.from_list(child) for child in tree_list[1:]],
)Assignments 5 and 6
Like Assignments 1/2 and 3/4, Assignments 5 and 6 are bundled together. You only need to do Task 1 for Assignment 5 and Tasks 2 and 3 for Assignment 6.
How can we search for a near match rather than an exact one? These assignments answer that question using edit distance and regular expressions. The appendix on Working with Annotated Corpora develops the exact tree-search algorithms that we will extend here. Read through it before starting, since the code used here is developed there. I’ve copied the relevant class below as TreeOld.
A fuzzy search loosens the exact-match restriction in two ways. We can require matches to be (i) within some fixed edit distance and/or (ii) closest to the query among all pieces of data. I’ve copied a simplified version of the edit-distance class that we developed in class below as EditDistance.
import numpy as np
from numpy.typing import NDArray
type SymbolSequence = Sequence[str]
class EditDistance:
"""Weighted Levenshtein distance computed by dynamic programming."""
def __init__(
self,
insertion_cost: float = 1.0,
deletion_cost: float = 1.0,
substitution_cost: float | None = None,
) -> None:
self._insertion_cost = insertion_cost
self._deletion_cost = deletion_cost
self._substitution_cost = (
insertion_cost + deletion_cost
if substitution_cost is None
else substitution_cost
)
def __call__(self, source: SymbolSequence, target: SymbolSequence) -> float:
"""Return the minimum cost of transforming ``source`` into ``target``."""
source_symbols = list(source)
target_symbols = list(target)
n = len(source_symbols)
m = len(target_symbols)
distance: NDArray[np.float64] = np.zeros((n + 1, m + 1))
for i in range(1, n + 1):
distance[i, 0] = distance[i - 1, 0] + self._deletion_cost
for j in range(1, m + 1):
distance[0, j] = distance[0, j - 1] + self._insertion_cost
for i in range(1, n + 1):
for j in range(1, m + 1):
substitution = (
0.0
if source_symbols[i - 1] == target_symbols[j - 1]
else self._substitution_cost
)
distance[i, j] = min(
distance[i - 1, j] + self._deletion_cost,
distance[i - 1, j - 1] + substitution,
distance[i, j - 1] + self._insertion_cost,
)
return float(distance[n, m])Task 1
Lines: 14
Define an instance method fuzzy_find. This method should take query data and an optional distance. It should return all nodes whose data is within edit distance distance of the query. If closest is True, it should instead return all closest nodes that also satisfy the distance bound.
For instance:
fuzzy_find('review', distance=3., closest=False)will return a tuple of every piece of data in the tree within edit distance 3 from review (e.g. view, reviewer, reviews, etc.), its distance to review, and its index; if there is nothing within that edit distance, an empty list will be returnedfuzzy_find('review', distance=3., closest=True)will return a tuple of the closest pieces of data in the tree that are also within edit distance 3 from review (e.g. view, reviewer, reviews, etc.), its distance to review, and its index; if there is nothing within that edit distance, an empty list will be returnedfuzzy_find('review', distance=np.inf, closest=True)will return a tuple of the closest pieces of data in the tree to review (e.g. view, reviewer, reviews, etc.), regardless of edit distance, its distance to review, and its index; this will always return somethingfuzzy_find('review', distance=np.inf, closest=False)will return a tuple of every piece of data in the tree to review (e.g. view, reviewer, reviews, etc.), regardless of edit distance, its distance to review, and its index; this will always return a list with as many elements as there are nodes in the tree
Each result is an (index, node_data, node_distance) tuple.
This method should also support only searching the terminal nodes (leaves) of the tree with the flag terminals_only.
Hint: you should look back at the methods we defined for searching and indexing the tree above. Specifically, to understand why you might want something like index_path defaulting to the empty tuple, look at the index method of TreeOld.
type FuzzyFindResult = tuple[tuple[int, ...], str, float]
class Tree(TreeOld):
"""A tree with fuzzy search capabilities via edit distance."""
DIST = EditDistance(1.0, 1.0)
def fuzzy_find(
self,
data: Sequence[str],
closest: bool = True,
distance: float = np.inf,
case_fold: bool = True,
terminals_only: bool = True,
index_path: tuple[int, ...] = (),
) -> list[FuzzyFindResult]:
"""Find tree data within a specified edit distance."""
raise NotImplementedErrorWrite tests that use the following tree as input data.
treestr = "( (SBARQ (WHNP-1 (WP What)) (SQ (NP-SBJ-1 (-NONE- *T*)) (VP (VBZ is) (NP-PRD (NP (DT the) (JJS best) (NN place)) (SBAR (WHADVP-2 (-NONE- *0*)) (S (NP-SBJ (-NONE- *PRO*)) (VP (TO to) (VP (VB get) (NP (NP (NNS discounts)) (PP (IN for) (NP (NML (NNP San) (NNP Francisco)) (NNS restaurants)))) (ADVP-LOC-2 (-NONE- *T*))))))))) (. ?)) )"
testtree = Tree.from_string(treestr)
testtreeThe tests should test the four combinations of distance and closest listed above with the same query data, both with and without terminals_only=True and terminals_only=False (eight tests in total). Two further tests should test distance=np.inf, closest=True, terminals_only=True for a case where only a single element should be returned and a case where multiple elements in the tree should be returned.
# write tests hereWhat should the distance be between a string and a collection of strings? As we discussed in class, it is the minimum of the edit distances between the string and each string in that set.
We can use this concept in two ways here. The first is to view the tree as a container for some data and to compute the minimum distance between a query and any data contained in the tree. Alternatively, we can think of the query itself as determining a set and compute the minimum distance of each piece of data in the tree to that set. Task 2 will implement the former and Task 3 the latter.
I’ve copied the corpus reader we developed for the English Web Treebank in class below. We’ll make use of this for Task 2. (You’ll need to grab LDC2012T13.tgz from the course Google drive.)
from pathlib import Path
archive_path = Path("LDC2012T13.tgz")
if not archive_path.exists():
print(
"Place the licensed LDC2012T13.tgz archive in the "
"notebook directory before running the corpus tasks."
)from collections.abc import Iterator
from pathlib import Path
import tarfile
type CorpusFuzzyFindResult = tuple[str, tuple[int, ...], str, float]
class EnglishWebTreebankOld:
"""A reusable lazy reader for the English Web Treebank archive."""
def __init__(self, root: str | Path = "LDC2012T13.tgz") -> None:
self._root = Path(root)
def items(self) -> Iterator[tuple[str, Tree]]:
"""Yield every nonempty tree line in every ``.xml.tree`` member."""
with tarfile.open(self._root) as corpus:
for member in corpus.getmembers():
if not (member.isfile() and member.name.endswith(".xml.tree")):
continue
tree_file = corpus.extractfile(member)
if tree_file is None:
raise OSError(f"could not read archive member {member.name}")
with tree_file:
for raw_line in tree_file:
tree_string = raw_line.decode()
if tree_string.strip():
yield member.name, Tree.from_string(tree_string)Task 2
Lines: 3
Define an instance method fuzzy_find for the corpus reader class. For every tree in the corpus, it should compute the minimum distance between the query and that tree. Return a list of tuples whose four elements are a tree ID, an index in that tree, the data at that index, and the distance between the query and that data. Include a tuple only when its distance equals the minimum across all trees in the corpus.
Hint: which parameterization of Tree1.fuzzy_find gives exactly this result?
class EnglishWebTreebank(EnglishWebTreebankOld):
"""An English Web Treebank reader with cross-tree fuzzy search."""
def fuzzy_find(self, data: Sequence[str]) -> list[CorpusFuzzyFindResult]:
"""Return every corpus node at the global minimum distance."""
raise NotImplementedErrorNow, load this corpus.
ewt = EnglishWebTreebank()Write a single test for a piece of data you know exists in some tree in the corpus. (Determiners or auxiliary verbs are good candidates.) The minimum distance will thus be zero, and your method should return only trees that contain that element. The test should use an existing method to produce the correct set of trees.
Hint: such a method already exists in the TreeOld class.
# write test hereThe next task will look at computing distance between the elements of a tree and a query set defined by a regular expression. Here is a regular expression class based on the formal definition of regular expressions I gave you in class.
from string import ascii_letters
from typing import Literal, cast
type RegexStructure = (
tuple[Literal["literal"], str]
| tuple[Literal["epsilon"], None]
| tuple[
Literal["alternation", "concatenation"],
tuple[RegexStructure, ...],
]
| tuple[Literal["optional", "star", "plus"], RegexStructure]
)
class Regex:
"""A regular expression with exact length-bounded enumeration."""
def __init__(self, regex_parsed: RegexStructure, maxlength: int) -> None:
if maxlength < 0:
raise ValueError("maxlength must be nonnegative")
self._regex_parsed = regex_parsed
self._maxlength = maxlength
@classmethod
def from_string(cls, regexstr: str, maxlength: int = 30) -> Regex:
"""Parse literals, grouping, alternation, concatenation, and *, +, ?."""
source = "".join(regexstr.split())
position = 0
def parse_alternation() -> RegexStructure:
nonlocal position
alternatives = [parse_concatenation()]
while position < len(source) and source[position] == "|":
position += 1
alternatives.append(parse_concatenation())
if len(alternatives) == 1:
return alternatives[0]
return ("alternation", tuple(alternatives))
def parse_concatenation() -> RegexStructure:
nonlocal position
factors = []
while position < len(source) and source[position] not in ")|":
factors.append(parse_repetition())
if not factors:
return ("epsilon", None)
if len(factors) == 1:
return factors[0]
return ("concatenation", tuple(factors))
def parse_repetition() -> RegexStructure:
nonlocal position
atom = parse_atom()
if position < len(source) and source[position] in "*+?":
quantifier = source[position]
position += 1
if quantifier == "*":
return ("star", atom)
if quantifier == "+":
return ("plus", atom)
return ("optional", atom)
return atom
def parse_atom() -> RegexStructure:
nonlocal position
if position >= len(source):
raise ValueError("expected a regular-expression atom")
symbol = source[position]
if symbol in ascii_letters:
position += 1
return ("literal", symbol)
if symbol == "(":
position += 1
expression = parse_alternation()
if position >= len(source) or source[position] != ")":
raise ValueError("unclosed parenthesis in regular expression")
position += 1
return expression
raise ValueError(
f"unexpected regular-expression symbol: {symbol!r}"
)
parsed = parse_alternation()
if position != len(source):
raise ValueError(
f"unexpected regular-expression symbol: {source[position]!r}"
)
return cls(parsed, maxlength)
@staticmethod
def _concatenate(
left: set[str],
right: set[str],
maxlength: int,
) -> set[str]:
return {
first + second
for first in left
for second in right
if len(first) + len(second) <= maxlength
}
@classmethod
def _language(
cls,
regex: RegexStructure,
maxlength: int,
) -> set[str]:
kind, value = regex
if kind == "literal":
literal = cast(str, value)
return {literal} if len(literal) <= maxlength else set()
if kind == "epsilon":
return {""}
if kind == "alternation":
branches = cast(tuple[RegexStructure, ...], value)
return set().union(
*(cls._language(branch, maxlength) for branch in branches)
)
if kind == "concatenation":
factors = cast(tuple[RegexStructure, ...], value)
language = {""}
for factor in factors:
language = cls._concatenate(
language,
cls._language(factor, maxlength),
maxlength,
)
return language
if kind == "optional":
child = cast(RegexStructure, value)
return {""} | cls._language(child, maxlength)
if kind == "star":
child = cast(RegexStructure, value)
base = cls._language(child, maxlength) - {""}
language = {""}
frontier = {""}
while frontier:
frontier = (
cls._concatenate(frontier, base, maxlength) - language
)
language.update(frontier)
return language
if kind == "plus":
child = cast(RegexStructure, value)
base = cls._language(child, maxlength)
closure = cls._language(("star", child), maxlength)
return cls._concatenate(base, closure, maxlength)
raise ValueError(f"unknown regular-expression node: {kind}")
@classmethod
def _shortest_word(cls, regex: RegexStructure) -> str:
kind, value = regex
if kind == "literal":
return cast(str, value)
if kind in {"epsilon", "optional", "star"}:
return ""
if kind == "alternation":
branches = cast(tuple[RegexStructure, ...], value)
return min(
(cls._shortest_word(branch) for branch in branches),
key=lambda word: (len(word), word),
)
if kind == "concatenation":
factors = cast(tuple[RegexStructure, ...], value)
return "".join(
cls._shortest_word(factor) for factor in factors
)
if kind == "plus":
return cls._shortest_word(cast(RegexStructure, value))
raise ValueError(f"unknown regular-expression node: {kind}")
def bounded_for_target_length(
self,
max_target_length: int,
) -> Regex:
"""Return a bound sufficient for exact unit-cost edit distance."""
if max_target_length < 0:
raise ValueError("target length must be nonnegative")
witness_length = len(self._shortest_word(self._regex_parsed))
safe_bound = (
max_target_length
+ max(max_target_length, witness_length)
)
return Regex(self._regex_parsed, safe_bound)
def __iter__(self) -> Iterator[str]:
words = sorted(
self._language(self._regex_parsed, self._maxlength),
key=lambda word: (len(word), word),
)
self._gen = iter(words)
return self
def __next__(self) -> str:
return next(self._gen)What does iterating over a Regex return? It returns every distinct string in the language whose length is at most maxlength. This cutoff makes an infinite regular language finitely enumerable. The parser supports literals, grouping, alternation, concatenation, and the postfix quantifiers *, +, and ?.
A cutoff chosen only for convenience does not by itself guarantee an exact edit distance to an infinite language. The method bounded_for_target_length(n) returns a copy with a sufficient cutoff for unit-cost edit distance to targets of length at most n. It uses a shortest regex string as a finite-distance witness; any longer candidate whose length difference already exceeds that witness distance cannot improve the minimum.
assert list(Regex.from_string("a*", 3)) == ["", "a", "aa", "aaa"]
assert set(Regex.from_string("(a|b)*", 2)) == {
"", "a", "b", "aa", "ab", "ba", "bb",
}
assert list(Regex.from_string("a|bc", 3)) == ["a", "bc"]
for s in Regex.from_string("co+lou?r", 20):
print(s)Task 3
Define a new version of fuzzy_find that behaves exactly like your Task 1 version, except that the query is a regular-expression string parsable by Regex.from_string or an existing Regex object. The distance from a node label to the query is the minimum edit distance from that label to any string in the regex language.
At the root call, first compute the maximum node-label length in the tree. Then use bounded_for_target_length to obtain a finite enumeration sufficient for exact unit-cost distance, and pass that same bounded Regex object through the recursive calls. This step is necessary when the regex language is infinite.
Hint: the tree-wide maximum can be computed by a short recursive helper. Apart from constructing and bounding the Regex once at the root and minimizing over its generated strings, the traversal and filtering logic can follow Task 1.
class Tree(TreeOld):
"""A tree with fuzzy search against a regular-expression language."""
DIST = EditDistance(1.0, 1.0, 1.0)
def _max_data_length(self, case_fold: bool = True) -> int:
"""Return the maximum normalized node-label length."""
current = self._data.casefold() if case_fold else self._data
return max([
len(current),
*(child._max_data_length(case_fold) for child in self._children),
])
def fuzzy_find(
self,
data: str | Regex,
closest: bool = True,
distance: float = np.inf,
case_fold: bool = True,
terminals_only: bool = True,
index_path: tuple[int, ...] = (),
) -> list[FuzzyFindResult]:
"""Find node labels nearest to a regular-expression language."""
raise NotImplementedErrorWrite tests analogous to the ones you wrote for Task 1.
# write tests here