Feat/pinder interface ingest

flows/configs/v3/make_components.yaml

@@ -1,5 +1,5 @@
 flow:
-  run_specific_stages: collate_partitions,make_component_reductions,merge_component_reductions,make_communities,make_directed_set_covers,summarize_clusters,finalize_index
+  run_specific_stages: collate_partitions,plan_clusters,make_symmetric_edge_fragments,make_symmetric_edge_shards,make_component_reductions,merge_component_reductions,make_communities,make_directed_set_covers,summarize_clusters,finalize_index
   make_components_force_update: true
 data:
   plinder_mount: /plinder

flows/configs/v3/make_protein_scores.yaml

@@ -1,8 +1,11 @@
 flow:
-  run_specific_stages: make_dbs,make_sub_dbs,run_batch_searches,map_batch_alignments,collate_alignments,finalize_alignments,make_batch_scores,collate_ligand_3d_candidates,plan_ligand_3d_scores,make_ligand_3d_scores,collate_ligand_3d_scores,merge_ligand_3d_scores,finalize_scores,export_sucos_shape_pocket_qcov,finalize_sucos_export
+  run_specific_stages: make_dbs,make_sub_dbs,run_batch_searches,map_batch_alignments,collate_alignments,finalize_alignments,plan_interface_scores,make_interface_scores,finalize_interface_scores,make_batch_scores,collate_ligand_3d_candidates,plan_ligand_3d_scores,make_ligand_3d_scores,collate_ligand_3d_scores,merge_ligand_3d_scores,finalize_scores,export_sucos_shape_pocket_qcov,finalize_sucos_export
   make_sub_dbs_cpu: 16
   run_batch_searches_batch_size: 5000
   map_batch_alignments_batch_size: 1
+  make_interface_scores_batch_size: 1
+  make_interface_scores_cpu: 4
+  make_interface_scores_memory_limit: 32GB
   make_batch_scores_batch_size: 50
   make_batch_scores_cpu: 4
   make_ligand_3d_scores_batch_size: 30000

flows/data_ingest.py

@@ -66,6 +66,7 @@ def start(self):
         )
         self.pipeline = IngestPipeline(conf=get_config(config_contents=contents))
         self.next(self.scatter_make_entries)
+
     @kubernetes(**{**K8S, **DATABASES})
     @environment(**ENV)
     @retry
@@ -292,6 +293,47 @@ def join_collate_alignments(self, inputs):
     @step
     def finalize_alignments(self):
         self.pipeline.finalize_alignments()
+        self.next(self.plan_interface_scores)
+
+    @kubernetes(**{**K8S, **WORKSTATION_MEM})
+    @environment(**ENV)
+    @retry
+    @step
+    def plan_interface_scores(self):
+        self.pipeline.plan_interface_scores()
+        self.next(self.scatter_make_interface_scores)
+
+    @kubernetes(**K8S)
+    @environment(**ENV)
+    @retry
+    @step
+    def scatter_make_interface_scores(self):
+        self.chunks = self.pipeline.scatter_make_interface_scores()
+        self.next(self.make_interface_scores, foreach="chunks")
+
+    @kubernetes(**{**K8S, **{"cpu": 4, "memory": 32000}})
+    @environment(**ENV)
+    @retry
+    @step
+    def make_interface_scores(self):
+        self.pipeline.make_interface_scores(self.input)
+        self.next(self.join_make_interface_scores)
+
+    @kubernetes(**K8S)
+    @environment(**ENV)
+    @retry
+    @step
+    def join_make_interface_scores(self, inputs):
+        self.pipeline = inputs[0].pipeline
+        self.merge_artifacts(inputs, exclude=["chunks"])
+        self.next(self.finalize_interface_scores)
+
+    @kubernetes(**{**K8S, **WORKSTATION_MEM})
+    @environment(**ENV)
+    @retry
+    @step
+    def finalize_interface_scores(self):
+        self.pipeline.finalize_interface_scores()
         self.next(self.scatter_make_batch_scores)
 
     @kubernetes(**K8S)
@@ -489,6 +531,64 @@ def collate_partitions(self):
     @retry
     @step
     def join_collate_partitions(self, inputs):
