brain_loop_search.packing

  1"""
  2This module contains the classes that can be used to commit any task of packing a graph, i.e. reduce the undesired
  3vertices based on a set of rules and an ontology tree that describe the affiliation among the vertices.
  4"""
  5import typing
  6
  7import numpy as np
  8import pandas as pd
  9
 10
 11class Ontology:
 12    """
 13    The prior knowledge about tree relationship among the vertices in graph, namely ontology, used for packing graph
 14    vertices.
 15This is a meta class that doesn't implement any function, so any subclass this with the same functions can use it
 16    with other modules smoothly.
 17
 18    I used this as an abstract interface and make derivation for ccfv3 ontology in brain.
 19    You can follow my subclass for your own.
 20
 21    I would recommend using pandas for tree representation for its speed and also because you can save some useful info
 22    as appended columns.
 23    """
 24
 25    def __init__(self):
 26        """
 27        Here, some preprocessing can be done make other functions easier and faster to operate. Also, you need to
 28        ensure that this is actually a tree.
 29        """
 30        pass
 31
 32    def check_include(self, vert: typing.Iterable):
 33        """
 34        check if the vertices given is included
 35        :param vert: the vertices to check
 36        :return: True or False.
 37        """
 38        pass
 39
 40    def levels_of(self, vert: typing.Iterable) -> pd.Series:
 41        """
 42        Level is defined as the largest count from the current node to its leaves in the tree, starting from 0.
 43        :param vert: the vertices to check
 44        :return: the levels of each vertex in the ontology
 45        """
 46        pass
 47
 48    def depths_of(self, vert: typing.Iterable) -> pd.Series:
 49        """
 50        Depth is defined as the count from the current node to its root in the tree, starting from 0.
 51        :param vert: the vertices to check
 52        :return: the depths of each vertex in the ontology
 53        """
 54        pass
 55
 56    def ancestors_of(self, vert: typing.Iterable) -> pd.Series:
 57        """
 58        The parents of each vertex as a list, starting from the tree root to its immediate parent, the length of which
 59        equals the depth.
 60
 61        :param vert: the vertices to check
 62        :return: the lists of parents of each vertex in the ontology
 63        """
 64        pass
 65
 66    def immediate_children_of(self, vert: typing.Iterable) -> pd.Series:
 67        """
 68        The immediate children of each vertex as a list, starting from the tree root to its direct parent.
 69
 70        :param vert: the vertices to check
 71        :return: the lists of children of each vertex in the ontology
 72        """
 73
 74
 75class VertexPacker:
 76    """
 77    Given a set of vertices in a graph, e.g. brain structures in a macroscopic brain connectome,
 78    rearrange them by filtering and merging into a new set, based on a filtering rule and according
 79    to an ontology.
 80
 81    This is the precursor step before packing the whole graph. A directed graph may not share its rows
 82    with columns, so this might need to be done once for each, even with different rules.
 83
 84    Also, for a complexed graph, where you may expect heterogeneous level hierarchies, you can pack the graph multiple
 85    times, and their vertices apply different rules naturally. This way, you'll also utilize this decoupled facility.
 86
 87    This class uses a state machine functionality, by which each time you trigger the alteration each the
 88    stash changes. When you need a diversion or something you can retrieve the intermediate result and start anew.
 89    """
 90
 91    def __init__(self, vertices: typing.Iterable, ontology: Ontology):
 92        """
 93        Initialize the vertex packer with an initial set of vertices and ontology. It also checks the viability of
 94        the vertices.
 95
 96        :param vertices: a set of entities in the ontology to filter and merge.
 97        :param ontology: a derivation of the abstract class `Ontology`.
 98        """
 99        vertices = pd.Series(vertices)
