import logging
from collections import defaultdict
from itertools import combinations, product
from pathlib import Path
import numpy as np
import pandas as pd
from tqdm import tqdm
from . import alignment
from .generalize import generalize_patterns, incremental_generalize_patterns
from .patterns import Pattern
from ..utils import memory_check
from ..lattice.lattice import ICLattice
import multiprocessing as mp
log = logging.getLogger("Qumin")
ctx = mp.get_context('spawn')
def _init_worker(inventory):
global _inv
_inv = inventory
[docs]
def patterns_job(*args):
return _find_cellpair_patterns(_inv, *args)
[docs]
class PatternStore(dict):
"""
This class stores all alternation patterns computed for a paradigm.
Attributes:
cells (list[str]): cells for which patterns are registered.
"""
algorithm = 'unset'
cells = []
pairs = []
has_applicable = False
def __init__(self, *args, **kwargs):
dict.__init__(self, *args, **kwargs)
[docs]
def get_pairs(self):
if self.pairs == []:
self.pairs = list(combinations(self.cells, 2))
return self.pairs
[docs]
def info(self):
log.debug('Patterns:')
log.debug(self.__repr__())
if len(self.keys()) > 0:
log.debug(list(self.values())[0].head())
else:
log.debug('Does not contain any dataframe')
[docs]
def add_frequencies(self, frequencies,
token_patterns=False, token_predictors=False,
token_oa=False):
"""
Retrieves frequencies for pairs of forms and patterns. Frequencies are added
to the following columns:
- `f_pred`: frequency of the predictor.
- `f_pair`: relative frequency of the pattern within the lexeme.
Arguments:
token_patterns (bool): Whether to use token frequencies to compute
pattern probabilities.
token_predictors (bool): Whether to use token frequencies to weight
the predictors.
token_oa (bool): Whether to use token frequencies to compute
the relative weight of overabundant forms.
"""
log.info('Computing pattern frequencies...')
# Prepare frequency data
tok_freq = (
frequencies
.get_absolute_freq(group_on='form')
.to_dict()
)
typ_freq = (
frequencies
.get_relative_freq(group_on=["lexeme", 'cell'],
uniform_duplicates=not token_oa)
.result.to_dict()
)
# Assign frequencies to each pair
for pair, df in tqdm(self.items()):
# Clean-up the table if frequencies were already computed.
to_drop = [i for i in ['f_pred', 'f_pair'] if i in df.columns]
if to_drop:
df.drop(to_drop, axis=1, inplace=True)
selector = df.pattern.notna()
# We compute the probability of the predictors.
df.loc[selector, 'f_pred'] = (
df.loc[selector].form_x
.apply(lambda x: x.id)
.map(tok_freq if token_predictors else typ_freq)
)
# We compute the probability of the pairs
freq = tok_freq if token_patterns else typ_freq
f_pred = (
df.loc[selector, 'f_pred'] if token_predictors else
df.loc[selector].form_x.apply(lambda x: x.id).map(freq)
)
f_out = df.form_y.apply(lambda x: x.id).map(freq)
# Final result
df.loc[selector, 'f_pair'] = f_pred * f_out
df.loc[~selector, ['f_pair', 'f_pred']] = 0
[docs]
def find_patterns(self, paradigms, *args, algorithm="edits", disable_tqdm=False,
cpus=1, optim_mem=False, **kwargs):
"""Find Patterns in a DataFrame.
Algorithms can be:
- edits (dynamic alignment using levenshtein scores)
- phon (dynamic alignment using segment similarity scores)
Patterns are chosen according to their coverage and accuracy among competing patterns,
and they are merged as much as possible. Their alternation can be generalized.
This method updates the internal dict and does not return anything.
The internal dict is of shape dict of tuples to pd.DataFrame,
where the tuples are pairs of cells and the dataframes hold patterns for pairs of cells.
Arguments:
paradigms (:class:`qumin.representations.paradigms.Paradigms`): Paradigms object.
algorithm (str): method for scoring the best pairwise alignments. Can be "edits" or "phon".
disable_tqdm (bool): if true, do not show progressbar
cpus (int): number of CPUs to use for parallelisation (defaults to 1)
optim_mem (bool): whether to convert patterns to str to use less memory (defaults to False)
"""
self.algorithm = algorithm
inv = paradigms.inventory
if algorithm == "phon":
inv.init_dissimilarity_matrix(**kwargs)
self.cells = list(paradigms.cells_dedup)
self.optim_mem = optim_mem
tqdm.pandas(leave=False, disable=disable_tqdm)
log.info("Looking for analogical patterns...")
log.info(f"Using {cpus} threads for pattern inference.")
# Create empty dfs of patterns
# This is to avoid threads depending on a shared paradigms object !
# Note: inventory is NOT passed in tasks - workers get it via initializer
tasks = [(paradigms.get_empty_pattern_df(*pair), pair, algorithm, self.optim_mem)
for pair in self.get_pairs()]
if cpus > 1:
with ctx.Pool(cpus, initializer=_init_worker, initargs=(inv,)) as pool:
self.update(pool.starmap(patterns_job, tqdm(tasks)))
else:
self.update((_find_cellpair_patterns(inv, *task) for task in tqdm(tasks)))
def __repr__(self):
if len(self.cells) == 0:
return "ParadigmPatterns(empty)"
return f"ParadigmPatterns({', '.join([a + '~' + b for a, b in dict.keys(self)])})"
[docs]
def export(self, md, optim_mem=False, **kwargs):
"""
Export dataframes to a folder for later use.
Arguments:
optim_mem (bool): Whether to not export human readable patterns too. Defaults to False.
"""
# Save machine readable patterns
self.to_csv(md, **kwargs)
# Save human readable patterns
if optim_mem:
log.warning("Since you asked for args.optim_mem,"
"I will not export the human_readable file.")
elif not self.has_applicable:
rel_path = "patterns/human_readable/"
abs_path = md.get_path(rel_path)
log.info("Writing pretty patterns (for manual examination) to %s", abs_path)
for pair in self.keys():
self.to_md(md, pair, rel_path, abs_path)
return md.prefix
[docs]
def to_md(self, md, pair, rel_path, abs_path):
"""Export a Patterns DataFrame as a pretty markdown file
Arguments:
md (qumin.utils.Metadata): Metadata handler.
pair (Tuple[str, str]): pair of cells for which a pattern file should
be exported.
rel_path (str): Relative path to the folder.
abs_path (str): Absolute path to the folder.
"""
a, b = pair
name = f"pat_{self.algorithm}_{a}-{b}.md"
export_data = self[pair]
template = "\n## Pattern {p_str}\n\nPairs of forms instantiating this pattern: " \
"{n}\nFull pattern: {p_repr}\nExamples:\n\n{pair_table}\n"
with open(abs_path / name, "w", encoding="utf-8") as f:
grouped = export_data.groupby("pattern")
for pattern, group in sorted(grouped, key=lambda x: x[1].shape[0], reverse=True):
table = group[["lexeme", 'form_x', 'form_y']]
f.write(template.format(
n=group.shape[0],
p_str=str(pattern),
p_repr=repr(pattern),
pair_table=table.to_markdown(index=False, headers=["lexeme", a, b])
))
file_path = Path(rel_path) / name
md.register_file(file_path,
name="human_" + str(file_path.with_suffix('').name),
custom=dict(cells=pair,
kind="human_readable",
algorithm=self.algorithm),
description=f"Human-readable patterns between cells '{a}' and '{b}', "
f"with the '{self.algorithm}' algorithm.")
[docs]
def to_csv(self, md, rel_path="patterns/machine_readable/"):
"""Export a Patterns DataFrame to csv.
Arguments:
md (qumin.utils.Metadata): Metadata handler.
rel_path (str): Relative path to the folder.
"""
# Path settings
abs_path = md.get_path(rel_path)
rel_path = Path(rel_path)
log.info("Writing patterns (importable by other scripts) to %s", abs_path)
# Patterns map
pattern_list = set()
for pair in self:
patterns = self[pair]['pattern'].unique()
pattern_list.update([repr(pat) for pat in patterns])
pattern_map = {pat: n for n, pat in enumerate(pattern_list)}
s = pd.Series({n: pat for pat, n in pattern_map.items()}, name="patterns")
s.to_csv(abs_path / "patterns_map.csv")
md.register_file(rel_path / "patterns_map.csv", override=self.has_applicable)
# One csv per pair of cells
for (a, b) in self.keys():
name = f"pat_{self.algorithm}_{a}-{b}.csv"
export = self[(a, b)].copy()
if not isinstance(export.pattern.iloc[0], str):
export.pattern = export.pattern.map(repr)
# Replace forms by ids
export[['form_x', 'form_y']] = \
export[['form_x', 'form_y']].map(lambda x: x.id)
# Replace patterns by ids
export.pattern = export.pattern.map(pattern_map)
# Exporting applicable patterns if available
if self.has_applicable:
export['applicable_reverse'] = self[(b, a)].applicable
for col in ['applicable', 'applicable_reverse']:
notdef = export[col].notna()
export.loc[notdef, col] = export.loc[notdef, col]\
.apply(lambda x: ":".join((str(pattern_map[repr(y)]) for y in sorted(x))))
export.drop(["lexeme"], axis=1).to_csv(abs_path / name, sep=",", index=False)
md.register_file(rel_path / name,
override=self.has_applicable,
custom=dict(cells=(a, b),
kind="machine_readable",
algorithm=self.algorithm,
applicable=self.has_applicable),
description=f"Machine-readable patterns between cells '{a}' and '{b}', "
f"with the '{self.algorithm}' algorithm.")
[docs]
def from_file(self, patterns_md, paradigms, force=False, **kwargs):
"""Read pattern data from a previous export.
Arguments:
patterns_md (qumin.utils.Metadata): metadata handler from a previous run.
paradigms (qumin.representations.paradigms.Paradigms): Paradigms representation.
"""
cells = paradigms.cells
collection = defaultdict(lambda: defaultdict(str))
cells = paradigms.cells
# Read patterns map
patterns_map = pd.read_csv(patterns_md.get_resource_path('patterns_map'),
index_col=0).patterns
# Get available patterns and their path
pattern_files = {}
for r in patterns_md.package.resources:
if ('cells' in r.custom
and 'kind' in r.custom
and r.custom['kind'] == "machine_readable"):
pattern_files[tuple(r.custom['cells'])] = patterns_md.get_resource_path(r.name)
self.algorithm = r.custom['algorithm']
# Parse patterns for each pair of cells
log.info('Reading patterns...')
self.pairs = [p if p in pattern_files else p[::-1]
for p in combinations(cells, 2)] if cells else list(pattern_files)
first = True
for pair in tqdm(self.pairs):
if pair not in pattern_files.keys():
raise ValueError("Couldn't find patterns for the following cell: "
f"{pair[0]}~{pair[1]}. Check your patterns.")
self.from_csv(pattern_files[pair], pair, patterns_map, collection,
paradigms, **kwargs)
if first:
memory_check(list(self.values())[0], len(self.pairs), force=force)
first = False
[docs]
def from_csv(self, path, pair, patterns_map, collection,
paradigms, defective=True, overabundant=True,
cells=None):
"""
Read a patterns dataframe for a specific pair of cells
Arguments:
paradigms (qumin.representations.paradigms.Paradigms): Paradigms representation.
defective (bool): whether to consider defective lexemes.
overabundant (bool): whether to consider overabundance.
collection (defaultdict): a defaultdict to avoid recomputing
patterns from strings.
pair (tuple) a tuple of cells to read.
patterns_map (pandas.DataFrame): a DataFrame of patterns and pattern ids.
"""
def read_pattern(string):
"""
Reads patterns from string representations.
Arguments:
string (str): a string representation of a pattern.
"""
if string and not pd.isnull(string):
if string in collection[pair]:
result = collection[pair][string]
else:
pattern = Pattern.from_str(pair, string, paradigms.inventory)
collection[pair][string] = pattern
result = pattern
return result
if defective:
return None
else:
return np.nan
table = pd.read_csv(path, sep=",", dtype="str")
# Drop lexemes and forms that were filtered out (defectives, sampling, etc)
table.drop(table[~table.form_x.isin(paradigms.data.index)].index, inplace=True)
for cell in pair:
if cell not in self.cells:
self.cells.append(cell)
# Restore the patterns
table.pattern = table.pattern.astype('int').map(patterns_map).apply(read_pattern)
if not defective:
table.dropna(axis=0, subset="pattern", inplace=True)
self._check_missing_patterns(paradigms.data, table, pair)
# Restore phon_form based on paradigms and form_ids
table[['lexeme', 'form_x']] = pd.merge(table, paradigms.data, right_index=True,
left_on='form_x')[['lexeme', 'form']]
table['form_y'] = table.form_y.map(paradigms.data.form)
# Parse applicable patterns if there are some
if "applicable" in table.columns:
app_cols = ['applicable', 'applicable_reverse']
notdef = table[app_cols].notna().all(axis=1)
table.loc[notdef, app_cols] = table.loc[notdef, app_cols].map(
lambda x: tuple(read_pattern(patterns_map[int(y)])
for y in x.split(':')) if x is not None else x)
table.loc[~notdef, app_cols] = None
app_col = table.pop('applicable')
table.rename(columns={"applicable_reverse": "applicable"}, inplace=True)
self[pair[::-1]] = table.rename(columns={"form_x": "form_y", "form_y": "form_x"})
table.applicable = app_col
self.has_applicable=True
self[pair] = table
@staticmethod
def _check_missing_patterns(paradigms, table, pair):
"""
This is a security check that patterns do exist for all pairs of cells.
To reduce computation time, we first check that the number of pairs is correct.
If this fails, we check all pairs of cells one by one.
"""
# Keep only those lexemes that have a record for both cells.
lexeme_cellcounts = paradigms[paradigms.cell.isin(pair)]\
.groupby('lexeme', observed=False).cell.nunique()
paradigm_sample = paradigms.loc[paradigms.cell.isin(pair) &
paradigms.lexeme.isin(
lexeme_cellcounts[lexeme_cellcounts == 2].index),
['lexeme', 'cell']].reset_index()
a = paradigm_sample[paradigm_sample.cell == pair[0]]
b = paradigm_sample[paradigm_sample.cell == pair[1]]
if len(pd.merge(a, b, on="lexeme")) != len(table):
form_pairs = pd.merge(a, b, on="lexeme")
table_pairs = table[['form_x', 'form_y']].apply(tuple, axis=1).to_list()
missing = form_pairs[form_pairs.apply(lambda x: (x.form_id_x, x.form_id_y) not in table_pairs, axis=1)]
raise ValueError("It looks like your paradigms and your patterns are not aligned."
f"Check that you have patterns for all pairs, missing {pair}:"
f"{missing.to_markdown()}")
[docs]
def unmerge_columns(self, paradigms):
"""
Recreates merged columnss
Arguments:
paradigms (qumin.representations.Paradigms): a Paradigms object.
"""
# Variables to store asynchronous changes.
mapping = {i: k for k, v in paradigms.cells_dedup.items() for i in v}
new_pairs = []
# Find which pairs need to be restored
for cell, dupl_cells in paradigms.cells_dedup.items():
if dupl_cells == []:
continue
new_pairs.extend(
list(product(dupl_cells,
set(paradigms.cells) - set(dupl_cells) - {cell})))
# We create the new pairs and delete the old ones.
for pair in new_pairs:
old_pair = (mapping.get(pair[0], pair[0]),
mapping.get(pair[1], pair[1]))
if old_pair not in self.keys():
old_pair = old_pair[::-1]
pair = pair[::-1]
new_df = paradigms.get_empty_pattern_df(*pair)
cols = ["lexeme", "form_x", "form_y"]
self[pair] = pd.merge(new_df[cols], self[old_pair], on=cols, how="left")
# We create identity patterns
for cell, dupl_cells in paradigms.cells_dedup.items():
identical_cells = [cell] + dupl_cells
if len(identical_cells) < 2:
continue
for pair in combinations(identical_cells, 2):
self[pair] = paradigms.get_empty_pattern_df(*pair)
defective = (self[pair].form_x == "") | (self[pair].form_y == "")
self[pair].loc[~defective, 'pattern'] = Pattern.new_identity(pair, paradigms.inventory)
self[pair].loc[defective, 'pattern'] = None
self.cells = paradigms.cells
[docs]
def find_applicable(self, cpus=1, **kwargs):
"""Find all applicable rules for each form.
We name sets of applicable rules *classes*. *Classes* are oriented:
we produce two separate columns (a, b) and (b, a)
for each pair of columns (a, b) in the paradigm..
Returns:
:class:`pandas:pandas.DataFrame`:
associating a lemma (index)
and an ordered pair of paradigm cells (columns)
to a tuple representing a class of applicable patterns.
"""
log.info("Looking for classes of applicable patterns")
# Adding oriented patterns
col_rename = {'form_x': 'form_y', 'form_y': 'form_x'}
to_add = {}
for key, value in self.items():
to_add[key[::-1]] = self[key].rename(columns=col_rename)
self.update(to_add)
log.info("total cpus: " + str(cpus))
tasks = [(pair, self[pair]) for pair in self]
if cpus > 1:
with ctx.Pool(cpus) as pool: # Create a multiprocessing Pool
for pair, applicables in tqdm(pool.starmap(_find_cellpair_applicable, tasks), total=len(self)):
df = self[pair]
# We're trying to avoid any pandas internal issues in merging the df/series
# First, we create a new column in the df...
df.loc[:, "applicable"] = None
# Then we assign the returned series, which is already called "applicable":
df.loc[applicables.index, "applicable"] = applicables
else:
for task in tqdm(tasks):
pair, applicables = _find_cellpair_applicable(*task)
df = self[pair]
df.loc[:, "applicable"] = None
df.loc[applicables.index, "applicable"] = applicables
self.has_applicable=True
[docs]
def to_incidence_table(self, weighted=False,
lexemes=None, flatten_columns=False):
"""
By default, creates a onehot encoding of the lexemes (the features are the patterns).
If `weighted=True`, the produced table weights overabundant patterns with their
relative frequency (it is no more a onehot encoding, but it can be useful in
various situations).
If you use the ``weighting`` option, please ensure you previously
added frequencies to your patterns with :func:`add_frequencies`.
This is needed even if using a uniform distribution, to initialize it.
Arguments:
lexemes (iterable): lexemes to consider, None defaults to all.
weighted (bool): Whether to weight overabundant patterns by their frequency.
flatten_columns (bool): Whether to flatten the multiindex.
Example:
========== ============ =========== =========== ===========
Lexeme (A, B), pat1 (A,B), pat2 (A,C), pat3 ...
========== ============ =========== =========== ===========
lexeme_1 1 0 1 ...
lexeme_2 0 1 1 ...
========== ============ =========== =========== ===========
Returns:
:class:`pandas.DataFrame`:
associating lexemes (rows) with patterns(columns)
"""
log.info('Creating an incidence table of lexemes/patterns...')
if weighted:
if 'f_pair' not in self[list(self)[0]].columns:
raise ValueError(
'To compute an incidence table with weights, you first have to '
'add frequencies to your patterns:\n'
'\t>>> patterns.add_frequencies()')
def pattern_to_df(df, weighted=False):
if not weighted:
# convert to dummies,
# reduce overabundant index,
# and finally concatenate
return (
df.set_index('lexeme')
.pattern
.pipe(pd.get_dummies)
.reset_index()
.groupby('lexeme', observed=True)
.any()
)
else:
return (
df.groupby(['lexeme', 'pattern'], observed=True)
.f_pair.sum().unstack()
)
table = pd.concat(
{cell_pair: pattern_to_df(df[df.lexeme.isin(lexemes)] if lexemes else df, weighted=weighted)
for cell_pair, df in tqdm(self.items(), total=len(self))}, axis=1)
table.fillna(0, inplace=True)
if not weighted:
table = table.astype(int)
def _flatten_columns(df):
df.columns = df.columns.to_flat_index().map(
lambda x: f'{x[0]}⇌{x[1]}=<{x[2]}>'
)
return df
if flatten_columns:
return _flatten_columns(table)
return table
[docs]
def to_microclasses(self, incidence_table, frequencies=None):
"""
Computes microclasses from an incidence table.
Arguments:
onehot_encoding (pd.DataFrame): a onehot_encoding or an incidence table.
frequencies (qumin.representations.frequencies.Frequencies): an optional frequencies object
(typically accessed through `paradigms.frequencies`) to sort the lexemes.
Results:
dict[Str, List]: a dictionary mapping a label to all lexemes of the microclass.
"""
if frequencies:
freq_dict = frequencies.lexemes.value.to_dict()
key = lambda x: x.map(freq_dict)
else:
key = None
grp = incidence_table.sort_index(key=key).groupby(list(incidence_table.columns))
return {group.index[0]: group.index.tolist() for _, group in grp}
[docs]
def to_lattice(self, frequencies=None, **kwargs):
"""
Creates a Lattice from a patterns object.
Arguments:
patterns (qumin.representations.patternstore.PatternStore): the patterns computed for the paradigms.
frequencies (qumin.representations.frequencies.Frequencies): Will be used
to sort microclasses.
kwargs (dict): Optional keyword arguments will be passed to `ICLattice.__init__()`
Results:
qumin.lattice.lattice.ICLattice: An inflection class lattice.
"""
log.info("Building the lattice...")
incidence_table = self.to_incidence_table(flatten_columns=True)
microclasses = self.to_microclasses(incidence_table, frequencies=frequencies)
return ICLattice(incidence_table, microclasses, **kwargs)
def _find_cellpair_applicable(pair, df):
""" Find applicable patterns for a single cell pair.
Args:
pair (tuple of str): pair of cells
df (pd.DataFrame): patterns dataframe for that pair
Returns:
(pair, df): pair of cell and dataframe of applicable patterns.
"""
def applicable(form):
""" Return a tuple of all applicable patterns for a given form"""
return tuple((p for p in available_patterns if p.applicable(form, cell_x)))
available_patterns = [p for p in df['pattern'].unique() if p is not None]
cell_x = pair[0]
has_pat = ~df['pattern'].isnull()
applicables = df.loc[has_pat, "form_x"].apply(applicable)
applicables.name = "applicable"
return (pair, applicables)
def _find_cellpair_patterns(inventory, df, pair, method, optim_mem):
"""
Finds patterns for a pair of cells and returns a Dataframe containing the patterns.
"""
def _score(p):
"""Scores each pattern"""
def test_all(row, p):
"""Tests each pattern against each form"""
correct = 0
lex, a, b = row.lexeme, row.form_x, row.form_y
if a and b:
if (lex, a, b) in p.lexemes:
correct += 1
else:
B = p.apply(a, pair, inventory, raiseOnFail=False)
A = p.apply(b, pair[::-1], inventory, raiseOnFail=False)
if (B == b) and (A == a):
correct += 1
p.lexemes.add((lex, a, b))
return correct
counts = df.apply(test_all, axis=1, p=p)
return counts.sum()
def _attribute_best_pattern(sorted_collection, lex, a, b):
for p in sorted_collection:
if (lex, a, b) in p.lexemes:
return p
collection = defaultdict(lambda: defaultdict(list))
if method == "edits":
insert_cost = alignment.edits_ins_cost
sub_cost = alignment.edits_sub_cost
if method == "phon":
insert_cost = inventory.insert_cost
sub_cost = inventory.sub_cost
# Identify pairwise alternations
df.apply(_generate_rules,
axis=1,
inv=inventory,
pair=pair,
collection=collection,
insert_cost=insert_cost,
sub_cost=sub_cost)
sorted_collection = []
for alt in collection:
log.debug("\n\n####Considering alt:" + str(alt))
# Attempt to generalize
types = list(collection[alt])
pats = []
log.debug("1.Generalizing in each type")
for alt_type in collection[alt]:
log.debug(
f"\tType {alt_type}, found {len(collection[alt][alt_type])} patterns:\n\t {collection[alt][alt_type]}")
if _compatible_context_type(alt_type):
collection[alt][alt_type] = [generalize_patterns(collection[alt][alt_type], inventory)]
else:
collection[alt][alt_type] = incremental_generalize_patterns(collection[alt][alt_type], inventory)
pats.extend(collection[alt][alt_type])
log.debug(f"Resulted in {len(collection[alt][alt_type])} patterns : {collection[alt][alt_type]}")
log.debug(f"2.Generalizing across types {types}")
if _compatible_context_type(*types):
collection[alt] = [generalize_patterns(pats, inventory)]
else:
collection[alt] = incremental_generalize_patterns(pats, inventory)
log.debug(f"Resulted in {len(collection[alt])} patterns : {collection[alt]}")
# Score
for p in collection[alt]:
p.score = _score(p)
sorted_collection.append(p)
# Sort by score and choose best patterns
sorted_collection = sorted(sorted_collection, key=lambda x: x.score, reverse=True)
def _best_pattern(row):
lex, a, b = row.lexeme, row.form_x, row.form_y
if a != '' and b != '':
return _attribute_best_pattern(sorted_collection, lex, a, b)
return None
df['pattern'] = df.apply(_best_pattern, axis=1)
if optim_mem:
df.pattern = df.pattern.apply(repr)
return (pair, df)
def _generate_rules(row, inv, pair, collection, insert_cost, sub_cost):
"""
Generates the patterns for each pair of forms.
Arguments:
row (pandas.Series): a dataframe row containing the two forms.
inv (Inventory): a sound inventory
pair: a pair of cells
collection (defaultdict): the patterns collection
insert_cost: insertion cost to pass to alignment
sub_cost: substitution cost to pass to alignment
"""
lex, a, b = row.lexeme, row.form_x, row.form_y
if a and b:
if a == b:
new_rule = Pattern.from_aligned(pair, zip(a.tokens, b.tokens), inv)
new_rule.lexemes = {(lex, a, b)}
alt = new_rule.to_alt(inv, exhaustive_blanks=False)
t = _get_pattern_matchtype(new_rule, pair[0], pair[1])
collection[alt][t].append(new_rule)
else:
done = []
log.debug("All alignments of {}, {}".format(a, b))
for aligned in alignment.align_auto(a.tokens, b.tokens, insert_cost, sub_cost):
log.debug(aligned)
new_rule = Pattern.from_aligned(pair, aligned, inv)
new_rule.lexemes = {(lex, a, b)}
log.debug("pattern: " + str(new_rule))
if str(new_rule) not in done:
done.append(str(new_rule))
if new_rule._gen_alt:
alt = new_rule.to_alt(inv, exhaustive_blanks=False, use_gen=True)
log.debug("gen alt: " + str(alt))
else:
alt = new_rule.to_alt(inv, exhaustive_blanks=False)
t = _get_pattern_matchtype(new_rule, pair[0], pair[1])
collection[alt][t].append(new_rule)
def _compatible_context_type(*args):
"""Returns whether several contexts with these types can be merged without verification."""
compat = []
greedyany = {"greedy", "any"}
for positions in zip(*args):
for types in zip(*positions):
if "precise" in types or "lazy" in types:
return False
settypes = set(types)
if len(settypes) == 1 or settypes == greedyany:
compat.append(True)
else:
compat.append(False)
return all(compat)
def _get_pattern_matchtype(p, c1, c2):
"""Determine a type of match for an elementary pattern"""
type_names = {(True, True): "precise",
(True, False): "greedy",
(False, True): "lazy",
(False, False): "any"}
match_type1 = []
match_type2 = []
for i, group in enumerate(p.context):
match_type1.append([False, False])
match_type2.append([False, False])
# print("i,group:",i,group)
if group.blank:
ctxt = repr(group)
# print("left context:",ctxt)
alt1 = "".join(str(s) for s in p.alternation[c1][i])
alt2 = "".join(str(s) for s in p.alternation[c2][i])
match_type1[i][0] = alt1 != "" and alt1 in ctxt
# print("is",alt1,"in the left context ?",match_type1[i][0])
match_type2[i][0] = alt2 != "" and alt2 in ctxt
# print("is",alt2,"in the left context ?",match_type1[i][0])
if i + 1 < len(p.context):
ctxt2 = repr(p.context[i + 1])
# print("right context:",ctxt2)
match_type1[i][1] = alt1 != "" and alt1 in ctxt2
# print("is",alt1,"in the right context",ctxt2," ?",match_type1[i][1])
match_type2[i][1] = alt2 != "" and alt2 in ctxt2
# print("is",alt2,"in the right context",ctxt2,"?",match_type1[i][1])
return tuple(type_names[tuple(x)] for x in match_type1), tuple(type_names[tuple(x)] for x in match_type2)