fixes and speedups

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/app_utils.py

@@ -2,27 +2,28 @@
 import sys
 import os
 from typing import Sequence
+from time import gmtime
 
 import numpy as np
 from flask import current_app, json, request
 from google.auth import credentials
 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
 
 CACHE = {}
+logger: logging.Logger = None
 
 
 def get_app_base_path():
@@ -176,28 +177,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,14 +193,43 @@ def get_cg(table_id, skip_cache: bool = False):
     )
     formatter.converter = gmtime
     handler.setFormatter(formatter)
-
     logger.addHandler(handler)
 
-    # Create ChunkedGraph
-    cg = ChunkedGraph(graph_id=table_id, client_info=get_default_client_info())
+
+def ensure_correct_version(cg: ChunkedGraph) -> bool:
+    current_major_version = int(__version__.split(".")[0])
+    try:
+        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
+
+
+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())
+    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"]:
+        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
+        return cg
+    raise ValueError(f"Graph {cg.graph_id} not supported.")
 
 
 def get_log_db(table_id):

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/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/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

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)

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