100        assert ontology.check_include(vertices), f"vertices contain unrecognizable ID "
101        self._vert = vertices
102        self._ont = ontology
103
104    def stash(self):
105        """
106        For retrieving the intermediate or final results. To remove mutability, it's turned to tuple. If you need a
107        diversion of processing, you can retrieve the intermediate result and make a new pathway.
108
109        :return: the stashed vertices' copy as a numpy array.
110        """
111        return self._vert.copy()
112
113    def filter_by_level(self, fro: int = None, to: int = None, invert=False):
114        """
115        Only retain the vertices within the level range, c style.
116
117        Levels are the max count from the current node to its leaves in the tree, staring from 0.
118
119        :param fro: the starting level, inclusive, default as None, meaning no limit.
120        :param to: the ending level, exclusive, default as None, meaning no limit.
121        :param invert: whether retain the otherwise non-descendents
122        """
123        levels = self._ont.levels_of(self._vert)
124        if fro is not None:
125            levels = levels[levels >= fro]
126        if to is not None:
127            levels = levels[levels < to]
128        if invert:
129            self._vert = self._vert[~self._vert.isin(levels.index)]
130        else:
131            self._vert = levels.index
132
133    def filter_by_depth(self, fro: int = None, to: int = None, invert=False):
134        """
135        Only retain the vertices within the depth range, c style.
136
137        Depths are the count from the current node to its root in the tree, staring from 0.
138
139        :param fro: the starting depth, inclusive, default as None, meaning no limit.
140        :param to: the ending depth, exclusive, default as None, meaning no limit.
141        :param invert: whether retain the otherwise non-descendents
142        """
143        depths = self._ont.depths_of(self._vert)
144        if fro is not None:
145            depths = depths[depths >= fro]
146        if to is not None:
147            depths = depths[depths < to]
148        if invert:
149            self._vert = self._vert[~self._vert.isin(depths.index)]
150        else:
151            self._vert = depths.index
152
153    def filter_super(self):
154        """
155        Remove any vertices that happen to be the ancestor of another.
156        """
157        un = set.union(*map(set, self._ont.ancestors_of(self._vert)))
158        self._vert = self._vert[~self._vert.isin(un)]
159
160    def filter_sub(self):
161        """
162        Remove any vertices that happen to be the descendent of another.
163        """
164        vert = set(self._vert)
165        tf = [vert.isdisjoint(i) for i in self._ont.ancestors_of(self._vert)]
166        self._vert = self._vert[tf]
167
168    def filter_by_immediate_child_of(self, parents: typing.Iterable, invert=False):
169        """
170        Only retain the direct children under some vertices, which is convenient when you have a big super node with
171        many branches below.
172
173        :param parents: the direct parent vertices.
174        :param invert: whether retain the otherwise non-descendents
175        """
176        un = set.union(*map(set, self._ont.immediate_children_of(parents)))
177        if invert:
178            self._vert = self._vert[~self._vert.isin(un)]
179        else:
180            self._vert = self._vert[self._vert.isin(un)]
181
182    def filter_by_descendants_of(self, parents: typing.Iterable, include_parents=False, invert=False):
183        """
184        Only retain the descendents of some vertices.
185
186        :param parents: the ancestors.
187        :param include_parents: whether to allow the parents to exist in the result, default as not.
188        :param invert: whether retain the otherwise non-descendents
189        """
190        parents = set(parents)
191        tf = map(lambda p: not parents.isdisjoint(p), self._ont.ancestors_of(self._vert))
192        if include_parents:
193            tf = [*tf] | self._vert.isin(parents)
194        if invert:
195            tf = [not i for i in tf]
196        self._vert = self._vert[tf]
197
198    def merge_by_level(self, thr):
199        """
200        Merge the vertices until they are all above the min level or no merge can be done.
201
202        Levels are the max count from the current node to its leaves in the tree, staring from 0.
203
204        :param thr: the min level.
205        """
206        vert = set()
207        for i, p, l in zip(self._vert, self._ont.ancestors_of(self._vert), self._ont.levels_of(self._vert)):
208            if l < thr:
209                vert.add(p[l - thr])
210            else:
211                vert.add(i)
212        self._vert = pd.Series(list(vert))
213
214    def merge_by_depth(self, thr):
215        """
216        Merge the vertice until they are all below the max depth or no merge can be done.
217
218        Depths are the count from the current node to its root in the tree, staring from 0.
219
220        :param thr: the max depth.
221        """
222        vert = set()
223        for i, p in zip(self._vert, self._ont.ancestors_of(self._vert)):
224            if len(p) > thr:
225                vert.add(p[thr])
226            else:
227                vert.add(i)
228        self._vert = pd.Series(list(vert))
229
230
231class GraphPacker:
232    """
233    Given a directed graph (in adjacent matrix form), according to an ontology,
234    rearrange as a new graph with superior structures or just merge redundant edges. The projection is from
235    rows to columns
236
237    Taking brain graph as an example, a group of neurons project from regions to regions,
238    meaning edge (a, b) can have redundant occurrences with different weights. This is where merging needs to be done.
239
240    Sometimes, you want to look into their coarser regions' relations rather than the finer ones. This is where
241    rearrangement needs to be done.
242    """
243
244    def __init__(self, adj_mat: pd.DataFrame, ontology: Ontology):
245        """
246        :param adj_mat: the adjacent matrix in pandas dataframe, the projection is from rows to columns.
247        :param ontology: a derivation of the abstract class `Ontology`.
248        """
249        # check adjacent matrix
250        assert ontology.check_include(adj_mat.index), f"rows contain unrecognizable ID"
251        assert ontology.check_include(adj_mat.columns), f"columns contain unrecognizable ID"
252        self._mat = adj_mat.to_numpy()
253        self._rows = adj_mat.index
254        self._cols = adj_mat.columns
255        self._ont = ontology
256
257    def pack(self, new_rows: typing.Iterable, new_cols: typing.Iterable, def_val: float = -1,
258             superior_as_complement: bool = False, aggr_func=lambda x: 1 / (1 / x).sum()):
259        """
260        Specifying new rows and columns (usually generated by feeding the original to VertexPacker, but you can
261        provide your own list, long as it's within the ontology),
262        retrieve a new graph in adjacent matrix.
263
264        Packing usually involves merging redundant edges or edges inferior to a superior one. Sometimes, conflicts can
265        take place where the superior edge and inferior edge coexist. Normally this is an undefined behavior, but
266        considering in some ontology assignment, areas where the inferior don't cover in the superior, namely,
267        for set A containing set B and set C, A - (B|C) can be represented using the superior. So here, by turning this
268        on, you can always merge them. Otherwise, merge is done only for mutually excluded vertices.
269
270        :param new_rows: the new 'project from' vertices.
271        :param new_cols: the new 'project to' vertices.
272        :param def_val: the default value for non-connecting edges.
273        :param superior_as_complement: do merge for both superior and inferior vertices when encountered,
274            default as False.
275        :param aggr_func: the aggregation function.
276        :return: a new adjacent matrix as pandas dataframe.
277        """
278        new_rows = pd.Series(new_rows)
279        new_cols = pd.Series(new_cols)
280        assert new_rows.is_unique, "new row items are not unique"
281        assert new_cols.is_unique, "new column items are not unique"
282
283        new_mat = np.ones([len(new_rows), len(new_cols)]) * def_val
284
285        def vertex_map_gen(all_vert, new_vert):
286            # prefilter
287            packer = VertexPacker(all_vert.unique(), self._ont)
288            packer.filter_by_descendants_of(new_vert, include_parents=True)
289            needed_vert = packer.stash()
290            # map
291            map = []
292            for v in new_vert:
293                packer = VertexPacker(needed_vert, self._ont)
294                packer.filter_by_descendants_of([v], include_parents=True)
295                nodes = packer.stash()
296                if not superior_as_complement:  # filter the sub nodes in this mode
297                    packer = VertexPacker(nodes, self._ont)
298                    packer.filter_sub()
299                    nodes = packer.stash()
300                map.append(np.where(all_vert.isin(nodes)))
301            return map
302
303        row_map = vertex_map_gen(self._rows, new_rows)
304        col_map = vertex_map_gen(self._cols, new_cols)
305
306        # calculate the edges in the new matrix
307        for i, (row, fro_ind) in enumerate(zip(new_rows, row_map)):
308            if len(fro_ind) == 0:
309                continue
310            for j, (col, to_ind) in enumerate(zip(new_cols, col_map)):
311                if row == col or len(to_ind) == 0:
312                    continue
313                dat = self._mat[fro_ind][:,to_ind].reshape(-1)
314                dat = dat[dat != def_val]
315                if len(dat) == 0:
316                    continue
317                new_mat[i, j] = aggr_func(dat)
318        return pd.DataFrame(new_mat, index=new_rows, columns=new_cols)