From a175e2d70db4de140595667caae326afe433ddf8 Mon Sep 17 00:00:00 2001 From: Akhilesh Halageri Date: Sun, 16 Oct 2022 18:06:50 +0000 Subject: [PATCH 01/28] feat: add debug functions --- pychunkedgraph/debug/edits.py | 58 ++++++++++++++++++++++++++--------- pychunkedgraph/debug/utils.py | 43 ++++++++++++++++++++++++++ 2 files changed, 87 insertions(+), 14 deletions(-) create mode 100644 pychunkedgraph/debug/utils.py diff --git a/pychunkedgraph/debug/edits.py b/pychunkedgraph/debug/edits.py index 11ea82676..4773bc912 100644 --- a/pychunkedgraph/debug/edits.py +++ b/pychunkedgraph/debug/edits.py @@ -1,4 +1,5 @@ from typing import Union +from typing import Tuple import numpy as np @@ -13,9 +14,33 @@ def _parse_merge_payload( - cg: ChunkedGraph, user_id: str, payload: dict + cg: ChunkedGraph, user_id: str, payload: list ) -> MergeOperation: - pass + + node_ids = [] + coords = [] + for node in payload: + node_ids.append(node[0]) + coords.append(np.array(node[1:]) / cg.segmentation_resolution) + + atomic_edge = handle_supervoxel_id_lookup(cg, coords, node_ids) + chunk_coord_delta = cg.get_chunk_coordinates( + atomic_edge[0] + ) - cg.get_chunk_coordinates(atomic_edge[1]) + if np.any(np.abs(chunk_coord_delta) > 3): + raise ValueError("Chebyshev distance exceeded allowed maximum.") + + return ( + node_ids, + atomic_edge, + MergeOperation( + cg, + user_id=user_id, + added_edges=np.array(atomic_edge, dtype=np.uint64), + source_coords=coords[:1], + sink_coords=coords[1:], + ), + ) def _parse_split_payload( @@ -46,28 +71,33 @@ def _parse_split_payload( sink_coords = coords[node_idents == 1] bb_offset = (240, 240, 24) - return MulticutOperation( - cg, - user_id=user_id, - source_ids=source_ids, - sink_ids=sink_ids, - source_coords=source_coords, - sink_coords=sink_coords, - bbox_offset=bb_offset, - path_augment=True, - disallow_isolating_cut=True, + return ( + source_ids, + sink_ids, + MulticutOperation( + cg, + user_id=user_id, + source_ids=source_ids, + sink_ids=sink_ids, + source_coords=source_coords, + sink_coords=sink_coords, + bbox_offset=bb_offset, + path_augment=True, + disallow_isolating_cut=True, + ), ) def get_operation_from_request_payload( cg: ChunkedGraph, - payload: dict, + payload: Union[list, dict], split: bool, *, mincut: bool = True, user_id: str = None, -) -> GraphEditOperation: +) -> Tuple[np.ndarray, np.ndarray, GraphEditOperation]: if user_id is None: user_id = USER_ID if split: return _parse_split_payload(cg, user_id, payload, mincut=mincut) + return _parse_merge_payload(cg, user_id, payload) diff --git a/pychunkedgraph/debug/utils.py b/pychunkedgraph/debug/utils.py new file mode 100644 index 000000000..5de5d2ae0 --- /dev/null +++ b/pychunkedgraph/debug/utils.py @@ -0,0 +1,43 @@ +import numpy as np + +from ..graph import ChunkedGraph +from ..graph.utils.basetypes import NODE_ID + + +def print_attrs(d): + for k, v in d.items(): + try: + print(k.key) + except: + print(k) + try: + print(v[:2], "...") if type(v) is np.ndarray and len(v) > 2 else print(v) + except: + print(v) + + +def print_node( + cg: ChunkedGraph, + node: NODE_ID, + indent: int = 0, + stop_layer: int = 2, +) -> None: + children = cg.get_children(node) + print(f"{' ' * indent}{node}[{len(children)}]") + if cg.get_chunk_layer(node) <= stop_layer: + return + for child in children: + print_node(cg, child, indent=indent + 1, stop_layer=stop_layer) + + +def get_l2children(cg: ChunkedGraph, node: NODE_ID) -> np.ndarray: + nodes = np.array([node], dtype=NODE_ID) + layers = cg.get_chunk_layers(nodes) + assert np.all(layers > 2), "nodes must be at layers > 2" + l2children = [] + while nodes.size: + children = cg.get_children(np.sort(nodes), flatten=True) + layers = cg.get_chunk_layers(children) + l2children.append(children[layers == 2]) + nodes = children[layers > 2] + return np.concatenate(l2children) From 7935fc8c2f15e382f03005f4833518320d77d3e4 Mon Sep 17 00:00:00 2001 From: Akhilesh Halageri Date: Sun, 16 Oct 2022 18:11:15 +0000 Subject: [PATCH 02/28] fix: refactor and fix incorrect bounding l2 children due to skips, and speedups --- pychunkedgraph/graph/chunkedgraph.py | 129 +++++++++--------- .../graph/client/bigtable/client.py | 3 +- pychunkedgraph/graph/edges/utils.py | 46 ++----- 3 files changed, 79 insertions(+), 99 deletions(-) diff --git a/pychunkedgraph/graph/chunkedgraph.py b/pychunkedgraph/graph/chunkedgraph.py index f5909caef..e4562d888 100644 --- a/pychunkedgraph/graph/chunkedgraph.py +++ b/pychunkedgraph/graph/chunkedgraph.py @@ -265,7 +265,7 @@ def get_children( if flatten: if not node_children_d: return types.empty_1d.copy() - return np.concatenate([*node_children_d.values()]) + return np.concatenate(list(node_children_d.values())) return node_children_d def _get_children_multiple( @@ -308,7 +308,7 @@ def get_atomic_cross_edges( return self.cache.atomic_cross_edges_multiple(l2_ids) def get_cross_chunk_edges( - self, node_ids: np.ndarray, uplift=True, all_layers=False + self, node_ids: typing.Iterable, uplift=True, all_layers=False ) -> typing.Dict[np.uint64, typing.Dict[int, typing.Iterable]]: """ Cross chunk edges for `node_id` at `node_layer`. @@ -322,13 +322,14 @@ def get_cross_chunk_edges( This is because cross edges are stored only in level 2 IDs. """ result = {} + node_ids = np.array(node_ids, dtype=basetypes.NODE_ID) if not node_ids.size: return result node_l2ids_d = {} layers_ = self.get_chunk_layers(node_ids) for l in set(layers_): - node_l2ids_d.update(self._bounding_l2_children(node_ids[layers_ == l])) + node_l2ids_d.update(self._get_bounding_l2_children(node_ids[layers_ == l])) l2_edges_d_d = self.get_atomic_cross_edges( np.concatenate(list(node_l2ids_d.values())) ) @@ -850,75 +851,69 @@ def redo_operation( ).execute() # PRIVATE - def _bounding_l2_children(self, parent_ids: typing.Iterable) -> typing.Dict: - """ - Helper function to get level 2 children IDs for each parent. - `parent_ids` must be node IDs at same layer. - TODO what have i done (describe algo) - """ - from collections import defaultdict - - layers = self.get_chunk_layers(parent_ids) - assert np.all(layers == layers[0]) - - parents_layer = self.get_chunk_layer(parent_ids[0]) - chunk_coords = self.get_chunk_coordinates_multiple(parent_ids) - parent_coords_d = { - node_id: coord for node_id, coord in zip(parent_ids, chunk_coords) - } - - parent_bounding_chunk_ids = defaultdict(lambda: types.empty_1d) - parent_layer_mask = {} - parent_children_d = { - parent_id: np.array([parent_id], dtype=basetypes.NODE_ID) - for parent_id in parent_ids - } - children_layer = parents_layer - 1 - while children_layer >= 2: - parent_masked_children_d = {} - for parent_id, (X, Y, Z) in parent_coords_d.items(): - coords = chunk_utils.get_bounding_children_chunks( + def _get_bounding_chunk_ids(self, parent_chunk_ids: typing.Iterable) -> typing.Dict: + """ + Returns bounding chunk IDs at layers < parent_layer for all chunk IDs. + Dict[parent_chunk_id] = np.array(bounding_chunk_ids) + """ + parent_chunk_coords = self.get_chunk_coordinates_multiple(parent_chunk_ids) + parents_layer = self.get_chunk_layer(parent_chunk_ids[0]) + chunk_id_bchunk_ids_d = {} + for i, chunk_id in enumerate(parent_chunk_ids): + if chunk_id in chunk_id_bchunk_ids_d: + # `parent_chunk_ids` can have duplicates + # avoid redundant calculations + continue + parent_coord = parent_chunk_coords[i] + chunk_ids = [types.empty_1d] + for child_layer in range(2, parents_layer): + bcoords = chunk_utils.get_bounding_children_chunks( self.meta, parents_layer, - (X, Y, Z), - children_layer, + parent_coord, + child_layer, return_unique=False, ) - chunks_ids = chunk_utils.get_chunk_ids_from_coords( - self.meta, children_layer, coords - ) - parent_bounding_chunk_ids[parent_id] = chunks_ids - children = parent_children_d[parent_id] - layer_mask = self.get_chunk_layers(children) > children_layer - parent_layer_mask[parent_id] = layer_mask - parent_masked_children_d[parent_id] = children[layer_mask] - - children_ids = np.concatenate(list(parent_masked_children_d.values())) - child_grand_children_d = self.get_children(children_ids) - for parent_id, masked_children in parent_masked_children_d.items(): - bounding_chunk_ids = parent_bounding_chunk_ids[parent_id] - grand_children = [types.empty_1d] - for child in masked_children: - grand_children_ = child_grand_children_d[child] - mask = self.get_chunk_layers(grand_children_) == children_layer - masked_grand_children_ = grand_children_[mask] - chunk_ids = self.get_chunk_ids_from_node_ids(masked_grand_children_) - masked_grand_children_ = masked_grand_children_[ - np.in1d(chunk_ids, bounding_chunk_ids) - ] - grand_children_ = np.concatenate( - [masked_grand_children_, grand_children_[~mask]] - ) - grand_children.append(grand_children_) - grand_children = np.concatenate(grand_children) - unmasked_children = parent_children_d[parent_id] - layer_mask = parent_layer_mask[parent_id] - parent_children_d[parent_id] = np.concatenate( - [unmasked_children[~layer_mask], grand_children] + bchunks_ids = chunk_utils.get_chunk_ids_from_coords( + self.meta, child_layer, bcoords ) - children_layer -= 1 - return parent_children_d + chunk_ids.append(bchunks_ids) + chunk_id_bchunk_ids_d[chunk_id] = np.unique(np.concatenate(chunk_ids)) + return chunk_id_bchunk_ids_d + + def _get_bounding_l2_children(self, parents: typing.Iterable) -> typing.Dict: + parent_chunk_ids = self.get_chunk_ids_from_node_ids(parents) + chunk_id_bchunk_ids_d = self._get_bounding_chunk_ids(parent_chunk_ids) + + parent_descendants_d = { + _id: np.array([_id], dtype=basetypes.NODE_ID) for _id in parents + } + descendants_all = np.concatenate(list(parent_descendants_d.values())) + descendants_layers = self.get_chunk_layers(descendants_all) + layer_mask = descendants_layers > 2 + descendants_all = descendants_all[layer_mask] + + while descendants_all.size: + descendant_children_d = self.get_children(np.sort(descendants_all)) + for i, parent_id in enumerate(parents): + _descendants = parent_descendants_d[parent_id] + _layers = self.get_chunk_layers(_descendants) + _l2mask = _layers == 2 + descendants = [_descendants[_l2mask]] + for child in _descendants[~_l2mask]: + descendants.append(descendant_children_d[child]) + descendants = np.concatenate(descendants) + chunk_ids = self.get_chunk_ids_from_node_ids(descendants) + bchunk_ids = chunk_id_bchunk_ids_d[parent_chunk_ids[i]] + bounding_descendants = descendants[np.in1d(chunk_ids, bchunk_ids)] + parent_descendants_d[parent_id] = bounding_descendants + + descendants_all = np.concatenate(list(parent_descendants_d.values())) + descendants_layers = self.get_chunk_layers(descendants_all) + layer_mask = descendants_layers > 2 + descendants_all = descendants_all[layer_mask] + return parent_descendants_d # HELPERS / WRAPPERS @@ -960,6 +955,8 @@ def get_chunk_coordinates(self, node_or_chunk_id: basetypes.NODE_ID): def get_chunk_coordinates_multiple(self, node_or_chunk_ids: typing.Sequence): node_or_chunk_ids = np.array(node_or_chunk_ids, dtype=basetypes.NODE_ID) + layers = self.get_chunk_layers(node_or_chunk_ids) + assert np.all(layers == layers[0]), "All IDs must have the same layer." return chunk_utils.get_chunk_coordinates_multiple(self.meta, node_or_chunk_ids) def get_chunk_id( diff --git a/pychunkedgraph/graph/client/bigtable/client.py b/pychunkedgraph/graph/client/bigtable/client.py index 4d9142be0..0aca696d4 100644 --- a/pychunkedgraph/graph/client/bigtable/client.py +++ b/pychunkedgraph/graph/client/bigtable/client.py @@ -697,8 +697,7 @@ def _read_byte_rows( # Create filters: Rows row_set = RowSet() if row_keys is not None: - for row_key in row_keys: - row_set.add_row_key(row_key) + row_set.row_keys = list(row_keys) elif start_key is not None and end_key is not None: row_set.add_row_range_from_keys( start_key=start_key, diff --git a/pychunkedgraph/graph/edges/utils.py b/pychunkedgraph/graph/edges/utils.py index 5d602fa64..cc2dfa909 100644 --- a/pychunkedgraph/graph/edges/utils.py +++ b/pychunkedgraph/graph/edges/utils.py @@ -2,9 +2,7 @@ helper functions for edge stuff """ -from collections import defaultdict from typing import Dict -from typing import List from typing import Tuple from typing import Iterable from typing import Callable @@ -12,19 +10,15 @@ import numpy as np - from . import Edges from . import EDGE_TYPES from ..types import empty_2d from ..utils import basetypes from ..chunks import utils as chunk_utils from ..meta import ChunkedGraphMeta -from ..connectivity.search import check_reachability -from ..utils.flatgraph import build_gt_graph -from ...utils.general import reverse_dictionary -def concatenate_chunk_edges(chunk_edge_dicts: List) -> Dict: +def concatenate_chunk_edges(chunk_edge_dicts: Iterable) -> Dict: """combine edge_dicts of multiple chunks into one edge_dict""" edges_dict = {} for edge_type in EDGE_TYPES: @@ -49,31 +43,22 @@ def concatenate_chunk_edges(chunk_edge_dicts: List) -> Dict: return edges_dict -def concatenate_cross_edge_dicts(cross_edge_dicts: Iterable) -> Dict: - """Combines multiple cross edge dicts.""" - # print(cross_edge_dicts) - result_d = {} - for cross_edge_d in cross_edge_dicts: - result_d = merge_cross_edge_dicts_single(result_d, cross_edge_d) - return result_d +def concatenate_cross_edge_dicts(edges_ds: Iterable[Dict]) -> Dict: + """Combines cross chunk edge dicts of form {layer id : edge list}.""" + from collections import defaultdict + result_d = defaultdict(list) -def merge_cross_edge_dicts_single(x_edges_d1: Dict, x_edges_d2: Dict) -> Dict: - """Combines two cross chunk edge dicts of form {layer id : edge list}.""" - result_d = {} - if not x_edges_d1 and not x_edges_d2: - return result_d - layers = np.unique(list(x_edges_d1.keys()) + list(x_edges_d2.keys())) - for layer in range(2, max(layers) + 1): - edges1 = x_edges_d1.get(layer, empty_2d) - edges2 = x_edges_d2.get(layer, empty_2d) - edges1 = np.array(edges1, dtype=basetypes.NODE_ID) - edges2 = np.array(edges2, dtype=basetypes.NODE_ID) - result_d[layer] = np.concatenate([edges1, edges2]) + for edges_d in edges_ds: + for layer, edges in edges_d.items(): + result_d[layer].append(edges) + + for layer, edge_lists in result_d.items(): + result_d[layer] = np.concatenate(edge_lists) return result_d -def merge_cross_edge_dicts_multiple(x_edges_d1: Dict, x_edges_d2: Dict) -> Dict: +def merge_cross_edge_dicts(x_edges_d1: Dict, x_edges_d2: Dict) -> Dict: """ Combines two cross chunk dictionaries of form {node_id: {layer id : edge list}}. @@ -81,9 +66,8 @@ def merge_cross_edge_dicts_multiple(x_edges_d1: Dict, x_edges_d2: Dict) -> Dict: node_ids = np.unique(list(x_edges_d1.keys()) + list(x_edges_d2.keys())) result_d = {} for node_id in node_ids: - result_d[node_id] = merge_cross_edge_dicts_single( - x_edges_d1.get(node_id, {}), x_edges_d2.get(node_id, {}) - ) + cross_edge_ds = [x_edges_d1.get(node_id, {}), x_edges_d2.get(node_id, {})] + result_d[node_id] = concatenate_cross_edge_dicts(cross_edge_ds) return result_d @@ -179,7 +163,7 @@ def filter_min_layer_cross_edges( def filter_min_layer_cross_edges_multiple( - meta: ChunkedGraphMeta, l2id_atomic_cross_edges_ds: List, node_layer: int = 2 + meta: ChunkedGraphMeta, l2id_atomic_cross_edges_ds: Iterable, node_layer: int = 2 ) -> Tuple[int, Iterable]: """ Given a list of dicts of cross chunk edges [{layer: edges}] From 5860be7738fdd84fae97f3f92e8e9bb6c0f52846 Mon Sep 17 00:00:00 2001 From: Akhilesh Halageri Date: Sun, 16 Oct 2022 18:11:34 +0000 Subject: [PATCH 03/28] code cleanup --- pychunkedgraph/graph/edits.py | 46 +++++++++++++-------------- pychunkedgraph/graph/utils/generic.py | 6 ---- pychunkedgraph/graph/utils/inspect.py | 20 ------------ 3 files changed, 22 insertions(+), 50 deletions(-) delete mode 100644 pychunkedgraph/graph/utils/inspect.py diff --git a/pychunkedgraph/graph/edits.py b/pychunkedgraph/graph/edits.py index ca51bebe8..1fb5675f6 100644 --- a/pychunkedgraph/graph/edits.py +++ b/pychunkedgraph/graph/edits.py @@ -4,7 +4,6 @@ from typing import List from typing import Tuple from typing import Iterable -from typing import Sequence from collections import defaultdict from . import types @@ -14,13 +13,7 @@ from .utils import flatgraph from .utils.context_managers import TimeIt from .utils.serializers import serialize_uint64 -from .connectivity.nodes import edge_exists -from .connectivity.search import check_reachability -from .edges.utils import filter_min_layer_cross_edges -from .edges.utils import concatenate_cross_edge_dicts -from .edges.utils import merge_cross_edge_dicts_multiple from ..utils.general import in2d -from ..utils.general import reverse_dictionary def _init_old_hierarchy(cg, l2ids: np.ndarray, parent_ts: datetime.datetime = None): @@ -190,6 +183,9 @@ def add_edges( rows: Iterable = None, allow_same_segment_merge=False, ): + from .edges.utils import concatenate_cross_edge_dicts + from .edges.utils import merge_cross_edge_dicts + # TODO add docs if rows == None: rows = [] @@ -206,7 +202,7 @@ def add_edges( cg, l2ids, parent_ts=parent_ts ) atomic_children_d = cg.get_children(l2ids) - atomic_cross_edges_d = merge_cross_edge_dicts_multiple( + atomic_cross_edges_d = merge_cross_edge_dicts( cg.get_atomic_cross_edges(l2ids), l2_atomic_cross_edges_d ) @@ -240,7 +236,10 @@ def add_edges( time_stamp=time_stamp, parent_ts=parent_ts, ) - return create_parents.run(), new_l2_ids, rows + create_parents.create_new_entries() + with TimeIt("create_parents.run()"): + new_roots = create_parents.run() + new_entries = create_parents.create_new_entries() + return new_roots, new_l2_ids, new_entries def _process_l2_agglomeration( @@ -341,8 +340,7 @@ def remove_edges( ) with TimeIt("create_parents.run()"): new_roots = create_parents.run() - with TimeIt("create_parents.create_new_entries()"): - new_entries = create_parents.create_new_entries() + new_entries = create_parents.create_new_entries() return new_roots, new_l2_ids, new_entries @@ -386,12 +384,19 @@ def _update_id_lineage( self._new_old_id_d[parent].add(old_id) self._old_new_id_d[old_id].add(parent) + def _get_old_ids(self, new_ids): + old_ids = [ + np.array(list(self._new_old_id_d[id_]), dtype=basetypes.NODE_ID) + for id_ in new_ids + ] + return np.concatenate(old_ids) + def _map_sv_to_parent(self, node_ids, layer): sv_parent_d = {} sv_cross_edges = [types.empty_2d] for id_ in node_ids: edges_ = self._cross_edges_d[id_].get(layer, types.empty_2d) - sv_parent_d.update(dict(zip(edges_[:, 0].tolist(), [id_] * len(edges_)))) + sv_parent_d.update(dict(zip(edges_[:, 0], [id_] * len(edges_)))) sv_cross_edges.append(edges_) return sv_parent_d, np.concatenate(sv_cross_edges) @@ -427,11 +432,7 @@ def _get_layer_node_ids( self, new_ids: np.ndarray, layer: int ) -> Tuple[np.ndarray, np.ndarray]: # get old identities of new IDs - old_ids = [ - np.array(list(self._new_old_id_d[id_]), dtype=basetypes.NODE_ID) - for id_ in new_ids - ] - old_ids = np.concatenate(old_ids) + old_ids = self._get_old_ids(new_ids) # get their parents, then children of those parents node_ids = self.cg.get_children( np.unique( @@ -500,17 +501,14 @@ def run(self) -> Iterable: for layer in range(2, self.cg.meta.layer_count): if len(self._new_ids_d[layer]) == 0: continue - # with TimeIt(f"self._create_new_parents(layer) {layer}"): + # with TimeIt(f"_create_new_parents {layer}"): self._create_new_parents(layer) return self._new_ids_d[self.cg.meta.layer_count] def _update_root_id_lineage(self): new_root_ids = self._new_ids_d[self.cg.meta.layer_count] - former_root_ids = [ - np.array(list(self._new_old_id_d[id_]), dtype=basetypes.NODE_ID) - for id_ in new_root_ids - ] - former_root_ids = np.unique(np.concatenate(former_root_ids)) + former_root_ids = self._get_old_ids(new_root_ids) + former_root_ids = np.unique(former_root_ids) assert ( len(former_root_ids) < 2 or len(new_root_ids) < 2 ), "Something went wrong." @@ -553,7 +551,7 @@ def _get_atomic_cross_edges_val_dict(self): val_dicts[id_] = val_dict return val_dicts - def create_new_entries(self): + def create_new_entries(self) -> List: rows = [] val_dicts = self._get_atomic_cross_edges_val_dict() for layer in range(2, self.cg.meta.layer_count + 1): diff --git a/pychunkedgraph/graph/utils/generic.py b/pychunkedgraph/graph/utils/generic.py index fc19b9e06..9a2b6f979 100644 --- a/pychunkedgraph/graph/utils/generic.py +++ b/pychunkedgraph/graph/utils/generic.py @@ -11,19 +11,13 @@ from typing import Union from typing import Sequence from typing import Tuple -from itertools import product from collections import defaultdict import numpy as np import pandas as pd import pytz -from google.cloud import bigtable -from google.cloud.bigtable.row_filters import RowFilter -from cloudvolume import CloudVolume -from .. import types from ..chunks import utils as chunk_utils -from . import serializers def compute_indices_pandas(data) -> pd.Series: diff --git a/pychunkedgraph/graph/utils/inspect.py b/pychunkedgraph/graph/utils/inspect.py deleted file mode 100644 index 0470a2da7..000000000 --- a/pychunkedgraph/graph/utils/inspect.py +++ /dev/null @@ -1,20 +0,0 @@ -"""Helper functions to visualize/inspect tree hierarchy.""" - - -def dfs_print_node(cg, node, limit=2, sep=" |", count_only=False): - """ - `limit` stop at this layer - """ - leaves = 0 - stack = [(node, 0)] - while stack: - node, indent = stack.pop() - children = cg.get_children(node) - if not count_only: - print(sep * indent, node, f"({cg.get_chunk_layer(node)}, {len(children)})") - if cg.get_chunk_layer(children[0]) < limit: - leaves += len(children) - continue - for c in children: - stack.append((c, indent + 1)) - print("leaves count", leaves) From 0528b365fc9cc04ad07027483f3ba44ca3ae95be Mon Sep 17 00:00:00 2001 From: Akhilesh Halageri Date: Sun, 16 Oct 2022 21:46:26 +0000 Subject: [PATCH 04/28] decide if unique on number of IDs involved --- pychunkedgraph/graph/chunkedgraph.py | 22 +++++++++++++++++----- 1 file changed, 17 insertions(+), 5 deletions(-) diff --git a/pychunkedgraph/graph/chunkedgraph.py b/pychunkedgraph/graph/chunkedgraph.py index e4562d888..60a07cbda 100644 --- a/pychunkedgraph/graph/chunkedgraph.py +++ b/pychunkedgraph/graph/chunkedgraph.py @@ -328,8 +328,11 @@ def get_cross_chunk_edges( node_l2ids_d = {} layers_ = self.get_chunk_layers(node_ids) + # with TimeIt(f"_get_bounding_l2_children {node_ids.size}"): for l in set(layers_): - node_l2ids_d.update(self._get_bounding_l2_children(node_ids[layers_ == l])) + node_l2ids_d.update( + self._get_bounding_l2_children(node_ids[layers_ == l]) + ) l2_edges_d_d = self.get_atomic_cross_edges( np.concatenate(list(node_l2ids_d.values())) ) @@ -852,7 +855,11 @@ def redo_operation( # PRIVATE - def _get_bounding_chunk_ids(self, parent_chunk_ids: typing.Iterable) -> typing.Dict: + def _get_bounding_chunk_ids( + self, + parent_chunk_ids: typing.Iterable, + unique: bool = False, + ) -> typing.Dict: """ Returns bounding chunk IDs at layers < parent_layer for all chunk IDs. Dict[parent_chunk_id] = np.array(bounding_chunk_ids) @@ -879,12 +886,17 @@ def _get_bounding_chunk_ids(self, parent_chunk_ids: typing.Iterable) -> typing.D self.meta, child_layer, bcoords ) chunk_ids.append(bchunks_ids) - chunk_id_bchunk_ids_d[chunk_id] = np.unique(np.concatenate(chunk_ids)) + chunk_ids = np.concatenate(chunk_ids) + if unique: + chunk_ids = np.unique(chunk_ids) + chunk_id_bchunk_ids_d[chunk_id] = chunk_ids return chunk_id_bchunk_ids_d def _get_bounding_l2_children(self, parents: typing.Iterable) -> typing.Dict: parent_chunk_ids = self.get_chunk_ids_from_node_ids(parents) - chunk_id_bchunk_ids_d = self._get_bounding_chunk_ids(parent_chunk_ids) + chunk_id_bchunk_ids_d = self._get_bounding_chunk_ids( + parent_chunk_ids, unique=len(parents) >= 200 + ) parent_descendants_d = { _id: np.array([_id], dtype=basetypes.NODE_ID) for _id in parents @@ -895,7 +907,7 @@ def _get_bounding_l2_children(self, parents: typing.Iterable) -> typing.Dict: descendants_all = descendants_all[layer_mask] while descendants_all.size: - descendant_children_d = self.get_children(np.sort(descendants_all)) + descendant_children_d = self.get_children(descendants_all) for i, parent_id in enumerate(parents): _descendants = parent_descendants_d[parent_id] _layers = self.get_chunk_layers(_descendants) From 2b646b9ee821ceb2108d6486ea8b65e4e222ab5f Mon Sep 17 00:00:00 2001 From: Akhilesh Halageri Date: Tue, 18 Oct 2022 15:39:03 +0000 Subject: [PATCH 05/28] sort IDs for faster reads --- pychunkedgraph/debug/utils.py | 2 +- pychunkedgraph/graph/client/bigtable/client.py | 8 ++++++-- pychunkedgraph/graph/subgraph.py | 2 +- 3 files changed, 8 insertions(+), 4 deletions(-) diff --git a/pychunkedgraph/debug/utils.py b/pychunkedgraph/debug/utils.py index 5de5d2ae0..179f50aef 100644 --- a/pychunkedgraph/debug/utils.py +++ b/pychunkedgraph/debug/utils.py @@ -36,7 +36,7 @@ def get_l2children(cg: ChunkedGraph, node: NODE_ID) -> np.ndarray: assert np.all(layers > 2), "nodes must be at layers > 2" l2children = [] while nodes.size: - children = cg.get_children(np.sort(nodes), flatten=True) + children = cg.get_children(nodes, flatten=True) layers = cg.get_chunk_layers(children) l2children.append(children[layers == 2]) nodes = children[layers > 2] diff --git a/pychunkedgraph/graph/client/bigtable/client.py b/pychunkedgraph/graph/client/bigtable/client.py index 0aca696d4..24e5ebee7 100644 --- a/pychunkedgraph/graph/client/bigtable/client.py +++ b/pychunkedgraph/graph/client/bigtable/client.py @@ -114,6 +114,10 @@ def read_nodes( Read nodes and their properties. Accepts a range of node IDs or specific node IDs. """ + if node_ids is not None and len(node_ids) > self._max_row_key_count: + # bigtable reading is faster + # when all IDs in a block are within a range + node_ids = np.sort(node_ids) rows = self._read_byte_rows( start_key=serialize_uint64(start_id, fake_edges=fake_edges) if start_id is not None @@ -564,7 +568,7 @@ def get_max_node_id( ) -> basetypes.NODE_ID: """Gets the current maximum segment ID in the chunk.""" if root_chunk: - n_counters = np.uint64(2 ** 8) + n_counters = np.uint64(2**8) max_value = 0 for counter in range(n_counters): row = self._read_byte_row( @@ -623,7 +627,7 @@ def _get_root_segment_ids_range( self, chunk_id: basetypes.CHUNK_ID, size: int = 1, counter: int = None ) -> np.ndarray: """Return unique segment ID for the root chunk.""" - n_counters = np.uint64(2 ** 8) + n_counters = np.uint64(2**8) counter = ( np.uint64(counter % n_counters) if counter diff --git a/pychunkedgraph/graph/subgraph.py b/pychunkedgraph/graph/subgraph.py index 0759abcb8..ab2593175 100644 --- a/pychunkedgraph/graph/subgraph.py +++ b/pychunkedgraph/graph/subgraph.py @@ -201,7 +201,7 @@ def _get_dict_key(raw_key): def _get_subgraph_multiple_nodes_threaded( node_ids_batch: Iterable[np.uint64], ) -> List[np.uint64]: - children = cg.get_children(np.sort(node_ids_batch)) + children = cg.get_children(node_ids_batch) if bounding_box is not None: filtered_children = {} for node_id, nodes_children in children.items(): From 4f32a9c5a054dded4ad00e312f0cc8789c6c56ef Mon Sep 17 00:00:00 2001 From: Akhilesh Halageri Date: Fri, 4 Nov 2022 20:54:47 +0000 Subject: [PATCH 06/28] test pyfqmr --- pychunkedgraph/meshing/meshgen.py | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/pychunkedgraph/meshing/meshgen.py b/pychunkedgraph/meshing/meshgen.py index 511b7352d..588685527 100644 --- a/pychunkedgraph/meshing/meshgen.py +++ b/pychunkedgraph/meshing/meshgen.py @@ -418,7 +418,7 @@ def _get_root_ids(args): start_id, end_id = args root_ids[start_id:end_id] = cg.get_roots( - combined_ids[start_id:end_id], stop_layer=stop_layer, time_stamp=time_stamp + combined_ids[start_id:end_id], stop_layer=stop_layer, time_stamp=time_stamp, fail_to_zero=True ) rr = cg.range_read_chunk( @@ -1191,6 +1191,22 @@ def chunk_stitch_remeshing_task( cg, old_fragments, chunk_id, mip, high_padding ) + import pyfqmr + simplifier = pyfqmr.Simplify() + simplifier.setMesh( + new_fragment["vertices"].reshape(new_fragment["num_vertices"], 3), + new_fragment["faces"].reshape(-1, 3) + ) + simplifier.simplify_mesh( + target_count=4, + aggressiveness=5.0, + preserve_border=True, + verbose=False, + ) + v,f,_ = simplifier.getMesh() + new_fragment["vertices"] = v.flatten() + new_fragment["faces"] = f.flatten() + try: print(f'num_vertices = {len(new_fragment["vertices"])}') new_fragment_b = DracoPy.encode_mesh_to_buffer( From 3285941ac376eaaac9a370bc9e52382094672956 Mon Sep 17 00:00:00 2001 From: Akhilesh Halageri Date: Sat, 19 Nov 2022 21:24:25 +0000 Subject: [PATCH 07/28] fix: log L2 IDs being remeshed --- workers/mesh_worker.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/workers/mesh_worker.py b/workers/mesh_worker.py index b7bc3a913..a950543c3 100644 --- a/workers/mesh_worker.py +++ b/workers/mesh_worker.py @@ -38,7 +38,7 @@ def callback(payload): format="%(asctime)s %(message)s", datefmt="%m/%d/%Y %I:%M:%S %p", ) - logging.log(INFO_PRIORITY, f"Remeshing {lvl2_ids.size} L2 IDs in graph {table_id}") + logging.log(INFO_PRIORITY, f"Remeshing {lvl2_ids} L2 IDs in graph {table_id}") logging.log(INFO_PRIORITY, f"stop_layer={layer}, mip={mip}, max_err={err}") logging.log(INFO_PRIORITY, f"mesh_dir={mesh_dir}, unsharded_mesh_path={mesh_path}") meshgen.remeshing( From 363ef7acf9680d7474b4c0f42290ef15d09a8182 Mon Sep 17 00:00:00 2001 From: Akhilesh Halageri Date: Tue, 29 Nov 2022 16:36:04 +0000 Subject: [PATCH 08/28] undo meshgen changes, handled in another pr --- pychunkedgraph/meshing/meshgen.py | 18 +----------------- 1 file changed, 1 insertion(+), 17 deletions(-) diff --git a/pychunkedgraph/meshing/meshgen.py b/pychunkedgraph/meshing/meshgen.py index 588685527..511b7352d 100644 --- a/pychunkedgraph/meshing/meshgen.py +++ b/pychunkedgraph/meshing/meshgen.py @@ -418,7 +418,7 @@ def _get_root_ids(args): start_id, end_id = args root_ids[start_id:end_id] = cg.get_roots( - combined_ids[start_id:end_id], stop_layer=stop_layer, time_stamp=time_stamp, fail_to_zero=True + combined_ids[start_id:end_id], stop_layer=stop_layer, time_stamp=time_stamp ) rr = cg.range_read_chunk( @@ -1191,22 +1191,6 @@ def chunk_stitch_remeshing_task( cg, old_fragments, chunk_id, mip, high_padding ) - import pyfqmr - simplifier = pyfqmr.Simplify() - simplifier.setMesh( - new_fragment["vertices"].reshape(new_fragment["num_vertices"], 3), - new_fragment["faces"].reshape(-1, 3) - ) - simplifier.simplify_mesh( - target_count=4, - aggressiveness=5.0, - preserve_border=True, - verbose=False, - ) - v,f,_ = simplifier.getMesh() - new_fragment["vertices"] = v.flatten() - new_fragment["faces"] = f.flatten() - try: print(f'num_vertices = {len(new_fragment["vertices"])}') new_fragment_b = DracoPy.encode_mesh_to_buffer( From 59147c3a4c615ec92557b82e55c5647f533c3c04 Mon Sep 17 00:00:00 2001 From: Akhilesh Halageri Date: Tue, 29 Nov 2022 20:18:01 -0600 Subject: [PATCH 09/28] ingest faster automatic enqueuing (#428) * fix: improved auto queuing * feat: faster tracker worker with preloaded imports --- pychunkedgraph/ingest/cluster.py | 23 ++++++++++++++--------- tracker.py | 22 ++++++++++++++++++++++ 2 files changed, 36 insertions(+), 9 deletions(-) create mode 100644 tracker.py diff --git a/pychunkedgraph/ingest/cluster.py b/pychunkedgraph/ingest/cluster.py index 193d85d2c..e3fa83891 100644 --- a/pychunkedgraph/ingest/cluster.py +++ b/pychunkedgraph/ingest/cluster.py @@ -2,7 +2,6 @@ Ingest / create chunkedgraph with workers. """ -from itertools import product from typing import Sequence, Tuple import numpy as np @@ -46,8 +45,8 @@ def _post_task_completion(imanager: IngestionManager, layer: int, coords: np.nda ) imanager.redis.hset(parent_layer, parent_chunk_str, children_count) - queue = imanager.get_task_queue(f"t{layer}") - queue.enqueue( + tracker_queue = imanager.get_task_queue(f"t{layer}") + tracker_queue.enqueue( enqueue_parent_task, job_id=f"t{layer}_{chunk_str}", job_timeout=f"30s", @@ -65,7 +64,6 @@ def enqueue_parent_task( ): redis = get_redis_connection() imanager = IngestionManager.from_pickle(redis.get(r_keys.INGESTION_MANAGER)) - parent_id_str = chunk_id_str(parent_layer, parent_coords) parent_chunk_str = "_".join(map(str, parent_coords)) @@ -75,10 +73,18 @@ def enqueue_parent_task( print("parent already queued.") return - children_count = int(redis.hget(parent_layer, parent_chunk_str).decode("utf-8")) - if children_done != children_count: - print("children not done.") - return + # if the previous layer is complete + # no need to check children progress for each parent chunk + child_layer = parent_layer - 1 + child_layer_done = redis.scard(f"{child_layer}c") + child_layer_count = imanager.cg_meta.layer_chunk_counts[child_layer - 2] + child_layer_finished = child_layer_done == child_layer_count + + if not child_layer_finished: + children_count = int(redis.hget(parent_layer, parent_chunk_str).decode("utf-8")) + if children_done != children_count: + print("children not done.") + return queue = imanager.get_task_queue(f"l{parent_layer}") queue.enqueue( @@ -125,7 +131,6 @@ def enqueue_atomic_tasks(imanager: IngestionManager): from os import environ from time import sleep from rq import Queue as RQueue - from ..utils.general import chunked chunk_coords = _get_test_chunks(imanager.cg.meta) chunk_count = len(chunk_coords) diff --git a/tracker.py b/tracker.py new file mode 100644 index 000000000..d2ae63cb3 --- /dev/null +++ b/tracker.py @@ -0,0 +1,22 @@ +import sys +from rq import Connection, Worker + +# Preload libraries from pychunkedgraph.ingest.cluster +from typing import Sequence, Tuple + +import numpy as np + +from pychunkedgraph.ingest.utils import chunk_id_str +from pychunkedgraph.ingest.manager import IngestionManager +from pychunkedgraph.ingest.common import get_atomic_chunk_data +from pychunkedgraph.ingest.ran_agglomeration import get_active_edges +from pychunkedgraph.ingest.create.atomic_layer import add_atomic_edges +from pychunkedgraph.ingest.create.abstract_layers import add_layer +from pychunkedgraph.graph.meta import ChunkedGraphMeta +from pychunkedgraph.graph.chunks.hierarchy import get_children_chunk_coords +from pychunkedgraph.utils.redis import keys as r_keys +from pychunkedgraph.utils.redis import get_redis_connection + +qs = sys.argv[1:] +w = Worker(qs, connection=get_redis_connection()) +w.work() \ No newline at end of file From 7624d0d05d07e857f6de3835b39ecfd3508d3aae Mon Sep 17 00:00:00 2001 From: Forrest Collman Date: Wed, 30 Nov 2022 15:32:28 -0800 Subject: [PATCH 10/28] Add public neurons (#426) * adding public info file * adding mesh manifest public neuron * using config for AUTH_TOKEN * adding more endpoints for public neurons * fixing AUTH_TOKEN * adding oldest timestamp * removing node id * removing auth required before/after request * trying to add reset auth * removing unused reset auth * pinning zmesh back * adding compiled req feature * trying to fix tests * compiled new protobuf files * new chunkMapping proto files * modifiying input reqs * removing unused chunkMapping files --- compile_reqs.sh | 1 + pychunkedgraph/app/meshing/v1/routes.py | 20 +- .../app/segmentation/generic/routes.py | 14 +- pychunkedgraph/app/segmentation/v1/routes.py | 39 +- .../io/protobuf/chunkComponents_pb2.py | 66 +-- pychunkedgraph/io/protobuf/chunkEdges_pb2.py | 144 +----- .../io/protobuf/chunkMapping_pb2.py | 77 --- requirements.in | 32 ++ requirements.txt | 438 ++++++++++++++++-- 9 files changed, 525 insertions(+), 306 deletions(-) create mode 100755 compile_reqs.sh delete mode 100644 pychunkedgraph/io/protobuf/chunkMapping_pb2.py create mode 100644 requirements.in diff --git a/compile_reqs.sh b/compile_reqs.sh new file mode 100755 index 000000000..2d74c225d --- /dev/null +++ b/compile_reqs.sh @@ -0,0 +1 @@ +docker run -v ${PWD}:/app caveconnectome/pychunkedgraph:v2.4.0 /bin/bash -c "pip install pip-tools && pip-compile requirements.in --resolver=backtracking -v --output-file requirements.txt" \ No newline at end of file diff --git a/pychunkedgraph/app/meshing/v1/routes.py b/pychunkedgraph/app/meshing/v1/routes.py index 3a8dac3c8..b7c09a672 100644 --- a/pychunkedgraph/app/meshing/v1/routes.py +++ b/pychunkedgraph/app/meshing/v1/routes.py @@ -1,9 +1,16 @@ from flask import Blueprint from middle_auth_client import auth_requires_permission, auth_required - from pychunkedgraph.app.meshing import common from pychunkedgraph.graph import exceptions as cg_exceptions from pychunkedgraph.app.app_utils import remap_public +import os +import json + +if os.environ.get("DAF_CREDENTIALS", None) is not None: + with open(os.environ.get("DAF_CREDENTIALS"), "r") as f: + AUTH_TOKEN = json.load(f)["token"] +else: + AUTH_TOKEN = "" bp = Blueprint( "pcg_meshing_v1", __name__, url_prefix=f"/{common.__meshing_url_prefix__}/api/v1" @@ -33,13 +40,13 @@ def home(): @bp.before_request -@auth_required +# @auth_required def before_request(): return common.before_request() @bp.after_request -@auth_required +# @auth_required def after_request(response): return common.after_request(response) @@ -68,7 +75,12 @@ def handle_valid_frags(table_id, node_id): @bp.route("/table//manifest/:0", methods=["GET"]) -@auth_requires_permission("view") +@auth_requires_permission( + "view", + public_table_key="table_id", + public_node_key="node_id", + service_token=AUTH_TOKEN, +) @remap_public def handle_get_manifest(table_id, node_id): return common.handle_get_manifest(table_id, node_id) diff --git a/pychunkedgraph/app/segmentation/generic/routes.py b/pychunkedgraph/app/segmentation/generic/routes.py index b17f4948f..a2d431699 100644 --- a/pychunkedgraph/app/segmentation/generic/routes.py +++ b/pychunkedgraph/app/segmentation/generic/routes.py @@ -1,4 +1,4 @@ -from flask import Blueprint +from flask import Blueprint, current_app from middle_auth_client import ( auth_requires_admin, auth_required, @@ -6,6 +6,14 @@ ) from pychunkedgraph.app.segmentation import common from pychunkedgraph.app.app_utils import remap_public +import os +import json + +if os.environ.get("DAF_CREDENTIALS", None) is not None: + with open(os.environ.get("DAF_CREDENTIALS"), "r") as f: + AUTH_TOKEN = json.load(f)["token"] +else: + AUTH_TOKEN = "" bp = Blueprint( "pcg_generic_v1", __name__, url_prefix=f"/{common.__segmentation_url_prefix__}" @@ -42,7 +50,9 @@ def sleep_me(sleep): @bp.route("/table//info", methods=["GET"]) -@auth_requires_permission("view") +@auth_requires_permission( + "view", public_table_key="table_id", service_token=AUTH_TOKEN +) @remap_public def handle_info(table_id): return common.handle_info(table_id) diff --git a/pychunkedgraph/app/segmentation/v1/routes.py b/pychunkedgraph/app/segmentation/v1/routes.py index 57ba52b52..e67c5e3e3 100644 --- a/pychunkedgraph/app/segmentation/v1/routes.py +++ b/pychunkedgraph/app/segmentation/v1/routes.py @@ -3,7 +3,7 @@ import pickle import pandas as pd -from flask import make_response, current_app +from flask import make_response from flask import Blueprint, request from middle_auth_client import auth_requires_permission from middle_auth_client import auth_requires_admin @@ -17,6 +17,14 @@ ) from pychunkedgraph.app.segmentation import common from pychunkedgraph.graph import exceptions as cg_exceptions +import os +import json + +if os.environ.get("DAF_CREDENTIALS", None) is not None: + with open(os.environ.get("DAF_CREDENTIALS"), "r") as f: + AUTH_TOKEN = json.load(f)["token"] +else: + AUTH_TOKEN = "" bp = Blueprint( "pcg_segmentation_v1", @@ -48,13 +56,11 @@ def home(): @bp.before_request -@auth_required def before_request(): return common.before_request() @bp.after_request -@auth_required def after_request(response): return common.after_request(response) @@ -283,7 +289,12 @@ def handle_l2_chunk_children_binary(table_id, chunk_id): @bp.route("/table//node//leaves", methods=["GET"]) -@auth_requires_permission("view") +@auth_requires_permission( + "view", + public_table_key="table_id", + public_node_key="node_id", + service_token=AUTH_TOKEN, +) @remap_public(edit=False) def handle_leaves(table_id, node_id): int64_as_str = request.args.get("int64_as_str", default=False, type=toboolean) @@ -297,7 +308,12 @@ def handle_leaves(table_id, node_id): @bp.route("/table//node/leaves_many", methods=["POST"]) @bp.route("/table//leaves_many", methods=["POST"]) -@auth_requires_permission("view") +@auth_requires_permission( + "view", + public_table_key="table_id", + public_node_key="node_id", + service_token=AUTH_TOKEN, +) @remap_public(check_node_ids=True) def handle_leaves_many(table_id): int64_as_str = request.args.get("int64_as_str", default=False, type=toboolean) @@ -464,7 +480,11 @@ def handle_past_id_mapping(table_id): @bp.route("/table//oldest_timestamp", methods=["GET"]) -@auth_requires_permission("view") +@auth_requires_permission( + "view", + public_table_key="table_id", + service_token=AUTH_TOKEN, +) @remap_public(edit=False) def oldest_timestamp(table_id): int64_as_str = request.args.get("int64_as_str", default=False, type=toboolean) @@ -538,7 +558,12 @@ def handle_roots_from_coords(table_id): ## Get level2 graph ------------------------------------------------------------- @bp.route("/table//node//lvl2_graph", methods=["GET"]) -@auth_requires_permission("view") +@auth_requires_permission( + "view", + public_table_key="table_id", + public_node_key="node_id", + service_token=AUTH_TOKEN, +) @remap_public(edit=False) def handle_get_lvl2_graph(table_id, node_id): int64_as_str = request.args.get("int64_as_str", default=False, type=toboolean) diff --git a/pychunkedgraph/io/protobuf/chunkComponents_pb2.py b/pychunkedgraph/io/protobuf/chunkComponents_pb2.py index eb94988a7..19a3914e8 100644 --- a/pychunkedgraph/io/protobuf/chunkComponents_pb2.py +++ b/pychunkedgraph/io/protobuf/chunkComponents_pb2.py @@ -1,12 +1,10 @@ # -*- coding: utf-8 -*- # Generated by the protocol buffer compiler. DO NOT EDIT! # source: chunkComponents.proto - -import sys -_b=sys.version_info[0]<3 and (lambda x:x) or (lambda x:x.encode('latin1')) +"""Generated protocol buffer code.""" +from google.protobuf.internal import builder as _builder from google.protobuf import descriptor as _descriptor -from google.protobuf import message as _message -from google.protobuf import reflection as _reflection +from google.protobuf import descriptor_pool as _descriptor_pool from google.protobuf import symbol_database as _symbol_database # @@protoc_insertion_point(imports) @@ -14,57 +12,13 @@ +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x15\x63hunkComponents.proto\x12\ncomponents\"(\n\x12\x43hunkComponentsMsg\x12\x12\n\ncomponents\x18\x01 \x03(\x04\x62\x06proto3') -DESCRIPTOR = _descriptor.FileDescriptor( - name='chunkComponents.proto', - package='components', - syntax='proto3', - serialized_options=None, - serialized_pb=_b('\n\x15\x63hunkComponents.proto\x12\ncomponents\"(\n\x12\x43hunkComponentsMsg\x12\x12\n\ncomponents\x18\x01 \x03(\x04\x62\x06proto3') -) - - - - -_CHUNKCOMPONENTSMSG = _descriptor.Descriptor( - name='ChunkComponentsMsg', - full_name='components.ChunkComponentsMsg', - filename=None, - file=DESCRIPTOR, - containing_type=None, - fields=[ - _descriptor.FieldDescriptor( - name='components', full_name='components.ChunkComponentsMsg.components', index=0, - number=1, type=4, cpp_type=4, label=3, - has_default_value=False, default_value=[], - message_type=None, enum_type=None, containing_type=None, - is_extension=False, extension_scope=None, - serialized_options=None, file=DESCRIPTOR), - ], - extensions=[ - ], - nested_types=[], - enum_types=[ - ], - serialized_options=None, - is_extendable=False, - syntax='proto3', - extension_ranges=[], - oneofs=[ - ], - serialized_start=37, - serialized_end=77, -) - -DESCRIPTOR.message_types_by_name['ChunkComponentsMsg'] = _CHUNKCOMPONENTSMSG -_sym_db.RegisterFileDescriptor(DESCRIPTOR) - -ChunkComponentsMsg = _reflection.GeneratedProtocolMessageType('ChunkComponentsMsg', (_message.Message,), { - 'DESCRIPTOR' : _CHUNKCOMPONENTSMSG, - '__module__' : 'chunkComponents_pb2' - # @@protoc_insertion_point(class_scope:components.ChunkComponentsMsg) - }) -_sym_db.RegisterMessage(ChunkComponentsMsg) - +_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, globals()) +_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'chunkComponents_pb2', globals()) +if _descriptor._USE_C_DESCRIPTORS == False: + DESCRIPTOR._options = None + _CHUNKCOMPONENTSMSG._serialized_start=37 + _CHUNKCOMPONENTSMSG._serialized_end=77 # @@protoc_insertion_point(module_scope) diff --git a/pychunkedgraph/io/protobuf/chunkEdges_pb2.py b/pychunkedgraph/io/protobuf/chunkEdges_pb2.py index 929137e88..b90c15c39 100644 --- a/pychunkedgraph/io/protobuf/chunkEdges_pb2.py +++ b/pychunkedgraph/io/protobuf/chunkEdges_pb2.py @@ -1,12 +1,10 @@ # -*- coding: utf-8 -*- # Generated by the protocol buffer compiler. DO NOT EDIT! # source: chunkEdges.proto - -import sys -_b=sys.version_info[0]<3 and (lambda x:x) or (lambda x:x.encode('latin1')) +"""Generated protocol buffer code.""" +from google.protobuf.internal import builder as _builder from google.protobuf import descriptor as _descriptor -from google.protobuf import message as _message -from google.protobuf import reflection as _reflection +from google.protobuf import descriptor_pool as _descriptor_pool from google.protobuf import symbol_database as _symbol_database # @@protoc_insertion_point(imports) @@ -15,133 +13,15 @@ -DESCRIPTOR = _descriptor.FileDescriptor( - name='chunkEdges.proto', - package='edges', - syntax='proto3', - serialized_options=None, - serialized_pb=_b('\n\x10\x63hunkEdges.proto\x12\x05\x65\x64ges\"S\n\x08\x45\x64gesMsg\x12\x11\n\tnode_ids1\x18\x01 \x01(\x0c\x12\x11\n\tnode_ids2\x18\x02 \x01(\x0c\x12\x12\n\naffinities\x18\x03 \x01(\x0c\x12\r\n\x05\x61reas\x18\x04 \x01(\x0c\"\x80\x01\n\rChunkEdgesMsg\x12!\n\x08in_chunk\x18\x01 \x01(\x0b\x32\x0f.edges.EdgesMsg\x12$\n\x0b\x63ross_chunk\x18\x02 \x01(\x0b\x32\x0f.edges.EdgesMsg\x12&\n\rbetween_chunk\x18\x03 \x01(\x0b\x32\x0f.edges.EdgesMsgb\x06proto3') -) - - - - -_EDGESMSG = _descriptor.Descriptor( - name='EdgesMsg', - full_name='edges.EdgesMsg', - filename=None, - file=DESCRIPTOR, - containing_type=None, - fields=[ - _descriptor.FieldDescriptor( - name='node_ids1', full_name='edges.EdgesMsg.node_ids1', index=0, - number=1, type=12, cpp_type=9, label=1, - has_default_value=False, default_value=_b(""), - message_type=None, enum_type=None, containing_type=None, - is_extension=False, extension_scope=None, - serialized_options=None, file=DESCRIPTOR), - _descriptor.FieldDescriptor( - name='node_ids2', full_name='edges.EdgesMsg.node_ids2', index=1, - number=2, type=12, cpp_type=9, label=1, - has_default_value=False, default_value=_b(""), - message_type=None, enum_type=None, containing_type=None, - is_extension=False, extension_scope=None, - serialized_options=None, file=DESCRIPTOR), - _descriptor.FieldDescriptor( - name='affinities', full_name='edges.EdgesMsg.affinities', index=2, - number=3, type=12, cpp_type=9, label=1, - has_default_value=False, default_value=_b(""), - message_type=None, enum_type=None, containing_type=None, - is_extension=False, extension_scope=None, - serialized_options=None, file=DESCRIPTOR), - _descriptor.FieldDescriptor( - name='areas', full_name='edges.EdgesMsg.areas', index=3, - number=4, type=12, cpp_type=9, label=1, - has_default_value=False, default_value=_b(""), - message_type=None, enum_type=None, containing_type=None, - is_extension=False, extension_scope=None, - serialized_options=None, file=DESCRIPTOR), - ], - extensions=[ - ], - nested_types=[], - enum_types=[ - ], - serialized_options=None, - is_extendable=False, - syntax='proto3', - extension_ranges=[], - oneofs=[ - ], - serialized_start=27, - serialized_end=110, -) - - -_CHUNKEDGESMSG = _descriptor.Descriptor( - name='ChunkEdgesMsg', - full_name='edges.ChunkEdgesMsg', - filename=None, - file=DESCRIPTOR, - containing_type=None, - fields=[ - _descriptor.FieldDescriptor( - name='in_chunk', full_name='edges.ChunkEdgesMsg.in_chunk', index=0, - number=1, type=11, cpp_type=10, label=1, - has_default_value=False, default_value=None, - message_type=None, enum_type=None, containing_type=None, - is_extension=False, extension_scope=None, - serialized_options=None, file=DESCRIPTOR), - _descriptor.FieldDescriptor( - name='cross_chunk', full_name='edges.ChunkEdgesMsg.cross_chunk', index=1, - number=2, type=11, cpp_type=10, label=1, - has_default_value=False, default_value=None, - message_type=None, enum_type=None, containing_type=None, - is_extension=False, extension_scope=None, - serialized_options=None, file=DESCRIPTOR), - _descriptor.FieldDescriptor( - name='between_chunk', full_name='edges.ChunkEdgesMsg.between_chunk', index=2, - number=3, type=11, cpp_type=10, label=1, - has_default_value=False, default_value=None, - message_type=None, enum_type=None, containing_type=None, - is_extension=False, extension_scope=None, - serialized_options=None, file=DESCRIPTOR), - ], - extensions=[ - ], - nested_types=[], - enum_types=[ - ], - serialized_options=None, - is_extendable=False, - syntax='proto3', - extension_ranges=[], - oneofs=[ - ], - serialized_start=113, - serialized_end=241, -) - -_CHUNKEDGESMSG.fields_by_name['in_chunk'].message_type = _EDGESMSG -_CHUNKEDGESMSG.fields_by_name['cross_chunk'].message_type = _EDGESMSG -_CHUNKEDGESMSG.fields_by_name['between_chunk'].message_type = _EDGESMSG -DESCRIPTOR.message_types_by_name['EdgesMsg'] = _EDGESMSG -DESCRIPTOR.message_types_by_name['ChunkEdgesMsg'] = _CHUNKEDGESMSG -_sym_db.RegisterFileDescriptor(DESCRIPTOR) - -EdgesMsg = _reflection.GeneratedProtocolMessageType('EdgesMsg', (_message.Message,), { - 'DESCRIPTOR' : _EDGESMSG, - '__module__' : 'chunkEdges_pb2' - # @@protoc_insertion_point(class_scope:edges.EdgesMsg) - }) -_sym_db.RegisterMessage(EdgesMsg) - -ChunkEdgesMsg = _reflection.GeneratedProtocolMessageType('ChunkEdgesMsg', (_message.Message,), { - 'DESCRIPTOR' : _CHUNKEDGESMSG, - '__module__' : 'chunkEdges_pb2' - # @@protoc_insertion_point(class_scope:edges.ChunkEdgesMsg) - }) -_sym_db.RegisterMessage(ChunkEdgesMsg) +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x10\x63hunkEdges.proto\x12\x05\x65\x64ges\"S\n\x08\x45\x64gesMsg\x12\x11\n\tnode_ids1\x18\x01 \x01(\x0c\x12\x11\n\tnode_ids2\x18\x02 \x01(\x0c\x12\x12\n\naffinities\x18\x03 \x01(\x0c\x12\r\n\x05\x61reas\x18\x04 \x01(\x0c\"\x80\x01\n\rChunkEdgesMsg\x12!\n\x08in_chunk\x18\x01 \x01(\x0b\x32\x0f.edges.EdgesMsg\x12$\n\x0b\x63ross_chunk\x18\x02 \x01(\x0b\x32\x0f.edges.EdgesMsg\x12&\n\rbetween_chunk\x18\x03 \x01(\x0b\x32\x0f.edges.EdgesMsgb\x06proto3') +_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, globals()) +_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'chunkEdges_pb2', globals()) +if _descriptor._USE_C_DESCRIPTORS == False: + DESCRIPTOR._options = None + _EDGESMSG._serialized_start=27 + _EDGESMSG._serialized_end=110 + _CHUNKEDGESMSG._serialized_start=113 + _CHUNKEDGESMSG._serialized_end=241 # @@protoc_insertion_point(module_scope) diff --git a/pychunkedgraph/io/protobuf/chunkMapping_pb2.py b/pychunkedgraph/io/protobuf/chunkMapping_pb2.py deleted file mode 100644 index 8801bfabb..000000000 --- a/pychunkedgraph/io/protobuf/chunkMapping_pb2.py +++ /dev/null @@ -1,77 +0,0 @@ -# -*- coding: utf-8 -*- -# Generated by the protocol buffer compiler. DO NOT EDIT! -# source: chunkMapping.proto - -import sys -_b=sys.version_info[0]<3 and (lambda x:x) or (lambda x:x.encode('latin1')) -from google.protobuf import descriptor as _descriptor -from google.protobuf import message as _message -from google.protobuf import reflection as _reflection -from google.protobuf import symbol_database as _symbol_database -# @@protoc_insertion_point(imports) - -_sym_db = _symbol_database.Default() - - - - -DESCRIPTOR = _descriptor.FileDescriptor( - name='chunkMapping.proto', - package='mapping', - syntax='proto3', - serialized_options=None, - serialized_pb=_b('\n\x12\x63hunkMapping.proto\x12\x07mapping\":\n\x0f\x43hunkMappingMsg\x12\x13\n\x0bsupervoxels\x18\x01 \x01(\x0c\x12\x12\n\ncomponents\x18\x02 \x01(\x0c\x62\x06proto3') -) - - - - -_CHUNKMAPPINGMSG = _descriptor.Descriptor( - name='ChunkMappingMsg', - full_name='mapping.ChunkMappingMsg', - filename=None, - file=DESCRIPTOR, - containing_type=None, - fields=[ - _descriptor.FieldDescriptor( - name='supervoxels', full_name='mapping.ChunkMappingMsg.supervoxels', index=0, - number=1, type=12, cpp_type=9, label=1, - has_default_value=False, default_value=_b(""), - message_type=None, enum_type=None, containing_type=None, - is_extension=False, extension_scope=None, - serialized_options=None, file=DESCRIPTOR), - _descriptor.FieldDescriptor( - name='components', full_name='mapping.ChunkMappingMsg.components', index=1, - number=2, type=12, cpp_type=9, label=1, - has_default_value=False, default_value=_b(""), - message_type=None, enum_type=None, containing_type=None, - is_extension=False, extension_scope=None, - serialized_options=None, file=DESCRIPTOR), - ], - extensions=[ - ], - nested_types=[], - enum_types=[ - ], - serialized_options=None, - is_extendable=False, - syntax='proto3', - extension_ranges=[], - oneofs=[ - ], - serialized_start=31, - serialized_end=89, -) - -DESCRIPTOR.message_types_by_name['ChunkMappingMsg'] = _CHUNKMAPPINGMSG -_sym_db.RegisterFileDescriptor(DESCRIPTOR) - -ChunkMappingMsg = _reflection.GeneratedProtocolMessageType('ChunkMappingMsg', (_message.Message,), { - 'DESCRIPTOR' : _CHUNKMAPPINGMSG, - '__module__' : 'chunkMapping_pb2' - # @@protoc_insertion_point(class_scope:mapping.ChunkMappingMsg) - }) -_sym_db.RegisterMessage(ChunkMappingMsg) - - -# @@protoc_insertion_point(module_scope) diff --git a/requirements.in b/requirements.in new file mode 100644 index 000000000..46d874e00 --- /dev/null +++ b/requirements.in @@ -0,0 +1,32 @@ +click >= 8.0 +protobuf>4 +urllib3>=1.26.3 +cloud-files>=1.25.2 +requests>=2.25.0 +grpcio>=1.36.1 +certifi>=2020.12.5 +cloud-volume==6.1.1 +numpy +pandas +networkx==2.1 +google-cloud-bigtable>=0.33.0 +google-cloud-datastore>=1.8<=2.0dev +flask +flask_cors +codecov +multiwrapper +python-json-logger +zstandard +redis +rq +middle-auth-client>=3.11.0 +zmesh<1.5.0 +fastremap +pyyaml +cachetools +task-queue==1.0.0 +messagingclient +jsonschema<4.0 +importlib-metadata>=4.5.0<5.0.0 +dracopy<=0.0.19 +tox>3.25.0 diff --git a/requirements.txt b/requirements.txt index a9ad97359..9597be1b6 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,31 +1,413 @@ -click >= 8.0 -protobuf~=3.20 -urllib3>=1.26.3 -cloud-files>=1.25.2 -requests>=2.25.0 -grpcio>=1.36.1 -certifi>=2020.12.5 +# +# This file is autogenerated by pip-compile with python 3.7 +# To update, run: +# +# pip-compile --output-file=requirements.txt --resolver=backtracking requirements.in +# +async-timeout==4.0.2 + # via redis +attrs==22.1.0 + # via + # jsonschema + # pytest +boto3==1.26.16 + # via + # cloud-files + # cloud-volume +botocore==1.29.16 + # via + # boto3 + # s3transfer +brotli==1.0.9 + # via + # cloud-files + # urllib3 +cachetools==5.2.0 + # via + # -r requirements.in + # google-auth + # middle-auth-client +certifi==2022.9.24 + # via + # -r requirements.in + # requests + # urllib3 +cffi==1.15.1 + # via cryptography +chardet==5.0.0 + # via + # cloud-files + # cloud-volume +charset-normalizer==2.1.1 + # via requests +click==8.1.3 + # via + # -r requirements.in + # cloud-files + # compressed-segmentation + # compresso + # flask + # rq +cloud-files==4.11.0 + # via + # -r requirements.in + # cloud-volume cloud-volume==6.1.1 -numpy -pandas + # via + # -r requirements.in + # task-queue +codecov==2.1.12 + # via -r requirements.in +compressed-segmentation==2.2.0 + # via cloud-volume +compresso==3.0.0 + # via cloud-volume +coverage==6.5.0 + # via codecov +crc32c==2.3 + # via cloud-files +cryptography==38.0.3 + # via + # pyopenssl + # urllib3 +decorator==5.1.1 + # via networkx +deflate==0.3.0 + # via cloud-files +dill==0.3.6 + # via + # multiprocess + # pathos +distlib==0.3.6 + # via virtualenv +dracopy==0.0.19 + # via + # -r requirements.in + # cloud-volume +exceptiongroup==1.0.4 + # via pytest +fastremap==1.13.3 + # via + # -r requirements.in + # cloud-volume +filelock==3.8.0 + # via + # tox + # virtualenv +flask==2.2.2 + # via + # -r requirements.in + # flask-cors + # middle-auth-client +flask-cors==3.0.10 + # via -r requirements.in +fpzip==1.2.0 + # via cloud-volume +furl==2.1.3 + # via middle-auth-client +gevent==22.10.2 + # via + # cloud-files + # cloud-volume + # task-queue +google-api-core[grpc]==2.10.2 + # via + # google-cloud-bigtable + # google-cloud-core + # google-cloud-datastore + # google-cloud-pubsub + # google-cloud-storage +google-auth==2.14.1 + # via + # cloud-files + # cloud-volume + # google-api-core + # google-cloud-core + # google-cloud-storage +google-cloud-bigtable==2.13.2 + # via -r requirements.in +google-cloud-core==2.3.2 + # via + # cloud-files + # cloud-volume + # google-cloud-bigtable + # google-cloud-datastore + # google-cloud-storage +google-cloud-datastore==2.10.0 + # via -r requirements.in +google-cloud-pubsub==2.13.11 + # via messagingclient +google-cloud-storage==2.6.0 + # via + # cloud-files + # cloud-volume +google-crc32c==1.5.0 + # via + # cloud-files + # google-resumable-media +google-resumable-media==2.4.0 + # via google-cloud-storage +googleapis-common-protos[grpc]==1.57.0 + # via + # google-api-core + # grpc-google-iam-v1 + # grpcio-status +greenlet==2.0.1 + # via gevent +grpc-google-iam-v1==0.12.4 + # via + # google-cloud-bigtable + # google-cloud-pubsub +grpcio==1.50.0 + # via + # -r requirements.in + # google-api-core + # google-cloud-pubsub + # googleapis-common-protos + # grpc-google-iam-v1 + # grpcio-status +grpcio-status==1.50.0 + # via + # google-api-core + # google-cloud-pubsub +idna==3.4 + # via + # requests + # urllib3 +importlib-metadata==5.1.0 + # via + # -r requirements.in + # click + # flask + # jsonschema + # markdown + # pluggy + # pytest + # redis + # tox + # virtualenv +inflection==0.5.1 + # via python-jsonschema-objects +iniconfig==1.1.1 + # via pytest +itsdangerous==2.1.2 + # via flask +jinja2==3.1.2 + # via flask +jmespath==1.0.1 + # via + # boto3 + # botocore +json5==0.9.10 + # via cloud-volume +jsonschema==3.2.0 + # via + # -r requirements.in + # cloud-volume + # python-jsonschema-objects +markdown==3.4.1 + # via python-jsonschema-objects +markupsafe==2.1.1 + # via + # jinja2 + # werkzeug +messagingclient==0.1.2 + # via -r requirements.in +middle-auth-client==3.15.0 + # via -r requirements.in +multiprocess==0.70.14 + # via pathos +multiwrapper==0.1.1 + # via -r requirements.in networkx==2.1 -google-cloud-bigtable>=0.33.0 -google-cloud-datastore>=1.8<=2.0dev -flask -flask_cors -codecov -multiwrapper -python-json-logger -zstandard -redis -rq -middle-auth-client>=3.11.0 -zmesh -fastremap -pyyaml -cachetools + # via + # -r requirements.in + # cloud-volume +numpy==1.21.6 + # via + # -r requirements.in + # cloud-volume + # compressed-segmentation + # compresso + # fastremap + # fpzip + # messagingclient + # multiwrapper + # pandas + # simplejpeg + # task-queue + # zmesh +orderedmultidict==1.0.1 + # via furl +orjson==3.8.2 + # via cloud-files +packaging==21.3 + # via + # pytest + # redis + # tox +pandas==1.3.5 + # via -r requirements.in +pathos==0.3.0 + # via + # cloud-files + # cloud-volume + # task-queue +pbr==5.11.0 + # via task-queue +pillow==9.3.0 + # via cloud-volume +platformdirs==2.5.4 + # via virtualenv +pluggy==1.0.0 + # via + # pytest + # tox +posix-ipc==1.0.5 + # via cloud-volume +pox==0.3.2 + # via pathos +ppft==1.7.6.6 + # via pathos +proto-plus==1.22.1 + # via + # google-cloud-bigtable + # google-cloud-datastore + # google-cloud-pubsub +protobuf==4.21.9 + # via + # -r requirements.in + # cloud-files + # cloud-volume + # google-api-core + # google-cloud-bigtable + # google-cloud-datastore + # google-cloud-pubsub + # googleapis-common-protos + # grpcio-status + # proto-plus +psutil==5.9.4 + # via cloud-volume +py==1.11.0 + # via tox +pyasn1==0.4.8 + # via + # pyasn1-modules + # rsa +pyasn1-modules==0.2.8 + # via google-auth +pycparser==2.21 + # via cffi +pyopenssl==22.1.0 + # via urllib3 +pyparsing==3.0.9 + # via packaging +pyrsistent==0.19.2 + # via jsonschema +pysimdjson==5.0.2 + # via cloud-volume +pytest==7.2.0 + # via + # compressed-segmentation + # dracopy + # task-queue +python-dateutil==2.8.2 + # via + # botocore + # cloud-volume + # pandas +python-json-logger==2.0.4 + # via -r requirements.in +python-jsonschema-objects==0.4.1 + # via cloud-volume +pytz==2022.6 + # via pandas +pyyaml==6.0 + # via -r requirements.in +redis==4.3.5 + # via + # -r requirements.in + # rq +requests==2.28.1 + # via + # -r requirements.in + # cloud-files + # cloud-volume + # codecov + # google-api-core + # google-cloud-storage + # middle-auth-client + # task-queue +rq==1.11.1 + # via -r requirements.in +rsa==4.9 + # via + # cloud-files + # google-auth +s3transfer==0.6.0 + # via boto3 +simplejpeg==1.6.5 + # via cloud-volume +six==1.16.0 + # via + # cloud-files + # cloud-volume + # flask-cors + # furl + # google-auth + # grpcio + # jsonschema + # orderedmultidict + # python-dateutil + # python-jsonschema-objects + # tox task-queue==1.0.0 -messagingclient -jsonschema<4.0 -importlib-metadata>=4.5.0 -dracopy<=0.0.19 + # via -r requirements.in +tenacity==8.1.0 + # via + # cloud-files + # cloud-volume +tomli==2.0.1 + # via + # pytest + # tox +tox==3.27.1 + # via -r requirements.in +tqdm==4.64.1 + # via + # cloud-files + # cloud-volume + # task-queue +typing-extensions==4.4.0 + # via + # async-timeout + # importlib-metadata + # redis +urllib3[brotli,secure]==1.26.13 + # via + # -r requirements.in + # botocore + # cloud-files + # cloud-volume + # requests +urllib3-secure-extra==0.1.0 + # via urllib3 +virtualenv==20.16.7 + # via tox +werkzeug==2.2.2 + # via flask +zipp==3.10.0 + # via importlib-metadata +zmesh==1.4.2 + # via -r requirements.in +zope-event==4.5.0 + # via gevent +zope-interface==5.5.2 + # via gevent +zstandard==0.19.0 + # via + # -r requirements.in + # cloud-files + +# The following packages are considered to be unsafe in a requirements file: +# setuptools From 7ab567c0022a7f52b7e2834947bc1ed0acbea9a3 Mon Sep 17 00:00:00 2001 From: Akhilesh Halageri Date: Thu, 1 Dec 2022 00:42:36 +0000 Subject: [PATCH 11/28] improve mesh worker logging --- workers/mesh_worker.py | 15 +++++++-------- 1 file changed, 7 insertions(+), 8 deletions(-) diff --git a/workers/mesh_worker.py b/workers/mesh_worker.py index a950543c3..413538aba 100644 --- a/workers/mesh_worker.py +++ b/workers/mesh_worker.py @@ -14,7 +14,8 @@ def callback(payload): from pychunkedgraph.meshing import meshgen data = pickle.loads(payload.data) - lvl2_ids = np.array(data["new_lvl2_ids"], dtype=basetypes.NODE_ID) + op_id = int(data["operation_id"]) + l2ids = np.array(data["new_lvl2_ids"], dtype=basetypes.NODE_ID) table_id = payload.attributes["table_id"] cg = ChunkedGraph(graph_id=table_id) @@ -32,25 +33,23 @@ def callback(payload): except KeyError: return - INFO_PRIORITY = 25 + INFO_HIGH = 25 logging.basicConfig( - level=INFO_PRIORITY, + level=INFO_HIGH, format="%(asctime)s %(message)s", datefmt="%m/%d/%Y %I:%M:%S %p", ) - logging.log(INFO_PRIORITY, f"Remeshing {lvl2_ids} L2 IDs in graph {table_id}") - logging.log(INFO_PRIORITY, f"stop_layer={layer}, mip={mip}, max_err={err}") - logging.log(INFO_PRIORITY, f"mesh_dir={mesh_dir}, unsharded_mesh_path={mesh_path}") + logging.log(INFO_HIGH, f"remeshing {l2ids}; graph {table_id} operation {op_id}.") meshgen.remeshing( cg, - lvl2_ids, + l2ids, stop_layer=layer, mip=mip, max_err=err, cv_sharded_mesh_dir=mesh_dir, cv_unsharded_mesh_path=mesh_path, ) - logging.log(INFO_PRIORITY, "Remeshing complete.") + logging.log(INFO_HIGH, f"remeshing complete; graph {table_id} operation {op_id}.") gc.collect() From ee8bf7b76498142e49d73ed804c573ff96b1713b Mon Sep 17 00:00:00 2001 From: Akhilesh Halageri Date: Thu, 1 Dec 2022 02:35:24 +0000 Subject: [PATCH 12/28] feat: add graph versioning --- pychunkedgraph/graph/attributes.py | 7 ++++- pychunkedgraph/graph/chunkedgraph.py | 3 +- pychunkedgraph/graph/client/base.py | 8 +++++ .../graph/client/bigtable/client.py | 29 ++++++++++++++----- pychunkedgraph/tests/test_uncategorized.py | 12 +++++--- 5 files changed, 46 insertions(+), 13 deletions(-) diff --git a/pychunkedgraph/graph/attributes.py b/pychunkedgraph/graph/attributes.py index 1d983eb55..3e48d204a 100644 --- a/pychunkedgraph/graph/attributes.py +++ b/pychunkedgraph/graph/attributes.py @@ -146,7 +146,12 @@ class Hierarchy: class GraphMeta: key = b"meta" - Meta = _Attribute(key=b"meta", family_id="0", serializer=serializers.Pickle()) + Meta = _Attribute(key=key, family_id="0", serializer=serializers.Pickle()) + + +class GraphVersion: + key = b"version" + Version = _Attribute(key=key, family_id="0", serializer=serializers.String("utf-8")) class OperationLogs: diff --git a/pychunkedgraph/graph/chunkedgraph.py b/pychunkedgraph/graph/chunkedgraph.py index 60a07cbda..3c9ec1f27 100644 --- a/pychunkedgraph/graph/chunkedgraph.py +++ b/pychunkedgraph/graph/chunkedgraph.py @@ -3,6 +3,7 @@ import datetime import numpy as np +from pychunkedgraph import __version__ from . import types from . import operation @@ -89,7 +90,7 @@ def cache(self, cache_service: CacheService): def create(self): """Creates the graph in storage client and stores meta.""" - self._client.create_graph(self._meta) + self._client.create_graph(self._meta, version=__version__) def update_meta(self, meta: ChunkedGraphMeta): """Update meta of an already existing graph.""" diff --git a/pychunkedgraph/graph/client/base.py b/pychunkedgraph/graph/client/base.py index 0ac6011fe..a66602a6a 100644 --- a/pychunkedgraph/graph/client/base.py +++ b/pychunkedgraph/graph/client/base.py @@ -12,6 +12,14 @@ class SimpleClient(ABC): def create_graph(self) -> None: """Initialize the graph and store associated meta.""" + @abstractmethod + def add_graph_version(self, version): + """Add a version to the graph.""" + + @abstractmethod + def read_graph_version(self): + """Read stored graph version.""" + @abstractmethod def update_graph_meta(self, meta): """Update stored graph meta.""" diff --git a/pychunkedgraph/graph/client/bigtable/client.py b/pychunkedgraph/graph/client/bigtable/client.py index 24e5ebee7..0cfc9f302 100644 --- a/pychunkedgraph/graph/client/bigtable/client.py +++ b/pychunkedgraph/graph/client/bigtable/client.py @@ -4,11 +4,9 @@ import logging import datetime from datetime import datetime -from datetime import timedelta import numpy as np from multiwrapper import multiprocessing_utils as mu -from google.auth import credentials from google.cloud import bigtable from google.api_core.retry import Retry from google.api_core.retry import if_exception_type @@ -34,7 +32,6 @@ from ...utils.serializers import deserialize_uint64 from ...meta import ChunkedGraphMeta from ...utils.generic import get_valid_timestamp -from ....ingest import IngestConfig class Client(bigtable.Client, ClientWithIDGen, OperationLogger): @@ -69,21 +66,39 @@ def __init__( sh.setLevel(logging.WARNING) self.logger.addHandler(sh) self._graph_meta = graph_meta + self._version = None self._max_row_key_count = config.MAX_ROW_KEY_COUNT @property def graph_meta(self): return self._graph_meta - # BASE - def create_graph(self, meta: ChunkedGraphMeta) -> None: + def create_graph(self, meta: ChunkedGraphMeta, version: str) -> None: """Initialize the graph and store associated meta.""" - if not meta.graph_config.OVERWRITE and self._table.exists(): + if self._table.exists(): ValueError(f"{self._table.table_id} already exists.") self._table.create() self._create_column_families() + self.add_graph_version(version) self.update_graph_meta(meta) + def add_graph_version(self, version: str): + assert self.read_graph_version() is None, "Graph has already been versioned." + self._version = version + row = self.mutate_row( + attributes.GraphVersion.key, + {attributes.GraphVersion.Version: version}, + ) + self.write([row]) + + def read_graph_version(self) -> str: + try: + row = self._read_byte_row(attributes.GraphVersion.key) + self._version = row[attributes.GraphVersion.Version][0].value + return self._version + except KeyError: + return None + def update_graph_meta(self, meta: ChunkedGraphMeta): self._graph_meta = meta row = self.mutate_row( @@ -799,7 +814,7 @@ def _execute_read_thread(self, args: typing.Tuple[Table, RowSet, RowFilter]): def _read( self, row_set: RowSet, row_filter: RowFilter = None ) -> typing.Dict[ - bytes, typing.Dict[attributes._Attribute, bigtable.row_data.PartialRowData] + bytes, typing.Dict[attributes._Attribute, PartialRowData] ]: """Core function to read rows from Bigtable. Uses standard Bigtable retry logic :param row_set: BigTable RowSet diff --git a/pychunkedgraph/tests/test_uncategorized.py b/pychunkedgraph/tests/test_uncategorized.py index 1855c05a3..93c41158d 100644 --- a/pychunkedgraph/tests/test_uncategorized.py +++ b/pychunkedgraph/tests/test_uncategorized.py @@ -138,7 +138,8 @@ def test_build_single_node(self, gen_graph): assert len(children) == 1 and children[0] == to_label(cg, 1, 0, 0, 0, 0) # Make sure there are not any more entries in the table - assert len(res.rows) == 1 + 1 + 1 + 1 + # include counters, meta and version rows + assert len(res.rows) == 1 + 1 + 1 + 1 + 1 @pytest.mark.timeout(30) def test_build_single_edge(self, gen_graph): @@ -190,7 +191,8 @@ def test_build_single_edge(self, gen_graph): ) # Make sure there are not any more entries in the table - assert len(res.rows) == 2 + 1 + 1 + 1 + # include counters, meta and version rows + assert len(res.rows) == 2 + 1 + 1 + 1 + 1 @pytest.mark.timeout(30) def test_build_single_across_edge(self, gen_graph): @@ -291,7 +293,8 @@ def test_build_single_across_edge(self, gen_graph): ) # Make sure there are not any more entries in the table - assert len(res.rows) == 2 + 2 + 1 + 3 + 1 + # include counters, meta and version rows + assert len(res.rows) == 2 + 2 + 1 + 3 + 1 + 1 @pytest.mark.timeout(30) def test_build_single_edge_and_single_across_edge(self, gen_graph): @@ -398,7 +401,8 @@ def test_build_single_edge_and_single_across_edge(self, gen_graph): ) # Make sure there are not any more entries in the table - assert len(res.rows) == 3 + 2 + 1 + 3 + 1 + # include counters, meta and version rows + assert len(res.rows) == 3 + 2 + 1 + 3 + 1 + 1 @pytest.mark.timeout(120) def test_build_big_graph(self, gen_graph): From e86cb88f37c2b6bce3b5185081ac2262a8a10aac Mon Sep 17 00:00:00 2001 From: Akhilesh Halageri Date: Sun, 16 Oct 2022 18:06:50 +0000 Subject: [PATCH 13/28] feat: add debug functions --- pychunkedgraph/debug/edits.py | 58 ++++++++++++++++++++++++++--------- pychunkedgraph/debug/utils.py | 43 ++++++++++++++++++++++++++ 2 files changed, 87 insertions(+), 14 deletions(-) create mode 100644 pychunkedgraph/debug/utils.py diff --git a/pychunkedgraph/debug/edits.py b/pychunkedgraph/debug/edits.py index 11ea82676..4773bc912 100644 --- a/pychunkedgraph/debug/edits.py +++ b/pychunkedgraph/debug/edits.py @@ -1,4 +1,5 @@ from typing import Union +from typing import Tuple import numpy as np @@ -13,9 +14,33 @@ def _parse_merge_payload( - cg: ChunkedGraph, user_id: str, payload: dict + cg: ChunkedGraph, user_id: str, payload: list ) -> MergeOperation: - pass + + node_ids = [] + coords = [] + for node in payload: + node_ids.append(node[0]) + coords.append(np.array(node[1:]) / cg.segmentation_resolution) + + atomic_edge = handle_supervoxel_id_lookup(cg, coords, node_ids) + chunk_coord_delta = cg.get_chunk_coordinates( + atomic_edge[0] + ) - cg.get_chunk_coordinates(atomic_edge[1]) + if np.any(np.abs(chunk_coord_delta) > 3): + raise ValueError("Chebyshev distance exceeded allowed maximum.") + + return ( + node_ids, + atomic_edge, + MergeOperation( + cg, + user_id=user_id, + added_edges=np.array(atomic_edge, dtype=np.uint64), + source_coords=coords[:1], + sink_coords=coords[1:], + ), + ) def _parse_split_payload( @@ -46,28 +71,33 @@ def _parse_split_payload( sink_coords = coords[node_idents == 1] bb_offset = (240, 240, 24) - return MulticutOperation( - cg, - user_id=user_id, - source_ids=source_ids, - sink_ids=sink_ids, - source_coords=source_coords, - sink_coords=sink_coords, - bbox_offset=bb_offset, - path_augment=True, - disallow_isolating_cut=True, + return ( + source_ids, + sink_ids, + MulticutOperation( + cg, + user_id=user_id, + source_ids=source_ids, + sink_ids=sink_ids, + source_coords=source_coords, + sink_coords=sink_coords, + bbox_offset=bb_offset, + path_augment=True, + disallow_isolating_cut=True, + ), ) def get_operation_from_request_payload( cg: ChunkedGraph, - payload: dict, + payload: Union[list, dict], split: bool, *, mincut: bool = True, user_id: str = None, -) -> GraphEditOperation: +) -> Tuple[np.ndarray, np.ndarray, GraphEditOperation]: if user_id is None: user_id = USER_ID if split: return _parse_split_payload(cg, user_id, payload, mincut=mincut) + return _parse_merge_payload(cg, user_id, payload) diff --git a/pychunkedgraph/debug/utils.py b/pychunkedgraph/debug/utils.py new file mode 100644 index 000000000..5de5d2ae0 --- /dev/null +++ b/pychunkedgraph/debug/utils.py @@ -0,0 +1,43 @@ +import numpy as np + +from ..graph import ChunkedGraph +from ..graph.utils.basetypes import NODE_ID + + +def print_attrs(d): + for k, v in d.items(): + try: + print(k.key) + except: + print(k) + try: + print(v[:2], "...") if type(v) is np.ndarray and len(v) > 2 else print(v) + except: + print(v) + + +def print_node( + cg: ChunkedGraph, + node: NODE_ID, + indent: int = 0, + stop_layer: int = 2, +) -> None: + children = cg.get_children(node) + print(f"{' ' * indent}{node}[{len(children)}]") + if cg.get_chunk_layer(node) <= stop_layer: + return + for child in children: + print_node(cg, child, indent=indent + 1, stop_layer=stop_layer) + + +def get_l2children(cg: ChunkedGraph, node: NODE_ID) -> np.ndarray: + nodes = np.array([node], dtype=NODE_ID) + layers = cg.get_chunk_layers(nodes) + assert np.all(layers > 2), "nodes must be at layers > 2" + l2children = [] + while nodes.size: + children = cg.get_children(np.sort(nodes), flatten=True) + layers = cg.get_chunk_layers(children) + l2children.append(children[layers == 2]) + nodes = children[layers > 2] + return np.concatenate(l2children) From d4b89ed53399bbbb3c0d19a56c8e7cce7a0fc0ae Mon Sep 17 00:00:00 2001 From: Akhilesh Halageri Date: Sun, 16 Oct 2022 18:11:15 +0000 Subject: [PATCH 14/28] fix: refactor and fix incorrect bounding l2 children due to skips, and speedups --- pychunkedgraph/graph/chunkedgraph.py | 129 +++++++++--------- .../graph/client/bigtable/client.py | 3 +- pychunkedgraph/graph/edges/utils.py | 46 ++----- 3 files changed, 79 insertions(+), 99 deletions(-) diff --git a/pychunkedgraph/graph/chunkedgraph.py b/pychunkedgraph/graph/chunkedgraph.py index f5909caef..e4562d888 100644 --- a/pychunkedgraph/graph/chunkedgraph.py +++ b/pychunkedgraph/graph/chunkedgraph.py @@ -265,7 +265,7 @@ def get_children( if flatten: if not node_children_d: return types.empty_1d.copy() - return np.concatenate([*node_children_d.values()]) + return np.concatenate(list(node_children_d.values())) return node_children_d def _get_children_multiple( @@ -308,7 +308,7 @@ def get_atomic_cross_edges( return self.cache.atomic_cross_edges_multiple(l2_ids) def get_cross_chunk_edges( - self, node_ids: np.ndarray, uplift=True, all_layers=False + self, node_ids: typing.Iterable, uplift=True, all_layers=False ) -> typing.Dict[np.uint64, typing.Dict[int, typing.Iterable]]: """ Cross chunk edges for `node_id` at `node_layer`. @@ -322,13 +322,14 @@ def get_cross_chunk_edges( This is because cross edges are stored only in level 2 IDs. """ result = {} + node_ids = np.array(node_ids, dtype=basetypes.NODE_ID) if not node_ids.size: return result node_l2ids_d = {} layers_ = self.get_chunk_layers(node_ids) for l in set(layers_): - node_l2ids_d.update(self._bounding_l2_children(node_ids[layers_ == l])) + node_l2ids_d.update(self._get_bounding_l2_children(node_ids[layers_ == l])) l2_edges_d_d = self.get_atomic_cross_edges( np.concatenate(list(node_l2ids_d.values())) ) @@ -850,75 +851,69 @@ def redo_operation( ).execute() # PRIVATE - def _bounding_l2_children(self, parent_ids: typing.Iterable) -> typing.Dict: - """ - Helper function to get level 2 children IDs for each parent. - `parent_ids` must be node IDs at same layer. - TODO what have i done (describe algo) - """ - from collections import defaultdict - - layers = self.get_chunk_layers(parent_ids) - assert np.all(layers == layers[0]) - - parents_layer = self.get_chunk_layer(parent_ids[0]) - chunk_coords = self.get_chunk_coordinates_multiple(parent_ids) - parent_coords_d = { - node_id: coord for node_id, coord in zip(parent_ids, chunk_coords) - } - - parent_bounding_chunk_ids = defaultdict(lambda: types.empty_1d) - parent_layer_mask = {} - parent_children_d = { - parent_id: np.array([parent_id], dtype=basetypes.NODE_ID) - for parent_id in parent_ids - } - children_layer = parents_layer - 1 - while children_layer >= 2: - parent_masked_children_d = {} - for parent_id, (X, Y, Z) in parent_coords_d.items(): - coords = chunk_utils.get_bounding_children_chunks( + def _get_bounding_chunk_ids(self, parent_chunk_ids: typing.Iterable) -> typing.Dict: + """ + Returns bounding chunk IDs at layers < parent_layer for all chunk IDs. + Dict[parent_chunk_id] = np.array(bounding_chunk_ids) + """ + parent_chunk_coords = self.get_chunk_coordinates_multiple(parent_chunk_ids) + parents_layer = self.get_chunk_layer(parent_chunk_ids[0]) + chunk_id_bchunk_ids_d = {} + for i, chunk_id in enumerate(parent_chunk_ids): + if chunk_id in chunk_id_bchunk_ids_d: + # `parent_chunk_ids` can have duplicates + # avoid redundant calculations + continue + parent_coord = parent_chunk_coords[i] + chunk_ids = [types.empty_1d] + for child_layer in range(2, parents_layer): + bcoords = chunk_utils.get_bounding_children_chunks( self.meta, parents_layer, - (X, Y, Z), - children_layer, + parent_coord, + child_layer, return_unique=False, ) - chunks_ids = chunk_utils.get_chunk_ids_from_coords( - self.meta, children_layer, coords - ) - parent_bounding_chunk_ids[parent_id] = chunks_ids - children = parent_children_d[parent_id] - layer_mask = self.get_chunk_layers(children) > children_layer - parent_layer_mask[parent_id] = layer_mask - parent_masked_children_d[parent_id] = children[layer_mask] - - children_ids = np.concatenate(list(parent_masked_children_d.values())) - child_grand_children_d = self.get_children(children_ids) - for parent_id, masked_children in parent_masked_children_d.items(): - bounding_chunk_ids = parent_bounding_chunk_ids[parent_id] - grand_children = [types.empty_1d] - for child in masked_children: - grand_children_ = child_grand_children_d[child] - mask = self.get_chunk_layers(grand_children_) == children_layer - masked_grand_children_ = grand_children_[mask] - chunk_ids = self.get_chunk_ids_from_node_ids(masked_grand_children_) - masked_grand_children_ = masked_grand_children_[ - np.in1d(chunk_ids, bounding_chunk_ids) - ] - grand_children_ = np.concatenate( - [masked_grand_children_, grand_children_[~mask]] - ) - grand_children.append(grand_children_) - grand_children = np.concatenate(grand_children) - unmasked_children = parent_children_d[parent_id] - layer_mask = parent_layer_mask[parent_id] - parent_children_d[parent_id] = np.concatenate( - [unmasked_children[~layer_mask], grand_children] + bchunks_ids = chunk_utils.get_chunk_ids_from_coords( + self.meta, child_layer, bcoords ) - children_layer -= 1 - return parent_children_d + chunk_ids.append(bchunks_ids) + chunk_id_bchunk_ids_d[chunk_id] = np.unique(np.concatenate(chunk_ids)) + return chunk_id_bchunk_ids_d + + def _get_bounding_l2_children(self, parents: typing.Iterable) -> typing.Dict: + parent_chunk_ids = self.get_chunk_ids_from_node_ids(parents) + chunk_id_bchunk_ids_d = self._get_bounding_chunk_ids(parent_chunk_ids) + + parent_descendants_d = { + _id: np.array([_id], dtype=basetypes.NODE_ID) for _id in parents + } + descendants_all = np.concatenate(list(parent_descendants_d.values())) + descendants_layers = self.get_chunk_layers(descendants_all) + layer_mask = descendants_layers > 2 + descendants_all = descendants_all[layer_mask] + + while descendants_all.size: + descendant_children_d = self.get_children(np.sort(descendants_all)) + for i, parent_id in enumerate(parents): + _descendants = parent_descendants_d[parent_id] + _layers = self.get_chunk_layers(_descendants) + _l2mask = _layers == 2 + descendants = [_descendants[_l2mask]] + for child in _descendants[~_l2mask]: + descendants.append(descendant_children_d[child]) + descendants = np.concatenate(descendants) + chunk_ids = self.get_chunk_ids_from_node_ids(descendants) + bchunk_ids = chunk_id_bchunk_ids_d[parent_chunk_ids[i]] + bounding_descendants = descendants[np.in1d(chunk_ids, bchunk_ids)] + parent_descendants_d[parent_id] = bounding_descendants + + descendants_all = np.concatenate(list(parent_descendants_d.values())) + descendants_layers = self.get_chunk_layers(descendants_all) + layer_mask = descendants_layers > 2 + descendants_all = descendants_all[layer_mask] + return parent_descendants_d # HELPERS / WRAPPERS @@ -960,6 +955,8 @@ def get_chunk_coordinates(self, node_or_chunk_id: basetypes.NODE_ID): def get_chunk_coordinates_multiple(self, node_or_chunk_ids: typing.Sequence): node_or_chunk_ids = np.array(node_or_chunk_ids, dtype=basetypes.NODE_ID) + layers = self.get_chunk_layers(node_or_chunk_ids) + assert np.all(layers == layers[0]), "All IDs must have the same layer." return chunk_utils.get_chunk_coordinates_multiple(self.meta, node_or_chunk_ids) def get_chunk_id( diff --git a/pychunkedgraph/graph/client/bigtable/client.py b/pychunkedgraph/graph/client/bigtable/client.py index 4d9142be0..0aca696d4 100644 --- a/pychunkedgraph/graph/client/bigtable/client.py +++ b/pychunkedgraph/graph/client/bigtable/client.py @@ -697,8 +697,7 @@ def _read_byte_rows( # Create filters: Rows row_set = RowSet() if row_keys is not None: - for row_key in row_keys: - row_set.add_row_key(row_key) + row_set.row_keys = list(row_keys) elif start_key is not None and end_key is not None: row_set.add_row_range_from_keys( start_key=start_key, diff --git a/pychunkedgraph/graph/edges/utils.py b/pychunkedgraph/graph/edges/utils.py index 5d602fa64..cc2dfa909 100644 --- a/pychunkedgraph/graph/edges/utils.py +++ b/pychunkedgraph/graph/edges/utils.py @@ -2,9 +2,7 @@ helper functions for edge stuff """ -from collections import defaultdict from typing import Dict -from typing import List from typing import Tuple from typing import Iterable from typing import Callable @@ -12,19 +10,15 @@ import numpy as np - from . import Edges from . import EDGE_TYPES from ..types import empty_2d from ..utils import basetypes from ..chunks import utils as chunk_utils from ..meta import ChunkedGraphMeta -from ..connectivity.search import check_reachability -from ..utils.flatgraph import build_gt_graph -from ...utils.general import reverse_dictionary -def concatenate_chunk_edges(chunk_edge_dicts: List) -> Dict: +def concatenate_chunk_edges(chunk_edge_dicts: Iterable) -> Dict: """combine edge_dicts of multiple chunks into one edge_dict""" edges_dict = {} for edge_type in EDGE_TYPES: @@ -49,31 +43,22 @@ def concatenate_chunk_edges(chunk_edge_dicts: List) -> Dict: return edges_dict -def concatenate_cross_edge_dicts(cross_edge_dicts: Iterable) -> Dict: - """Combines multiple cross edge dicts.""" - # print(cross_edge_dicts) - result_d = {} - for cross_edge_d in cross_edge_dicts: - result_d = merge_cross_edge_dicts_single(result_d, cross_edge_d) - return result_d +def concatenate_cross_edge_dicts(edges_ds: Iterable[Dict]) -> Dict: + """Combines cross chunk edge dicts of form {layer id : edge list}.""" + from collections import defaultdict + result_d = defaultdict(list) -def merge_cross_edge_dicts_single(x_edges_d1: Dict, x_edges_d2: Dict) -> Dict: - """Combines two cross chunk edge dicts of form {layer id : edge list}.""" - result_d = {} - if not x_edges_d1 and not x_edges_d2: - return result_d - layers = np.unique(list(x_edges_d1.keys()) + list(x_edges_d2.keys())) - for layer in range(2, max(layers) + 1): - edges1 = x_edges_d1.get(layer, empty_2d) - edges2 = x_edges_d2.get(layer, empty_2d) - edges1 = np.array(edges1, dtype=basetypes.NODE_ID) - edges2 = np.array(edges2, dtype=basetypes.NODE_ID) - result_d[layer] = np.concatenate([edges1, edges2]) + for edges_d in edges_ds: + for layer, edges in edges_d.items(): + result_d[layer].append(edges) + + for layer, edge_lists in result_d.items(): + result_d[layer] = np.concatenate(edge_lists) return result_d -def merge_cross_edge_dicts_multiple(x_edges_d1: Dict, x_edges_d2: Dict) -> Dict: +def merge_cross_edge_dicts(x_edges_d1: Dict, x_edges_d2: Dict) -> Dict: """ Combines two cross chunk dictionaries of form {node_id: {layer id : edge list}}. @@ -81,9 +66,8 @@ def merge_cross_edge_dicts_multiple(x_edges_d1: Dict, x_edges_d2: Dict) -> Dict: node_ids = np.unique(list(x_edges_d1.keys()) + list(x_edges_d2.keys())) result_d = {} for node_id in node_ids: - result_d[node_id] = merge_cross_edge_dicts_single( - x_edges_d1.get(node_id, {}), x_edges_d2.get(node_id, {}) - ) + cross_edge_ds = [x_edges_d1.get(node_id, {}), x_edges_d2.get(node_id, {})] + result_d[node_id] = concatenate_cross_edge_dicts(cross_edge_ds) return result_d @@ -179,7 +163,7 @@ def filter_min_layer_cross_edges( def filter_min_layer_cross_edges_multiple( - meta: ChunkedGraphMeta, l2id_atomic_cross_edges_ds: List, node_layer: int = 2 + meta: ChunkedGraphMeta, l2id_atomic_cross_edges_ds: Iterable, node_layer: int = 2 ) -> Tuple[int, Iterable]: """ Given a list of dicts of cross chunk edges [{layer: edges}] From c0e6abb83953c0ecd8c7c0bf81e99b1f323a1611 Mon Sep 17 00:00:00 2001 From: Akhilesh Halageri Date: Sun, 16 Oct 2022 18:11:34 +0000 Subject: [PATCH 15/28] code cleanup --- pychunkedgraph/graph/edits.py | 46 +++++++++++++-------------- pychunkedgraph/graph/utils/generic.py | 6 ---- pychunkedgraph/graph/utils/inspect.py | 20 ------------ 3 files changed, 22 insertions(+), 50 deletions(-) delete mode 100644 pychunkedgraph/graph/utils/inspect.py diff --git a/pychunkedgraph/graph/edits.py b/pychunkedgraph/graph/edits.py index ca51bebe8..1fb5675f6 100644 --- a/pychunkedgraph/graph/edits.py +++ b/pychunkedgraph/graph/edits.py @@ -4,7 +4,6 @@ from typing import List from typing import Tuple from typing import Iterable -from typing import Sequence from collections import defaultdict from . import types @@ -14,13 +13,7 @@ from .utils import flatgraph from .utils.context_managers import TimeIt from .utils.serializers import serialize_uint64 -from .connectivity.nodes import edge_exists -from .connectivity.search import check_reachability -from .edges.utils import filter_min_layer_cross_edges -from .edges.utils import concatenate_cross_edge_dicts -from .edges.utils import merge_cross_edge_dicts_multiple from ..utils.general import in2d -from ..utils.general import reverse_dictionary def _init_old_hierarchy(cg, l2ids: np.ndarray, parent_ts: datetime.datetime = None): @@ -190,6 +183,9 @@ def add_edges( rows: Iterable = None, allow_same_segment_merge=False, ): + from .edges.utils import concatenate_cross_edge_dicts + from .edges.utils import merge_cross_edge_dicts + # TODO add docs if rows == None: rows = [] @@ -206,7 +202,7 @@ def add_edges( cg, l2ids, parent_ts=parent_ts ) atomic_children_d = cg.get_children(l2ids) - atomic_cross_edges_d = merge_cross_edge_dicts_multiple( + atomic_cross_edges_d = merge_cross_edge_dicts( cg.get_atomic_cross_edges(l2ids), l2_atomic_cross_edges_d ) @@ -240,7 +236,10 @@ def add_edges( time_stamp=time_stamp, parent_ts=parent_ts, ) - return create_parents.run(), new_l2_ids, rows + create_parents.create_new_entries() + with TimeIt("create_parents.run()"): + new_roots = create_parents.run() + new_entries = create_parents.create_new_entries() + return new_roots, new_l2_ids, new_entries def _process_l2_agglomeration( @@ -341,8 +340,7 @@ def remove_edges( ) with TimeIt("create_parents.run()"): new_roots = create_parents.run() - with TimeIt("create_parents.create_new_entries()"): - new_entries = create_parents.create_new_entries() + new_entries = create_parents.create_new_entries() return new_roots, new_l2_ids, new_entries @@ -386,12 +384,19 @@ def _update_id_lineage( self._new_old_id_d[parent].add(old_id) self._old_new_id_d[old_id].add(parent) + def _get_old_ids(self, new_ids): + old_ids = [ + np.array(list(self._new_old_id_d[id_]), dtype=basetypes.NODE_ID) + for id_ in new_ids + ] + return np.concatenate(old_ids) + def _map_sv_to_parent(self, node_ids, layer): sv_parent_d = {} sv_cross_edges = [types.empty_2d] for id_ in node_ids: edges_ = self._cross_edges_d[id_].get(layer, types.empty_2d) - sv_parent_d.update(dict(zip(edges_[:, 0].tolist(), [id_] * len(edges_)))) + sv_parent_d.update(dict(zip(edges_[:, 0], [id_] * len(edges_)))) sv_cross_edges.append(edges_) return sv_parent_d, np.concatenate(sv_cross_edges) @@ -427,11 +432,7 @@ def _get_layer_node_ids( self, new_ids: np.ndarray, layer: int ) -> Tuple[np.ndarray, np.ndarray]: # get old identities of new IDs - old_ids = [ - np.array(list(self._new_old_id_d[id_]), dtype=basetypes.NODE_ID) - for id_ in new_ids - ] - old_ids = np.concatenate(old_ids) + old_ids = self._get_old_ids(new_ids) # get their parents, then children of those parents node_ids = self.cg.get_children( np.unique( @@ -500,17 +501,14 @@ def run(self) -> Iterable: for layer in range(2, self.cg.meta.layer_count): if len(self._new_ids_d[layer]) == 0: continue - # with TimeIt(f"self._create_new_parents(layer) {layer}"): + # with TimeIt(f"_create_new_parents {layer}"): self._create_new_parents(layer) return self._new_ids_d[self.cg.meta.layer_count] def _update_root_id_lineage(self): new_root_ids = self._new_ids_d[self.cg.meta.layer_count] - former_root_ids = [ - np.array(list(self._new_old_id_d[id_]), dtype=basetypes.NODE_ID) - for id_ in new_root_ids - ] - former_root_ids = np.unique(np.concatenate(former_root_ids)) + former_root_ids = self._get_old_ids(new_root_ids) + former_root_ids = np.unique(former_root_ids) assert ( len(former_root_ids) < 2 or len(new_root_ids) < 2 ), "Something went wrong." @@ -553,7 +551,7 @@ def _get_atomic_cross_edges_val_dict(self): val_dicts[id_] = val_dict return val_dicts - def create_new_entries(self): + def create_new_entries(self) -> List: rows = [] val_dicts = self._get_atomic_cross_edges_val_dict() for layer in range(2, self.cg.meta.layer_count + 1): diff --git a/pychunkedgraph/graph/utils/generic.py b/pychunkedgraph/graph/utils/generic.py index fc19b9e06..9a2b6f979 100644 --- a/pychunkedgraph/graph/utils/generic.py +++ b/pychunkedgraph/graph/utils/generic.py @@ -11,19 +11,13 @@ from typing import Union from typing import Sequence from typing import Tuple -from itertools import product from collections import defaultdict import numpy as np import pandas as pd import pytz -from google.cloud import bigtable -from google.cloud.bigtable.row_filters import RowFilter -from cloudvolume import CloudVolume -from .. import types from ..chunks import utils as chunk_utils -from . import serializers def compute_indices_pandas(data) -> pd.Series: diff --git a/pychunkedgraph/graph/utils/inspect.py b/pychunkedgraph/graph/utils/inspect.py deleted file mode 100644 index 0470a2da7..000000000 --- a/pychunkedgraph/graph/utils/inspect.py +++ /dev/null @@ -1,20 +0,0 @@ -"""Helper functions to visualize/inspect tree hierarchy.""" - - -def dfs_print_node(cg, node, limit=2, sep=" |", count_only=False): - """ - `limit` stop at this layer - """ - leaves = 0 - stack = [(node, 0)] - while stack: - node, indent = stack.pop() - children = cg.get_children(node) - if not count_only: - print(sep * indent, node, f"({cg.get_chunk_layer(node)}, {len(children)})") - if cg.get_chunk_layer(children[0]) < limit: - leaves += len(children) - continue - for c in children: - stack.append((c, indent + 1)) - print("leaves count", leaves) From 31b7db22ce4951a8eb6a32313d3043d03405bb22 Mon Sep 17 00:00:00 2001 From: Akhilesh Halageri Date: Sun, 16 Oct 2022 21:46:26 +0000 Subject: [PATCH 16/28] decide if unique on number of IDs involved --- pychunkedgraph/graph/chunkedgraph.py | 22 +++++++++++++++++----- 1 file changed, 17 insertions(+), 5 deletions(-) diff --git a/pychunkedgraph/graph/chunkedgraph.py b/pychunkedgraph/graph/chunkedgraph.py index e4562d888..60a07cbda 100644 --- a/pychunkedgraph/graph/chunkedgraph.py +++ b/pychunkedgraph/graph/chunkedgraph.py @@ -328,8 +328,11 @@ def get_cross_chunk_edges( node_l2ids_d = {} layers_ = self.get_chunk_layers(node_ids) + # with TimeIt(f"_get_bounding_l2_children {node_ids.size}"): for l in set(layers_): - node_l2ids_d.update(self._get_bounding_l2_children(node_ids[layers_ == l])) + node_l2ids_d.update( + self._get_bounding_l2_children(node_ids[layers_ == l]) + ) l2_edges_d_d = self.get_atomic_cross_edges( np.concatenate(list(node_l2ids_d.values())) ) @@ -852,7 +855,11 @@ def redo_operation( # PRIVATE - def _get_bounding_chunk_ids(self, parent_chunk_ids: typing.Iterable) -> typing.Dict: + def _get_bounding_chunk_ids( + self, + parent_chunk_ids: typing.Iterable, + unique: bool = False, + ) -> typing.Dict: """ Returns bounding chunk IDs at layers < parent_layer for all chunk IDs. Dict[parent_chunk_id] = np.array(bounding_chunk_ids) @@ -879,12 +886,17 @@ def _get_bounding_chunk_ids(self, parent_chunk_ids: typing.Iterable) -> typing.D self.meta, child_layer, bcoords ) chunk_ids.append(bchunks_ids) - chunk_id_bchunk_ids_d[chunk_id] = np.unique(np.concatenate(chunk_ids)) + chunk_ids = np.concatenate(chunk_ids) + if unique: + chunk_ids = np.unique(chunk_ids) + chunk_id_bchunk_ids_d[chunk_id] = chunk_ids return chunk_id_bchunk_ids_d def _get_bounding_l2_children(self, parents: typing.Iterable) -> typing.Dict: parent_chunk_ids = self.get_chunk_ids_from_node_ids(parents) - chunk_id_bchunk_ids_d = self._get_bounding_chunk_ids(parent_chunk_ids) + chunk_id_bchunk_ids_d = self._get_bounding_chunk_ids( + parent_chunk_ids, unique=len(parents) >= 200 + ) parent_descendants_d = { _id: np.array([_id], dtype=basetypes.NODE_ID) for _id in parents @@ -895,7 +907,7 @@ def _get_bounding_l2_children(self, parents: typing.Iterable) -> typing.Dict: descendants_all = descendants_all[layer_mask] while descendants_all.size: - descendant_children_d = self.get_children(np.sort(descendants_all)) + descendant_children_d = self.get_children(descendants_all) for i, parent_id in enumerate(parents): _descendants = parent_descendants_d[parent_id] _layers = self.get_chunk_layers(_descendants) From 02ee442e029620b04aaa5359a6f7a33b36a68b7f Mon Sep 17 00:00:00 2001 From: Akhilesh Halageri Date: Tue, 18 Oct 2022 15:39:03 +0000 Subject: [PATCH 17/28] sort IDs for faster reads --- pychunkedgraph/debug/utils.py | 2 +- pychunkedgraph/graph/client/bigtable/client.py | 8 ++++++-- pychunkedgraph/graph/subgraph.py | 2 +- 3 files changed, 8 insertions(+), 4 deletions(-) diff --git a/pychunkedgraph/debug/utils.py b/pychunkedgraph/debug/utils.py index 5de5d2ae0..179f50aef 100644 --- a/pychunkedgraph/debug/utils.py +++ b/pychunkedgraph/debug/utils.py @@ -36,7 +36,7 @@ def get_l2children(cg: ChunkedGraph, node: NODE_ID) -> np.ndarray: assert np.all(layers > 2), "nodes must be at layers > 2" l2children = [] while nodes.size: - children = cg.get_children(np.sort(nodes), flatten=True) + children = cg.get_children(nodes, flatten=True) layers = cg.get_chunk_layers(children) l2children.append(children[layers == 2]) nodes = children[layers > 2] diff --git a/pychunkedgraph/graph/client/bigtable/client.py b/pychunkedgraph/graph/client/bigtable/client.py index 0aca696d4..24e5ebee7 100644 --- a/pychunkedgraph/graph/client/bigtable/client.py +++ b/pychunkedgraph/graph/client/bigtable/client.py @@ -114,6 +114,10 @@ def read_nodes( Read nodes and their properties. Accepts a range of node IDs or specific node IDs. """ + if node_ids is not None and len(node_ids) > self._max_row_key_count: + # bigtable reading is faster + # when all IDs in a block are within a range + node_ids = np.sort(node_ids) rows = self._read_byte_rows( start_key=serialize_uint64(start_id, fake_edges=fake_edges) if start_id is not None @@ -564,7 +568,7 @@ def get_max_node_id( ) -> basetypes.NODE_ID: """Gets the current maximum segment ID in the chunk.""" if root_chunk: - n_counters = np.uint64(2 ** 8) + n_counters = np.uint64(2**8) max_value = 0 for counter in range(n_counters): row = self._read_byte_row( @@ -623,7 +627,7 @@ def _get_root_segment_ids_range( self, chunk_id: basetypes.CHUNK_ID, size: int = 1, counter: int = None ) -> np.ndarray: """Return unique segment ID for the root chunk.""" - n_counters = np.uint64(2 ** 8) + n_counters = np.uint64(2**8) counter = ( np.uint64(counter % n_counters) if counter diff --git a/pychunkedgraph/graph/subgraph.py b/pychunkedgraph/graph/subgraph.py index 0759abcb8..ab2593175 100644 --- a/pychunkedgraph/graph/subgraph.py +++ b/pychunkedgraph/graph/subgraph.py @@ -201,7 +201,7 @@ def _get_dict_key(raw_key): def _get_subgraph_multiple_nodes_threaded( node_ids_batch: Iterable[np.uint64], ) -> List[np.uint64]: - children = cg.get_children(np.sort(node_ids_batch)) + children = cg.get_children(node_ids_batch) if bounding_box is not None: filtered_children = {} for node_id, nodes_children in children.items(): From a718450587f9312e34c95fc30578d3563d4de355 Mon Sep 17 00:00:00 2001 From: Akhilesh Halageri Date: Fri, 4 Nov 2022 20:54:47 +0000 Subject: [PATCH 18/28] test pyfqmr --- pychunkedgraph/meshing/meshgen.py | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/pychunkedgraph/meshing/meshgen.py b/pychunkedgraph/meshing/meshgen.py index 511b7352d..588685527 100644 --- a/pychunkedgraph/meshing/meshgen.py +++ b/pychunkedgraph/meshing/meshgen.py @@ -418,7 +418,7 @@ def _get_root_ids(args): start_id, end_id = args root_ids[start_id:end_id] = cg.get_roots( - combined_ids[start_id:end_id], stop_layer=stop_layer, time_stamp=time_stamp + combined_ids[start_id:end_id], stop_layer=stop_layer, time_stamp=time_stamp, fail_to_zero=True ) rr = cg.range_read_chunk( @@ -1191,6 +1191,22 @@ def chunk_stitch_remeshing_task( cg, old_fragments, chunk_id, mip, high_padding ) + import pyfqmr + simplifier = pyfqmr.Simplify() + simplifier.setMesh( + new_fragment["vertices"].reshape(new_fragment["num_vertices"], 3), + new_fragment["faces"].reshape(-1, 3) + ) + simplifier.simplify_mesh( + target_count=4, + aggressiveness=5.0, + preserve_border=True, + verbose=False, + ) + v,f,_ = simplifier.getMesh() + new_fragment["vertices"] = v.flatten() + new_fragment["faces"] = f.flatten() + try: print(f'num_vertices = {len(new_fragment["vertices"])}') new_fragment_b = DracoPy.encode_mesh_to_buffer( From 59ada59834f044f5085113dbf25f93d6a5603dd0 Mon Sep 17 00:00:00 2001 From: Akhilesh Halageri Date: Sat, 19 Nov 2022 21:24:25 +0000 Subject: [PATCH 19/28] fix: log L2 IDs being remeshed --- workers/mesh_worker.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/workers/mesh_worker.py b/workers/mesh_worker.py index b7bc3a913..a950543c3 100644 --- a/workers/mesh_worker.py +++ b/workers/mesh_worker.py @@ -38,7 +38,7 @@ def callback(payload): format="%(asctime)s %(message)s", datefmt="%m/%d/%Y %I:%M:%S %p", ) - logging.log(INFO_PRIORITY, f"Remeshing {lvl2_ids.size} L2 IDs in graph {table_id}") + logging.log(INFO_PRIORITY, f"Remeshing {lvl2_ids} L2 IDs in graph {table_id}") logging.log(INFO_PRIORITY, f"stop_layer={layer}, mip={mip}, max_err={err}") logging.log(INFO_PRIORITY, f"mesh_dir={mesh_dir}, unsharded_mesh_path={mesh_path}") meshgen.remeshing( From 8ddee32e0a93573c6b84d8f7d4f2820e74c036d4 Mon Sep 17 00:00:00 2001 From: Akhilesh Halageri Date: Tue, 29 Nov 2022 16:36:04 +0000 Subject: [PATCH 20/28] undo meshgen changes, handled in another pr --- pychunkedgraph/meshing/meshgen.py | 18 +----------------- 1 file changed, 1 insertion(+), 17 deletions(-) diff --git a/pychunkedgraph/meshing/meshgen.py b/pychunkedgraph/meshing/meshgen.py index 588685527..511b7352d 100644 --- a/pychunkedgraph/meshing/meshgen.py +++ b/pychunkedgraph/meshing/meshgen.py @@ -418,7 +418,7 @@ def _get_root_ids(args): start_id, end_id = args root_ids[start_id:end_id] = cg.get_roots( - combined_ids[start_id:end_id], stop_layer=stop_layer, time_stamp=time_stamp, fail_to_zero=True + combined_ids[start_id:end_id], stop_layer=stop_layer, time_stamp=time_stamp ) rr = cg.range_read_chunk( @@ -1191,22 +1191,6 @@ def chunk_stitch_remeshing_task( cg, old_fragments, chunk_id, mip, high_padding ) - import pyfqmr - simplifier = pyfqmr.Simplify() - simplifier.setMesh( - new_fragment["vertices"].reshape(new_fragment["num_vertices"], 3), - new_fragment["faces"].reshape(-1, 3) - ) - simplifier.simplify_mesh( - target_count=4, - aggressiveness=5.0, - preserve_border=True, - verbose=False, - ) - v,f,_ = simplifier.getMesh() - new_fragment["vertices"] = v.flatten() - new_fragment["faces"] = f.flatten() - try: print(f'num_vertices = {len(new_fragment["vertices"])}') new_fragment_b = DracoPy.encode_mesh_to_buffer( From e855aafccfa2403d87a59e5484ca519a676fa6df Mon Sep 17 00:00:00 2001 From: Akhilesh Halageri Date: Thu, 1 Dec 2022 00:42:36 +0000 Subject: [PATCH 21/28] improve mesh worker logging --- workers/mesh_worker.py | 15 +++++++-------- 1 file changed, 7 insertions(+), 8 deletions(-) diff --git a/workers/mesh_worker.py b/workers/mesh_worker.py index a950543c3..413538aba 100644 --- a/workers/mesh_worker.py +++ b/workers/mesh_worker.py @@ -14,7 +14,8 @@ def callback(payload): from pychunkedgraph.meshing import meshgen data = pickle.loads(payload.data) - lvl2_ids = np.array(data["new_lvl2_ids"], dtype=basetypes.NODE_ID) + op_id = int(data["operation_id"]) + l2ids = np.array(data["new_lvl2_ids"], dtype=basetypes.NODE_ID) table_id = payload.attributes["table_id"] cg = ChunkedGraph(graph_id=table_id) @@ -32,25 +33,23 @@ def callback(payload): except KeyError: return - INFO_PRIORITY = 25 + INFO_HIGH = 25 logging.basicConfig( - level=INFO_PRIORITY, + level=INFO_HIGH, format="%(asctime)s %(message)s", datefmt="%m/%d/%Y %I:%M:%S %p", ) - logging.log(INFO_PRIORITY, f"Remeshing {lvl2_ids} L2 IDs in graph {table_id}") - logging.log(INFO_PRIORITY, f"stop_layer={layer}, mip={mip}, max_err={err}") - logging.log(INFO_PRIORITY, f"mesh_dir={mesh_dir}, unsharded_mesh_path={mesh_path}") + logging.log(INFO_HIGH, f"remeshing {l2ids}; graph {table_id} operation {op_id}.") meshgen.remeshing( cg, - lvl2_ids, + l2ids, stop_layer=layer, mip=mip, max_err=err, cv_sharded_mesh_dir=mesh_dir, cv_unsharded_mesh_path=mesh_path, ) - logging.log(INFO_PRIORITY, "Remeshing complete.") + logging.log(INFO_HIGH, f"remeshing complete; graph {table_id} operation {op_id}.") gc.collect() From 81164d0b5b6147c5e80dea7e26ccaee4fa390d6a Mon Sep 17 00:00:00 2001 From: Akhilesh Halageri Date: Thu, 1 Dec 2022 02:35:24 +0000 Subject: [PATCH 22/28] feat: add graph versioning --- pychunkedgraph/graph/attributes.py | 7 ++++- pychunkedgraph/graph/chunkedgraph.py | 3 +- pychunkedgraph/graph/client/base.py | 8 +++++ .../graph/client/bigtable/client.py | 29 ++++++++++++++----- pychunkedgraph/tests/test_uncategorized.py | 12 +++++--- 5 files changed, 46 insertions(+), 13 deletions(-) diff --git a/pychunkedgraph/graph/attributes.py b/pychunkedgraph/graph/attributes.py index 1d983eb55..3e48d204a 100644 --- a/pychunkedgraph/graph/attributes.py +++ b/pychunkedgraph/graph/attributes.py @@ -146,7 +146,12 @@ class Hierarchy: class GraphMeta: key = b"meta" - Meta = _Attribute(key=b"meta", family_id="0", serializer=serializers.Pickle()) + Meta = _Attribute(key=key, family_id="0", serializer=serializers.Pickle()) + + +class GraphVersion: + key = b"version" + Version = _Attribute(key=key, family_id="0", serializer=serializers.String("utf-8")) class OperationLogs: diff --git a/pychunkedgraph/graph/chunkedgraph.py b/pychunkedgraph/graph/chunkedgraph.py index 60a07cbda..3c9ec1f27 100644 --- a/pychunkedgraph/graph/chunkedgraph.py +++ b/pychunkedgraph/graph/chunkedgraph.py @@ -3,6 +3,7 @@ import datetime import numpy as np +from pychunkedgraph import __version__ from . import types from . import operation @@ -89,7 +90,7 @@ def cache(self, cache_service: CacheService): def create(self): """Creates the graph in storage client and stores meta.""" - self._client.create_graph(self._meta) + self._client.create_graph(self._meta, version=__version__) def update_meta(self, meta: ChunkedGraphMeta): """Update meta of an already existing graph.""" diff --git a/pychunkedgraph/graph/client/base.py b/pychunkedgraph/graph/client/base.py index 0ac6011fe..a66602a6a 100644 --- a/pychunkedgraph/graph/client/base.py +++ b/pychunkedgraph/graph/client/base.py @@ -12,6 +12,14 @@ class SimpleClient(ABC): def create_graph(self) -> None: """Initialize the graph and store associated meta.""" + @abstractmethod + def add_graph_version(self, version): + """Add a version to the graph.""" + + @abstractmethod + def read_graph_version(self): + """Read stored graph version.""" + @abstractmethod def update_graph_meta(self, meta): """Update stored graph meta.""" diff --git a/pychunkedgraph/graph/client/bigtable/client.py b/pychunkedgraph/graph/client/bigtable/client.py index 24e5ebee7..0cfc9f302 100644 --- a/pychunkedgraph/graph/client/bigtable/client.py +++ b/pychunkedgraph/graph/client/bigtable/client.py @@ -4,11 +4,9 @@ import logging import datetime from datetime import datetime -from datetime import timedelta import numpy as np from multiwrapper import multiprocessing_utils as mu -from google.auth import credentials from google.cloud import bigtable from google.api_core.retry import Retry from google.api_core.retry import if_exception_type @@ -34,7 +32,6 @@ from ...utils.serializers import deserialize_uint64 from ...meta import ChunkedGraphMeta from ...utils.generic import get_valid_timestamp -from ....ingest import IngestConfig class Client(bigtable.Client, ClientWithIDGen, OperationLogger): @@ -69,21 +66,39 @@ def __init__( sh.setLevel(logging.WARNING) self.logger.addHandler(sh) self._graph_meta = graph_meta + self._version = None self._max_row_key_count = config.MAX_ROW_KEY_COUNT @property def graph_meta(self): return self._graph_meta - # BASE - def create_graph(self, meta: ChunkedGraphMeta) -> None: + def create_graph(self, meta: ChunkedGraphMeta, version: str) -> None: """Initialize the graph and store associated meta.""" - if not meta.graph_config.OVERWRITE and self._table.exists(): + if self._table.exists(): ValueError(f"{self._table.table_id} already exists.") self._table.create() self._create_column_families() + self.add_graph_version(version) self.update_graph_meta(meta) + def add_graph_version(self, version: str): + assert self.read_graph_version() is None, "Graph has already been versioned." + self._version = version + row = self.mutate_row( + attributes.GraphVersion.key, + {attributes.GraphVersion.Version: version}, + ) + self.write([row]) + + def read_graph_version(self) -> str: + try: + row = self._read_byte_row(attributes.GraphVersion.key) + self._version = row[attributes.GraphVersion.Version][0].value + return self._version + except KeyError: + return None + def update_graph_meta(self, meta: ChunkedGraphMeta): self._graph_meta = meta row = self.mutate_row( @@ -799,7 +814,7 @@ def _execute_read_thread(self, args: typing.Tuple[Table, RowSet, RowFilter]): def _read( self, row_set: RowSet, row_filter: RowFilter = None ) -> typing.Dict[ - bytes, typing.Dict[attributes._Attribute, bigtable.row_data.PartialRowData] + bytes, typing.Dict[attributes._Attribute, PartialRowData] ]: """Core function to read rows from Bigtable. Uses standard Bigtable retry logic :param row_set: BigTable RowSet diff --git a/pychunkedgraph/tests/test_uncategorized.py b/pychunkedgraph/tests/test_uncategorized.py index 1855c05a3..93c41158d 100644 --- a/pychunkedgraph/tests/test_uncategorized.py +++ b/pychunkedgraph/tests/test_uncategorized.py @@ -138,7 +138,8 @@ def test_build_single_node(self, gen_graph): assert len(children) == 1 and children[0] == to_label(cg, 1, 0, 0, 0, 0) # Make sure there are not any more entries in the table - assert len(res.rows) == 1 + 1 + 1 + 1 + # include counters, meta and version rows + assert len(res.rows) == 1 + 1 + 1 + 1 + 1 @pytest.mark.timeout(30) def test_build_single_edge(self, gen_graph): @@ -190,7 +191,8 @@ def test_build_single_edge(self, gen_graph): ) # Make sure there are not any more entries in the table - assert len(res.rows) == 2 + 1 + 1 + 1 + # include counters, meta and version rows + assert len(res.rows) == 2 + 1 + 1 + 1 + 1 @pytest.mark.timeout(30) def test_build_single_across_edge(self, gen_graph): @@ -291,7 +293,8 @@ def test_build_single_across_edge(self, gen_graph): ) # Make sure there are not any more entries in the table - assert len(res.rows) == 2 + 2 + 1 + 3 + 1 + # include counters, meta and version rows + assert len(res.rows) == 2 + 2 + 1 + 3 + 1 + 1 @pytest.mark.timeout(30) def test_build_single_edge_and_single_across_edge(self, gen_graph): @@ -398,7 +401,8 @@ def test_build_single_edge_and_single_across_edge(self, gen_graph): ) # Make sure there are not any more entries in the table - assert len(res.rows) == 3 + 2 + 1 + 3 + 1 + # include counters, meta and version rows + assert len(res.rows) == 3 + 2 + 1 + 3 + 1 + 1 @pytest.mark.timeout(120) def test_build_big_graph(self, gen_graph): From dc98765b546b1d98ff67120c5e0556c5c48219e5 Mon Sep 17 00:00:00 2001 From: Akhilesh Halageri Date: Thu, 1 Dec 2022 03:11:49 +0000 Subject: [PATCH 23/28] ensure correct version in frontend --- pychunkedgraph/app/app_utils.py | 50 +++++++++++++++++++-------------- 1 file changed, 29 insertions(+), 21 deletions(-) diff --git a/pychunkedgraph/app/app_utils.py b/pychunkedgraph/app/app_utils.py index 4367cd4ca..778843bba 100644 --- a/pychunkedgraph/app/app_utils.py +++ b/pychunkedgraph/app/app_utils.py @@ -2,6 +2,7 @@ import sys import os from typing import Sequence +from time import gmtime import numpy as np from flask import current_app, json, request @@ -9,15 +10,14 @@ from google.auth import default as default_creds from google.cloud import bigtable, datastore +from pychunkedgraph import __version__ from pychunkedgraph.graph import ChunkedGraph from pychunkedgraph.logging import flask_log_db, jsonformatter -from pychunkedgraph.graph import ( - exceptions as cg_exceptions, -) +from pychunkedgraph.graph.client import get_default_client_info +from pychunkedgraph.graph import exceptions as cg_exceptions from functools import wraps from werkzeug.datastructures import ImmutableMultiDict - import networkx as nx from scipy import spatial import requests @@ -176,28 +176,14 @@ def get_datastore_client(config): return client -def get_cg(table_id, skip_cache: bool = False): - from time import gmtime - from pychunkedgraph.graph.client import get_default_client_info - - assert table_id in current_app.config["PCG_GRAPH_IDS"] - - current_app.table_id = table_id - if skip_cache is False: - try: - return CACHE[table_id] - except KeyError: - pass - +def setup_logger(table_id: str): instance_id = current_app.config["CHUNKGRAPH_INSTANCE_ID"] - # Create ChunkedGraph logging logger = logging.getLogger(f"{instance_id}/{table_id}") logger.setLevel(current_app.config["LOGGING_LEVEL"]) # prevent duplicate logs from Flasks(?) parent logger logger.propagate = False - handler = logging.StreamHandler(sys.stdout) handler.setLevel(current_app.config["LOGGING_LEVEL"]) formatter = jsonformatter.JsonFormatter( @@ -206,13 +192,35 @@ def get_cg(table_id, skip_cache: bool = False): ) formatter.converter = gmtime handler.setFormatter(formatter) - logger.addHandler(handler) - # Create ChunkedGraph + +def ensure_correct_version(cg: ChunkedGraph) -> bool: + version = cg.client.read_graph_version() + current_major_version = int(__version__.split(".")[0]) + try: + graph_major_version = int(version.split(".")[0]) + return graph_major_version == current_major_version + except (AttributeError, TypeError): + return False + + +def get_cg(table_id, skip_cache: bool = False): + current_app.table_id = table_id + if skip_cache is False: + try: + return CACHE[table_id] + except KeyError: + pass + + setup_logger(table_id) cg = ChunkedGraph(graph_id=table_id, client_info=get_default_client_info()) if skip_cache is False: CACHE[table_id] = cg + + version_validity = ensure_correct_version(cg) + whitelisted = table_id in current_app.config["PCG_GRAPH_IDS"] + assert version_validity or whitelisted, f"Invalid graph name {table_id}." return cg From 0188d2fbc10399ef728bc0ab707cbe1bd74623c0 Mon Sep 17 00:00:00 2001 From: Akhilesh Halageri Date: Thu, 1 Dec 2022 18:58:14 +0000 Subject: [PATCH 24/28] fix: valid version takes precedence over whitelist --- pychunkedgraph/app/app_utils.py | 21 ++++++++++++++------- pychunkedgraph/graph/chunkedgraph.py | 4 ++++ 2 files changed, 18 insertions(+), 7 deletions(-) diff --git a/pychunkedgraph/app/app_utils.py b/pychunkedgraph/app/app_utils.py index 778843bba..917e7a60a 100644 --- a/pychunkedgraph/app/app_utils.py +++ b/pychunkedgraph/app/app_utils.py @@ -23,6 +23,7 @@ import requests CACHE = {} +logger: logging.Logger = None def get_app_base_path(): @@ -196,12 +197,14 @@ def setup_logger(table_id: str): def ensure_correct_version(cg: ChunkedGraph) -> bool: - version = cg.client.read_graph_version() current_major_version = int(__version__.split(".")[0]) try: - graph_major_version = int(version.split(".")[0]) - return graph_major_version == current_major_version + graph_major_version = int(cg.version.split(".")[0]) + valid = graph_major_version == current_major_version + assert valid, f"v{cg.version} not supported, server version {__version__}." + return True except (AttributeError, TypeError): + # graph not versioned, later checked if whitelisted return False @@ -218,10 +221,14 @@ def get_cg(table_id, skip_cache: bool = False): if skip_cache is False: CACHE[table_id] = cg - version_validity = ensure_correct_version(cg) - whitelisted = table_id in current_app.config["PCG_GRAPH_IDS"] - assert version_validity or whitelisted, f"Invalid graph name {table_id}." - return cg + version_valid = ensure_correct_version(cg) + if version_valid: + return cg + + if cg.graph_id in current_app.config["PCG_GRAPH_IDS"]: + logger.warning(f"Serving whitelisted graph {cg.graph_id}.") + return cg + raise ValueError(f"Graph {cg.graph_id} not supported.") def get_log_db(table_id): diff --git a/pychunkedgraph/graph/chunkedgraph.py b/pychunkedgraph/graph/chunkedgraph.py index 3c9ec1f27..204469418 100644 --- a/pychunkedgraph/graph/chunkedgraph.py +++ b/pychunkedgraph/graph/chunkedgraph.py @@ -68,6 +68,10 @@ def meta(self) -> ChunkedGraphMeta: def graph_id(self) -> str: return self.meta.graph_config.ID_PREFIX + self.meta.graph_config.ID + @property + def version(self) -> str: + return self.client.read_graph_version() + @property def client(self) -> base.SimpleClient: return self._client From 9c71adde9fff050804894eca03c5790b726871ad Mon Sep 17 00:00:00 2001 From: Akhilesh Halageri Date: Thu, 1 Dec 2022 23:27:40 +0000 Subject: [PATCH 25/28] fix: add to cache after checks pass --- pychunkedgraph/app/app_utils.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/pychunkedgraph/app/app_utils.py b/pychunkedgraph/app/app_utils.py index 917e7a60a..34d9aaff3 100644 --- a/pychunkedgraph/app/app_utils.py +++ b/pychunkedgraph/app/app_utils.py @@ -218,15 +218,14 @@ def get_cg(table_id, skip_cache: bool = False): setup_logger(table_id) cg = ChunkedGraph(graph_id=table_id, client_info=get_default_client_info()) - if skip_cache is False: - CACHE[table_id] = cg - version_valid = ensure_correct_version(cg) if version_valid: + CACHE[table_id] = cg return cg if cg.graph_id in current_app.config["PCG_GRAPH_IDS"]: logger.warning(f"Serving whitelisted graph {cg.graph_id}.") + CACHE[table_id] = cg return cg raise ValueError(f"Graph {cg.graph_id} not supported.") From 115e19841192e3dddcb5d6dcf255872a70a87023 Mon Sep 17 00:00:00 2001 From: Akhilesh Halageri Date: Mon, 5 Dec 2022 18:02:41 +0000 Subject: [PATCH 26/28] fix: retain error message --- pychunkedgraph/graph/chunkedgraph.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/pychunkedgraph/graph/chunkedgraph.py b/pychunkedgraph/graph/chunkedgraph.py index 204469418..2697efeb7 100644 --- a/pychunkedgraph/graph/chunkedgraph.py +++ b/pychunkedgraph/graph/chunkedgraph.py @@ -203,11 +203,11 @@ def get_parents( for id_ in node_ids: try: parents.append(parent_rows[id_][0].value) - except KeyError: + except KeyError as exc: if fail_to_zero: parents.append(0) else: - raise KeyError + raise KeyError from exc parents = np.array(parents, dtype=basetypes.NODE_ID) else: for id_ in node_ids: @@ -215,11 +215,11 @@ def get_parents( parents.append( [(p.value, p.timestamp) for p in parent_rows[id_]] ) - except KeyError: + except KeyError as exc: if fail_to_zero: parents.append([(0, datetime.datetime.fromtimestamp(0))]) else: - raise KeyError + raise KeyError from exc return parents return self.cache.parents_multiple(node_ids, time_stamp=time_stamp) From 544fc71e85e6b146342be6127da7e9c7dd2d0d3e Mon Sep 17 00:00:00 2001 From: Forrest Collman Date: Thu, 29 Dec 2022 16:33:52 -0800 Subject: [PATCH 27/28] fixing whitelist logging --- pychunkedgraph/app/app_utils.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/pychunkedgraph/app/app_utils.py b/pychunkedgraph/app/app_utils.py index 34d9aaff3..2dd9af67a 100644 --- a/pychunkedgraph/app/app_utils.py +++ b/pychunkedgraph/app/app_utils.py @@ -224,6 +224,8 @@ def get_cg(table_id, skip_cache: bool = False): return cg if cg.graph_id in current_app.config["PCG_GRAPH_IDS"]: + instance_id = current_app.config["CHUNKGRAPH_INSTANCE_ID"] + logger = logging.getLogger(f"{instance_id}/{table_id}") logger.warning(f"Serving whitelisted graph {cg.graph_id}.") CACHE[table_id] = cg return cg From bacc7ade9109d1e799e4cd43291b816b3fae7d1b Mon Sep 17 00:00:00 2001 From: Forrest Collman Date: Tue, 3 Jan 2023 14:19:54 -0800 Subject: [PATCH 28/28] fixing merge with skip connections (#432) * fixing merge with skip connections * fixing merge with skip connections * fix: mistaken change * remove hardcoded credentials from tests --- pychunkedgraph/debug/cross_edge_test.py | 2 +- pychunkedgraph/debug/existence_test.py | 2 +- pychunkedgraph/debug/family_test.py | 2 +- pychunkedgraph/graph/edits.py | 14 ++++++++++---- 4 files changed, 13 insertions(+), 7 deletions(-) diff --git a/pychunkedgraph/debug/cross_edge_test.py b/pychunkedgraph/debug/cross_edge_test.py index 9b9b02c6c..25bacfa0b 100644 --- a/pychunkedgraph/debug/cross_edge_test.py +++ b/pychunkedgraph/debug/cross_edge_test.py @@ -5,7 +5,7 @@ from pychunkedgraph.graph import chunkedgraph from pychunkedgraph.graph import attributes -os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "/home/svenmd/.cloudvolume/secrets/google-secret.json" +#os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "/home/svenmd/.cloudvolume/secrets/google-secret.json" layer = 2 n_chunks = 1000 diff --git a/pychunkedgraph/debug/existence_test.py b/pychunkedgraph/debug/existence_test.py index b43e4aa07..757d3d542 100644 --- a/pychunkedgraph/debug/existence_test.py +++ b/pychunkedgraph/debug/existence_test.py @@ -5,7 +5,7 @@ from pychunkedgraph.graph import chunkedgraph from pychunkedgraph.graph import attributes -os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "/home/svenmd/.cloudvolume/secrets/google-secret.json" +#os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "/home/svenmd/.cloudvolume/secrets/google-secret.json" layer = 2 n_chunks = 100 diff --git a/pychunkedgraph/debug/family_test.py b/pychunkedgraph/debug/family_test.py index 1d4d4a223..198351e74 100644 --- a/pychunkedgraph/debug/family_test.py +++ b/pychunkedgraph/debug/family_test.py @@ -5,7 +5,7 @@ from pychunkedgraph.graph import chunkedgraph from pychunkedgraph.graph import attributes -os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "/home/svenmd/.cloudvolume/secrets/google-secret.json" +# os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "/home/svenmd/.cloudvolume/secrets/google-secret.json" layers = [2, 3, 4, 5, 6, 7] n_chunks = 10 diff --git a/pychunkedgraph/graph/edits.py b/pychunkedgraph/graph/edits.py index 1fb5675f6..77d086d94 100644 --- a/pychunkedgraph/graph/edits.py +++ b/pychunkedgraph/graph/edits.py @@ -391,12 +391,15 @@ def _get_old_ids(self, new_ids): ] return np.concatenate(old_ids) - def _map_sv_to_parent(self, node_ids, layer): + def _map_sv_to_parent(self, node_ids, layer, node_map=None): sv_parent_d = {} sv_cross_edges = [types.empty_2d] + if node_map is None: + node_map = {} for id_ in node_ids: + id_eff = node_map.get(id_,id_) edges_ = self._cross_edges_d[id_].get(layer, types.empty_2d) - sv_parent_d.update(dict(zip(edges_[:, 0], [id_] * len(edges_)))) + sv_parent_d.update(dict(zip(edges_[:, 0], [id_eff] * len(edges_)))) sv_cross_edges.append(edges_) return sv_parent_d, np.concatenate(sv_cross_edges) @@ -417,11 +420,14 @@ def _get_connected_components( except TypeError: # NoneType error # if there is a missing parent, try including lower layer ids # this can happen due to skip connections - node_ids = np.concatenate([node_ids, lower_layer_ids]) - sv_parent_d, sv_cross_edges = self._map_sv_to_parent(node_ids, layer) + + lower_layer_to_layer = self.cg.get_roots(lower_layer_ids, stop_layer=layer) + node_map = {k:v for k,v in zip(lower_layer_ids,lower_layer_to_layer)} + sv_parent_d, sv_cross_edges = self._map_sv_to_parent(_node_ids, layer, node_map=node_map) get_sv_parents = np.vectorize(sv_parent_d.get, otypes=[np.uint64]) cross_edges = get_sv_parents(sv_cross_edges) + cross_edges = np.concatenate([cross_edges, np.vstack([node_ids, node_ids]).T]) graph, _, _, graph_ids = flatgraph.build_gt_graph( cross_edges, make_directed=True