#!/usr/bin/env python3
import operator
from collections.abc import Iterable
from functools import reduce
import networkx as nx
import numpy as np
import pandas as pd
from fosf.syntax.base import DisjunctiveSort, FrozenDisjunctiveSort, Sort
from fosf.syntax.taxonomy import FuzzySortTaxonomy, SortTaxonomy
from fosf.utils.graph import add_or_find_bot_and_top, graph_to_dag
class SimilarityClass:
def __init__(self, graph: nx.Graph, class_: set[Sort]):
# A similarity class is represented as a matrix
self.n = len(class_)
# TODO: since the matrix is always symmetric, is there a way to save space?
self.matrix = np.zeros((self.n, self.n))
self.index: dict[Sort, int] = {node: i for i, node in enumerate(class_)}
self._preprocess(graph)
def _preprocess(self, graph):
# transitive closure of the similarity
# Initialize matrix with values according to the edges
for s, si in self.index.items():
self.matrix[si, si] = 1 # reflexivity
for t in graph[s]:
if t not in self.index:
continue
w = graph[s][t]["weight"]
ti = self.index[t]
self.matrix[si, ti] = w
self.matrix[ti, si] = w
# Close matrix with Warshall's algorithm
for k in range(self.n):
for i in range(self.n):
for j in range(self.n):
self.matrix[i, j] = max(
self.matrix[i, j], min(self.matrix[i, k], self.matrix[k, j])
)
def __getitem__(self, pair: tuple[Sort, Sort]):
# Returns the similarity of a pair of nodes
return self.matrix[self.index[pair[0]], self.index[pair[1]]]
def __iter__(self):
# iterate over nodes inside this similarity class
return iter(self.index.keys())
def __str__(self):
# String representation of the similarity class
# Using pandas for convenience for the moment
# TODO: SimilarityClass __str__ method
ind = sorted(self.index.keys(), key=lambda x: self.index[x])
df = pd.DataFrame(data=self.matrix, index=ind, columns=ind)
return str(df)
def __repr__(self):
# TODO: correctly implement __repr__ for SimilarityClass
return str(self)
class Similarity:
def __init__(self, graph: nx.Graph, alpha: float = 0.0, strict: bool = False):
self.graph = graph # TODO: make a copy?
# map node to representative
self.rep: dict[Sort, Sort] = {}
# map rep node to similarity class
self._sim_class: dict[Sort, SimilarityClass] = {}
# TODO: also store minimum degree of each sim class?
self._preprocess(alpha, strict)
def _preprocess(self, alpha, strict):
def _iter(s, alpha, strict):
# Yield each non-yet-visited node t adjacent to s if the weight of the edge
# is >= alpha (strict=False) or > alpha (strict=True)
for node in self.graph[s]:
if node not in self.rep:
if strict:
if self.graph[s][node]["weight"] > alpha:
yield node
else:
if self.graph[s][node]["weight"] >= alpha:
yield node
# Compute similarity classes
# Map each node to a representative of the similarity class
# map each representative to the SimilarityClass
for s in self.graph.nodes():
if s in self.rep:
# s is already visited
continue
self.rep[s] = s
class_ = {s}
# iterative DFS to find similar nodes
stack = [_iter(s, alpha, strict)]
while stack:
nodes = stack[-1]
node = next(nodes, None)
if node is None:
stack.pop()
continue
if node in self.rep:
continue
self.rep[node] = s
class_.add(node)
stack.append(_iter(node, alpha, strict))
self._sim_class[s] = SimilarityClass(self.graph, class_)
def sim_class(self, s: Sort):
"Return the similarity class of a node."
rep = self.rep.get(s, s)
return self._sim_class[rep]
def degree(self, *sorts: Sort):
"Return the similarity degree among sorts."
if len(sorts) < 2:
return 1.0 # reflexivity
reps = {self.rep.get(s, s) for s in sorts}
if len(reps) > 1:
# Different similarity classes
return 0.0
sim_class = self.sim_class(sorts[0])
return min(sim_class[(sorts[0], s)] for s in sorts[1:])
def similar_to(self, s: Sort):
"Generate nodes similar to a node together with similarity degree."
for t in self.sim_class(s):
if t != s:
yield t, self.degree(s, t)
[docs]
class SubsumptionSimilarityTaxonomy(FuzzySortTaxonomy):
def __init__(
self,
subsumption: nx.DiGraph | SortTaxonomy,
similarity: nx.Graph | Similarity,
instances: dict[str, dict[Sort, float]] | None = None,
keep_base: bool = False,
):
"Combine a (fuzzy) taxonomy and a similarity relation into a fuzzy taxonomy."
self._similarity: Similarity
self._is_quotiented: bool = False
self._rep: dict[Sort, Sort] = {}
self._classes: dict[Sort, Sort] = {}
if keep_base:
if isinstance(subsumption, SortTaxonomy):
self._base_taxonomy = subsumption
else:
self._base_taxonomy = SortTaxonomy(subsumption.edges(data=True))
self._similarity, edges, self._rep, self._classes = (
self._combine_subsumption_and_similarity(subsumption, similarity)
)
super().__init__(edges, instances)
# TODO: self._rep and instances
if self._is_quotiented:
self.top = self._rep[self.top]
self.bot = self._rep[self.bot]
# TODO: method to add instances
[docs]
def glb(self, *nodes):
if not self._is_quotiented:
glb = super().glb(*nodes)
if glb == self.bot:
# It might happen that the 'MLBs' are minimal similar sorts
# {s0, .., sn} (minimal_inputs)
# hence glb = bot
minimal_inputs = set(nodes).intersection(self.graph[self.bot])
if not minimal_inputs:
return self.bot
minimal_inputs_code = self.code(minimal_inputs)
for sort in nodes:
# Check if every input sort is a supersort of some s_i
if minimal_inputs_code & self.code(sort) == 0:
# if not, glb is actually bot
return self.bot
# otherwise, all input sorts are supersorts of a minimal set
# minimal_inputs = {s0, .., sn}
# If all of these are similar, we can return DisjunctiveSort(s0, ..., sn)
if self._similarity.degree(*minimal_inputs) > 0:
return FrozenDisjunctiveSort(*minimal_inputs)
return glb
nodes = [self._rep[node] for node in nodes]
glb = super().glb(*nodes)
if not isinstance(glb, (DisjunctiveSort, FrozenDisjunctiveSort)):
# GLB is ordinary sort
# TODO: what if glb is bot and sorts are similar?
return self._classes[self._rep[glb]]
# GLB is a (Frozen)DisjunctiveSort
sorts = set()
for mlb in glb:
class_ = self._classes[self._rep[mlb]]
if isinstance(class_, (DisjunctiveSort, FrozenDisjunctiveSort)):
sorts.update(class_)
else:
sorts.add(class_)
return DisjunctiveSort(*sorts)
[docs]
def code(self, node: Sort | Iterable[Sort]) -> int:
"""
Return the code associated to a node or set of nodes.
Parameters
----------
node : T | Iterable[T]
The node or nodes for which to return a code. If an iterable of nodes is
passed, the code is the bitwise OR of the codes of each node in the iterable.
"""
if not self._is_quotiented:
return super().code(node)
if isinstance(
node, (set, self._FROZEN_DISJUNCTIVE_TYPE, self._DISJUNCTIVE_TYPE)
):
return reduce(operator.or_, (self.code(n) for n in node))
return self.node_to_code[self._rep[node]]
[docs]
def is_subsort(self, s: Sort, t: Sort) -> bool:
"""
Check if a sort is subsumed by another sort.
Parameters
----------
s : Sort
A :class:`Sort`
t : Sort
A :class:`Sort`
"""
if self._is_quotiented:
s = self._rep[s]
t = self._rep[t]
return super().is_subsort(s, t)
[docs]
def degree(self, s: Sort | Iterable[Sort], t: Sort | Iterable[Sort]):
if not self._is_quotiented:
return super().degree(s, t)
args = []
for arg in [s, t]:
if isinstance(arg, DisjunctiveSort):
arg = DisjunctiveSort(*(self._rep[u] for u in arg))
elif isinstance(arg, FrozenDisjunctiveSort):
arg = FrozenDisjunctiveSort(*(self._rep[u] for u in arg))
elif isinstance(arg, Sort):
arg = self._rep[arg]
elif isinstance(arg, Iterable):
arg = [self._rep[Sort(u)] for u in arg]
args.append(arg)
return super().degree(args[0], args[1])
def _combine_subsumption_and_similarity(self, subsumption, similarity):
if isinstance(subsumption, SortTaxonomy):
subsumption = subsumption.graph
# Initialize new fuzzy taxonomy
# Take a copy of the graph, adding weights if necessary
graph = nx.DiGraph()
for s, t, data in subsumption.edges(data=True):
w = data["weight"] if "weight" in data else 1.0
graph.add_edge(s, t, weight=w)
bot, top = add_or_find_bot_and_top(graph, node_type=Sort)
# Initialize similarity
if isinstance(similarity, Similarity):
sim = similarity
elif isinstance(similarity, nx.Graph):
similarity = similarity.copy()
similarity.add_nodes_from(graph.nodes()) # TODO: essential?
sim = Similarity(similarity)
else:
TypeError("sim_graph must be an nx.Graph or a Similarity")
# TODO: what if sim_graph/digraph is missing nodes from digraph/sim_graph?
# Gather new edges combining (fuzzy) taxonomy + similarity
new_edges = []
# if s < t ~ u, add weighted edge from s to u
# if s ~ t < u, add weighted edge from s to u
for s in sim.rep:
for t, alpha in sim.similar_to(s):
for v in graph.pred[s]:
if (v, t) not in graph.edges() or alpha > graph[v][t]["weight"]:
new_edges.append((v, t, alpha))
for v in graph[s]:
if (t, v) not in graph.edges() or alpha > graph[t][v]["weight"]:
new_edges.append((t, v, alpha))
for v in graph.pred[t]:
if (v, s) not in graph.edges() or alpha > graph[v][s]["weight"]:
new_edges.append((v, s, alpha))
for v in graph[t]:
if (s, v) not in graph.edges() or alpha > graph[s][v]["weight"]:
new_edges.append((s, v, alpha))
# Add new edges to graph
for u, v, alpha in new_edges:
if u == bot or v == top:
continue
current_weight = graph[u][v]["weight"] if v in graph[u] else 0
new_weight = max(alpha, current_weight)
# Did the edge exist in the original graph? We keep that edge
if v in subsumption[u]:
original_weight = subsumption[u][v]["weight"]
new_weight = max(original_weight, new_weight)
graph.add_edge(u, v, weight=new_weight)
rep = {}
classes = {}
if not nx.is_directed_acyclic_graph(graph):
graph, rep, classes = graph_to_dag(graph)
self._is_quotiented = True
# Remove redundant edges from bot/top
for node in graph.pred[top].copy():
if len(graph[node]) > 1:
graph.remove_edge(node, top)
for node in graph[bot].copy():
if len(graph.pred[node]) > 1:
graph.remove_edge(bot, node)
# Restore correct weights for bottom and top
for node in graph[bot]:
graph[bot][node]["weight"] = 1
for node in graph.pred[top]:
graph[node][top]["weight"] = 1
return sim, graph.edges(data=True), rep, classes