meshing v3

Dockerfile

@@ -3,5 +3,5 @@ FROM seunglab/pychunkedgraph:graph-tool_dracopy
 COPY override/timeout.conf /etc/nginx/conf.d/timeout.conf
 COPY override/supervisord.conf /etc/supervisor/conf.d/supervisord.conf
 COPY requirements.txt /app
-RUN pip install pip==20.2 && pip install --no-cache-dir --upgrade -r requirements.txt
+RUN pip install pip==20.2 pyfqmr && pip install --no-cache-dir --upgrade -r requirements.txt
 COPY . /app

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"

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/<table_id>/manifest/<node_id>: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)

pychunkedgraph/app/segmentation/common.py

@@ -278,7 +278,7 @@ def handle_roots(table_id, is_binary=False):
     stop_layer = int(request.args.get("stop_layer", cg.meta.layer_count))
     is_root_layer = stop_layer == cg.meta.layer_count
     assert_roots = bool(request.args.get("assert_roots", False))
-    fail_to_zero = bool(request.args.get("fail_to_zero", False))
+    fail_to_zero = bool(request.args.get("fail_to_zero", True))
     root_ids = cg.get_roots(
         node_ids,
         stop_layer=stop_layer,

pychunkedgraph/app/segmentation/generic/routes.py

@@ -1,11 +1,19 @@
-from flask import Blueprint
+from flask import Blueprint, current_app
 from middle_auth_client import (
     auth_requires_admin,
     auth_required,
     auth_requires_permission,
 )
 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/<table_id>/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)

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/<table_id>/node/<node_id>/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/<table_id>/node/leaves_many", methods=["POST"])
 @bp.route("/table/<table_id>/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/<table_id>/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/<table_id>/node/<node_id>/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)

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)