+        self.pipeline = inputs[0].pipeline
+        self.merge_artifacts(inputs, exclude=["chunks"])
+        self.next(self.plan_clusters)
+
+    @kubernetes(**K8S)
+    @environment(**ENV)
+    @retry
+    @step
+    def plan_clusters(self):
+        self.pipeline.plan_clusters()
+        self.next(self.scatter_make_symmetric_edge_fragments)
+
+    @kubernetes(**K8S)
+    @environment(**ENV)
+    @retry
+    @step
+    def scatter_make_symmetric_edge_fragments(self):
+        self.chunks = self.pipeline.scatter_make_symmetric_edge_fragments()
+        self.next(self.make_symmetric_edge_fragments, foreach="chunks")
+
+    @kubernetes(**{**K8S, **LARGE_MEM})
+    @environment(**ENV)
+    @retry
+    @step
+    def make_symmetric_edge_fragments(self):
+        self.pipeline.make_symmetric_edge_fragments(self.input)
+        self.next(self.join_make_symmetric_edge_fragments)
+
+    @kubernetes(**K8S)
+    @environment(**ENV)
+    @retry
+    @step
+    def join_make_symmetric_edge_fragments(self, inputs):
+        self.pipeline = inputs[0].pipeline
+        self.merge_artifacts(inputs, exclude=["chunks"])
+        self.next(self.scatter_make_symmetric_edge_shards)
+
+    @kubernetes(**K8S)
+    @environment(**ENV)
+    @retry
+    @step
+    def scatter_make_symmetric_edge_shards(self):
+        self.chunks = self.pipeline.scatter_make_symmetric_edge_shards()
+        self.next(self.make_symmetric_edge_shards, foreach="chunks")
+
+    @kubernetes(**{**K8S, **LARGE_MEM})
+    @environment(**ENV)
+    @retry
+    @step
+    def make_symmetric_edge_shards(self):
+        self.pipeline.make_symmetric_edge_shards(self.input)
+        self.next(self.join_make_symmetric_edge_shards)
+
+    @kubernetes(**K8S)
+    @environment(**ENV)
+    @retry
+    @step
+    def join_make_symmetric_edge_shards(self, inputs):
         self.pipeline = inputs[0].pipeline
         self.merge_artifacts(inputs, exclude=["chunks"])
         self.next(self.scatter_make_component_reductions)

pyproject.toml

@@ -111,6 +111,8 @@ plinder = [
     "**.yaml",
     "**.tsv",
     "plinder/data/annotations/static_files/artifacts_badlist.csv",
+    "data/annotations/static_files/prodigy_classifier.npz",
+    "data/annotations/static_files/PRODIGY_CRYST_NOTICE.txt",
 ]
 
 [tool.ruff]

scripts/slurm/README.md

@@ -85,19 +85,38 @@ whole-process resource use is written once per array task to
 
 ## Collate the V3 annotation index
 
-After entry ingest is complete, inventory the exact per-entry outputs. The plan
-fails if any materialized entry is missing an annotation, chain, biological-
-assembly-chain, source, or ligand Parquet:
+After entry ingest is complete, freeze the two-character code list, inventory
+the exact per-entry outputs in parallel, and atomically merge the inventories.
+The plan fails if any materialized entry is missing an annotation, chain,
+biological-assembly-chain, source, or ligand Parquet:
 
 ```bash
 sbatch \
-  --qos=6hours \
-  --cpus-per-task=8 --mem=16G \
-  --output="${OUTPUT_ROOT}/logs/collate-plan-%j.out" \
+  --output="${OUTPUT_ROOT}/logs/collate-plan-start-%j.out" \
+  --export=ALL,PLINDER_ENV_ROOT,PLINDER_REPO_ROOT \
+  scripts/slurm/collate_v3_shards.sbatch plan-start "${OUTPUT_ROOT}"
+
+# Read code_count from index/.staging/v3_collation/plan-build.json. With a
+# batch size of four, LAST_PLAN_BATCH_INDEX is ceil(code_count / 4) - 1.
+sbatch \
+  --array=0-LAST_PLAN_BATCH_INDEX --cpus-per-task=4 --mem=8G \
+  --output="${OUTPUT_ROOT}/logs/collate-plan-%A-%a.out" \
+  --export=ALL,PLINDER_ENV_ROOT,PLINDER_REPO_ROOT \
+  scripts/slurm/collate_v3_shards.sbatch plan-shard "${OUTPUT_ROOT}" 4
+
+sbatch \
+  --dependency=afterok:PLAN_ARRAY_JOB_ID \
+  --output="${OUTPUT_ROOT}/logs/collate-plan-finish-%j.out" \
   --export=ALL,PLINDER_ENV_ROOT,PLINDER_REPO_ROOT \
-  scripts/slurm/collate_v3_shards.sbatch plan "${OUTPUT_ROOT}"
+  scripts/slurm/collate_v3_shards.sbatch plan-finish "${OUTPUT_ROOT}"
 ```
 
