Version 3 with cached cross chunk edges

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 2.21.1
+current_version = 3.1.0
 commit = True
 tag = True
 

.github/workflows/main.yml

@@ -4,9 +4,11 @@ on:
   push:
     branches:
       - "main"
+      - "pcgv3"
   pull_request:
     branches:
       - "main"
+      - "pcgv3"
 
 jobs:
   unit-tests:

pychunkedgraph/__init__.py

@@ -1 +1 @@
-__version__ = "2.21.1"
+__version__ = "3.1.0"

pychunkedgraph/app/__init__.py

@@ -105,6 +105,8 @@ def configure_app(app):
         with app.app_context():
             from ..ingest.rq_cli import init_rq_cmds
             from ..ingest.cli import init_ingest_cmds
+            from ..ingest.cli_upgrade import init_upgrade_cmds
 
             init_rq_cmds(app)
             init_ingest_cmds(app)
+            init_upgrade_cmds(app)

pychunkedgraph/app/meshing/common.py

@@ -4,8 +4,6 @@
 import threading
 
 import numpy as np
-import redis
-from rq import Queue, Connection, Retry
 from flask import Response, current_app, jsonify, make_response, request
 
 from pychunkedgraph import __version__
@@ -145,37 +143,15 @@ def _check_post_options(cg, resp, data, seg_ids):
 def handle_remesh(table_id):
     current_app.request_type = "remesh_enque"
     current_app.table_id = table_id
-    is_priority = request.args.get("priority", True, type=str2bool)
-    is_redisjob = request.args.get("use_redis", False, type=str2bool)
-
     new_lvl2_ids = json.loads(request.data)["new_lvl2_ids"]
-
-    if is_redisjob:
-        with Connection(redis.from_url(current_app.config["REDIS_URL"])):
-
-            if is_priority:
-                retry = Retry(max=3, interval=[1, 10, 60])
-                queue_name = "mesh-chunks"
-            else:
-                retry = Retry(max=3, interval=[60, 60, 60])
-                queue_name = "mesh-chunks-low-priority"
-            q = Queue(queue_name, retry=retry, default_timeout=1200)
-            task = q.enqueue(meshing_tasks.remeshing, table_id, new_lvl2_ids)
-
-        response_object = {"status": "success", "data": {"task_id": task.get_id()}}
-
-        return jsonify(response_object), 202
-    else:
-        new_lvl2_ids = np.array(new_lvl2_ids, dtype=np.uint64)
-        cg = app_utils.get_cg(table_id)
-
-        if len(new_lvl2_ids) > 0:
-            t = threading.Thread(
-                target=_remeshing, args=(cg.get_serialized_info(), new_lvl2_ids)
-            )
-            t.start()
-
-        return Response(status=202)
+    new_lvl2_ids = np.array(new_lvl2_ids, dtype=np.uint64)
+    cg = app_utils.get_cg(table_id)
+    if len(new_lvl2_ids) > 0:
+        t = threading.Thread(
+            target=_remeshing, args=(cg.get_serialized_info(), new_lvl2_ids)
+        )
+        t.start()
+    return Response(status=202)
 
 
 def _remeshing(serialized_cg_info, lvl2_nodes):

pychunkedgraph/debug/utils.py

@@ -1,7 +1,8 @@
+# pylint: disable=invalid-name, missing-docstring, bare-except, unidiomatic-typecheck
+
 import numpy as np
 
-from ..graph import ChunkedGraph
-from ..graph.utils.basetypes import NODE_ID
+from pychunkedgraph.graph.meta import ChunkedGraphMeta, GraphConfig
 
 
 def print_attrs(d):
@@ -16,28 +17,55 @@ def print_attrs(d):
             print(v)
 
 
-def print_node(
-    cg: ChunkedGraph,
-    node: NODE_ID,
-    indent: int = 0,
-    stop_layer: int = 2,
-) -> None:
+def print_node(cg, node: np.uint64, 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)
+        print_node(cg, child, indent=indent + 4, stop_layer=stop_layer)
 
 
-def get_l2children(cg: ChunkedGraph, node: NODE_ID) -> np.ndarray:
-    nodes = np.array([node], dtype=NODE_ID)
+def get_l2children(cg, node: np.uint64) -> np.ndarray:
+    nodes = np.array([node], dtype=np.uint64)
     layers = cg.get_chunk_layers(nodes)
-    assert np.all(layers > 2), "nodes must be at layers > 2"
+    assert np.all(layers >= 2), "nodes must be at layers >= 2"
     l2children = []
     while nodes.size:
         children = cg.get_children(nodes, flatten=True)
         layers = cg.get_chunk_layers(children)
         l2children.append(children[layers == 2])
         nodes = children[layers > 2]
     return np.concatenate(l2children)
+
+
+def sanity_check(cg, new_roots, operation_id):
+    """
+    Check for duplicates in hierarchy, useful for debugging.
+    """
+    # print(f"{len(new_roots)} new ids from {operation_id}")
+    l2c_d = {}
+    for new_root in new_roots:
+        l2c_d[new_root] = get_l2children(cg, new_root)
+    success = True
+    for k, v in l2c_d.items():
+        success = success and (len(v) == np.unique(v).size)
+        # print(f"{k}: {np.unique(v).size}, {len(v)}")
+    if not success:
+        raise RuntimeError("Some ids are not valid.")
+
+
+def sanity_check_single(cg, node, operation_id):
+    v = get_l2children(cg, node)
+    msg = f"invalid node {node}:"
+    msg += f" found {len(v)} l2 ids, must be {np.unique(v).size}"
+    assert np.unique(v).size == len(v), f"{msg}, from {operation_id}."
+    return v
+
+
+def update_graph_id(cg, new_graph_id:str):
+    old_gc = cg.meta.graph_config._asdict()
+    old_gc["ID"] = new_graph_id
+    new_gc = GraphConfig(**old_gc)
+    new_meta = ChunkedGraphMeta(new_gc, cg.meta.data_source, cg.meta.custom_data)
+    cg.update_meta(new_meta, overwrite=True)