+When augmenting an existing ligand release with interface-only ingest, set
+`PLINDER_COLLATE_INTERFACES_ONLY=true` on `plan-start`. This preserves the
+installed `index/annotation_table.parquet` byte-for-byte and collates only the
+shared chain, entry, source, and interface tables. The setting is frozen in the
+plan, so later stages do not need the environment variable.
+
 Read `index/.staging/v3_collation/plan.json` after that job succeeds. With a
 batch size of four, set `LAST_CODE_BATCH_INDEX` to
 `ceil(code_count / 4) - 1`, then submit the unthrottled shard array:
@@ -122,15 +141,18 @@ sbatch \
   scripts/slurm/collate_v3_shards.sbatch finalize "${OUTPUT_ROOT}"
 ```
 
-Finalization writes the four local `index/*.parquet` files only after validating
-row counts, keys, cross-table references, and ligand scoreability. It performs
-no upload or external release operation.
+Finalization writes the local `index/*.parquet` files only after validating row
+counts, keys, cross-table references, and ligand scoreability. Each shard
+verifies its raw inputs before producing a frozen output; finalization validates
+and merges those outputs without rescanning every per-entry source file. It
+performs no upload or external release operation.
 
 ## Build publishable protein-search shards
 
-Protein scoring uses only holo receptor chains labeled `protein` in
-`index/entry_chains.parquet`; other protein chains retained for biological-unit
-context are not searched. Freeze that query universe before starting any arrays:
+Protein scoring uses the union of protein chains that are holo ligand receptors
+or members of `index/interface_annotation_table.parquet`; unrelated chains
+retained only for biological-unit context are not searched. Freeze that query
+universe before starting any arrays:
 
 ```bash
 python -m plinder.data.pipeline.score plan "${OUTPUT_ROOT}" --max-seqs 10000
@@ -246,6 +268,41 @@ sbatch \
   scripts/slurm/score_v3.sbatch finalize-alignments "${OUTPUT_ROOT}"
 ```
 
+The same mapped protein alignments also drive protein-interface scoring. Plan
+one two-character query shard per array task, run the shards independently,
+and then publish the compact all-vs-all table:
+
+```bash
+sbatch \
+  --qos=30min --cpus-per-task=1 --mem=8G \
+  --output="${OUTPUT_ROOT}/logs/interface-score-plan-%j.out" \
+  --export=ALL,PLINDER_ENV_ROOT,PLINDER_REPO_ROOT \
+  scripts/slurm/score_v3.sbatch plan-interface-scores "${OUTPUT_ROOT}" 1
+
+# Read batch_count from manifests/interface_scoring_plan.json, then:
+sbatch \
+  --qos=30min --array=0-LAST_INTERFACE_INDEX --cpus-per-task=4 --mem=32G \
+  --output="${OUTPUT_ROOT}/logs/interface-score-%A-%a.out" \
+  --export=ALL,PLINDER_ENV_ROOT,PLINDER_REPO_ROOT \
+  scripts/slurm/score_v3.sbatch score-interface-shards "${OUTPUT_ROOT}" 1
+
+sbatch \
+  --qos=6hours --cpus-per-task=8 --mem=64G \
+  --output="${OUTPUT_ROOT}/logs/interface-finalize-%j.out" \
+  --export=ALL,PLINDER_ENV_ROOT,PLINDER_REPO_ROOT,PLINDER_DUCKDB_MEMORY_LIMIT=56GB \
+  scripts/slurm/score_v3.sbatch finalize-interface-scores "${OUTPUT_ROOT}"
+```
+
+Each winning direct or swapped chain assignment retains `iface1_qcov` and
+`iface2_qcov` before multiplication. They are the coverages of the query
+interface's canonical first and second chain, respectively, under that winning
+assignment. The compact release file
+`exports/all_interface_qcov.parquet` contains the query and target interface
+IDs, both side coverages, and the final directional 0--100 similarity. Positive
+scores below the lowest clustering threshold are retained. Interface clustering
+uses the whole-interface `interface_qcov` product; the two side coverages remain
+available as score diagnostics but are not clustered separately.
+
 Derived per-ligand system scores are generation intermediates used for graph
 clustering, not release artifacts. Before scattering them, estimate work from
 the compact alignment hits and greedily spread expensive queries across fixed
@@ -509,8 +566,27 @@ sbatch \
   scripts/slurm/score_v3.sbatch plan-clusters "${OUTPUT_ROOT}" 1
 ```
 
+Protein-interface clustering uses the same commands and thresholds in a
+separate namespace. Set these variables for every command in the sequence:
+
+```bash
+export PLINDER_CLUSTER_ENTITY_TYPE=interface
+export PLINDER_CLUSTER_METRICS='interface_qcov'
+export PLINDER_CLUSTER_THRESHOLDS='100,90,70,50,30'
+```
+
+The interface plan reads `interface_scores/shard=*.parquet`. Reciprocal
+components and communities are written below `interface_clusters/`; the
+directional centroid cover is written below `interface_sampling/`. Final index
+enrichment adds `interface_qcov__THRESHOLD__component`, `__community`, and
+`__directed_set_cover` columns for whole-interface product scores.
+The ligand and interface plans and artifacts never share cache paths.
+Run both entity sequences before `finalize-index`; finalization rejects a
+non-empty interface annotation table when its interface clusters are absent.
+
 After component, community, and directed-cover branches complete, validate every
-published artifact and write `ligand_clusters/stats.parquet` plus `stats.json`.
+published artifact and write `ligand_clusters/stats.parquet` (or
+`interface_clusters/stats.parquet`) plus `stats.json`.
 This gate checks artifact coverage, duplicate and null labels, consistent ligand
 counts, and monotonic component counts across thresholds. Community and cover
 counts are reported but are not required to be monotonic:

scripts/slurm/collate_v3_shards.sbatch

@@ -10,12 +10,12 @@ umask 002
 
 MODE=${1:-shard}
 case "${MODE}" in
-    plan|finalize|repair)
+    plan|plan-start|plan-finish|finalize|repair)
         DATA_DIR=${2:?'usage: sbatch collate_v3_shards.sbatch MODE DATA_DIR'}
         BATCH_SIZE=0
         ;;
-    shard)
-        DATA_DIR=${2:?'usage: sbatch collate_v3_shards.sbatch shard DATA_DIR BATCH_SIZE'}
+    plan-shard|shard)
+        DATA_DIR=${2:?'usage: sbatch collate_v3_shards.sbatch MODE DATA_DIR BATCH_SIZE'}
         BATCH_SIZE=${3:?'usage: sbatch collate_v3_shards.sbatch shard DATA_DIR BATCH_SIZE'}
         ;;
     *)
@@ -52,6 +52,23 @@ if [[ "${MODE}" == "plan" ]]; then
     /usr/bin/time -v -o "${TIME_OUTPUT}" \
         python -m plinder.data.pipeline.collate plan "${DATA_DIR}" \
             --threads "${SLURM_CPUS_PER_TASK:-1}"
+elif [[ "${MODE}" == "plan-start" ]]; then
+    PLAN_ARGS=()
+    if [[ "${PLINDER_COLLATE_INTERFACES_ONLY:-false}" == "true" ]]; then
+        PLAN_ARGS+=(--interfaces-only)
+    fi
+    /usr/bin/time -v -o "${TIME_OUTPUT}" \
+        python -m plinder.data.pipeline.collate plan-start "${DATA_DIR}" \
+            "${PLAN_ARGS[@]}"
+elif [[ "${MODE}" == "plan-shard" ]]; then
+    /usr/bin/time -v -o "${TIME_OUTPUT}" \
+        python -m plinder.data.pipeline.collate plan-shard "${DATA_DIR}" \
+            --batch-index "${BATCH_INDEX}" \
+            --batch-size "${BATCH_SIZE}" \
+            --threads "${SLURM_CPUS_PER_TASK:-1}"
+elif [[ "${MODE}" == "plan-finish" ]]; then
+    /usr/bin/time -v -o "${TIME_OUTPUT}" \
+        python -m plinder.data.pipeline.collate plan-finish "${DATA_DIR}"
 elif [[ "${MODE}" == "finalize" ]]; then
     /usr/bin/time -v -o "${TIME_OUTPUT}" \
         python -m plinder.data.pipeline.collate finalize "${DATA_DIR}" \

scripts/slurm/ingest_pdb_batch.sbatch

@@ -43,7 +43,15 @@ ARGS=(
 if [[ "${FORCE:-0}" == "1" ]]; then
     ARGS+=(--force)
 fi
+INGEST_MODE=${INGEST_MODE:-all}
+ARGS+=(--mode "${INGEST_MODE}")
+case "${INGEST_MODE}" in
+    all) METRICS_PREFIX=ingest-batch ;;
+    ligands) METRICS_PREFIX=ligand-ingest-batch ;;
+    interfaces) METRICS_PREFIX=interface-ingest-batch ;;
+    *) echo "INGEST_MODE must be all, ligands, or interfaces" >&2; exit 2 ;;
+esac
 
 /usr/bin/time -v \
-    -o "${OUTPUT_ROOT}/metrics/ingest-batch-${JOB_ID}-${BATCH_INDEX}.time-v.txt" \
+    -o "${OUTPUT_ROOT}/metrics/${METRICS_PREFIX}-${JOB_ID}-${BATCH_INDEX}.time-v.txt" \
     python -m plinder.data.pipeline.ingest batch "${ARGS[@]}"