[DNM] Move docking scorers

.github/workflows/docking.yaml

@@ -83,11 +83,8 @@ jobs:
     - name: Install packages
       run: |
         python -m pip install -e ./asapdiscovery-data --no-deps
-        python -m pip install -e ./asapdiscovery-dataviz --no-deps
         python -m pip install -e ./asapdiscovery-docking --no-deps
-        python -m pip install -e ./asapdiscovery-ml --no-deps
         python -m pip install -e ./asapdiscovery-modeling --no-deps
-        python -m pip install -e ./asapdiscovery-spectrum --no-deps
         micromamba list
 
     - name: Run tests

asapdiscovery-dataviz/asapdiscovery/dataviz/fint_score.py

@@ -0,0 +1,116 @@
+from pathlib import Path
+from typing import ClassVar, Union
+
+from multimethod import multimethod
+from pydantic.v1 import Field, validator
+
+from asapdiscovery.dataviz.plip import compute_fint_score
+from asapdiscovery.docking.docking import DockingResult
+from asapdiscovery.docking.scorer import (
+    ScorerBase,
+    ScoreType,
+    ScoreUnits,
+    Score,
+    _get_disk_path_from_docking_result,
+)
+from asapdiscovery.spectrum.fitness import target_has_fitness_data
+from asapdiscovery.data.schema.complex import Complex
+from asapdiscovery.data.services.postera.manifold_data_validation import TargetTags
+from asapdiscovery.data.util.dask_utils import dask_vmap, backend_wrapper
+
+
+class FINTScorer(ScorerBase):
+    """
+    Score using Fitness Interaction Score
+
+    Overloaded to accept DockingResults, Complexes, or Paths to PDB files.
+    """
+
+    score_type: ScoreType = Field(ScoreType.FINT, description="Type of score")
+    units: ClassVar[ScoreUnits.arbitrary] = ScoreUnits.arbitrary
+    target: TargetTags = Field(..., description="Which target to use for scoring")
+
+    @validator("target")
+    @classmethod
+    def validate_target(cls, v):
+        if not target_has_fitness_data(v):
+            raise ValueError(
+                "target does not have fitness data so cannot use FINTScorer"
+            )
+        return v
+
+    @dask_vmap(["inputs"])
+    @backend_wrapper("inputs")
+    def _score(
+        self,
+        inputs: Union[list[DockingResult], list[Complex], list[Path]],
+        return_for_disk_backend: bool = False,
+        **kwargs,
+    ) -> list[Score]:
+        """
+        Score the inputs, dispatching based on type.
+        """
+        return self._dispatch(
+            inputs, return_for_disk_backend=return_for_disk_backend, **kwargs
+        )
+
+    @multimethod
+    def _dispatch(
+        self,
+        inputs: list[DockingResult],
+        return_for_disk_backend: bool = False,
+        **kwargs,
+    ) -> list[Score]:
+        """
+        Dispatch for DockingResults
+        """
+        results = []
+        for inp in inputs:
+            _, fint_score = compute_fint_score(
+                inp.to_protein(), inp.posed_ligand.to_oemol(), self.target
+            )
+
+            sc = Score.from_score_and_docking_result(
+                fint_score, self.score_type, self.units, inp
+            )
+            # overwrite the input with the path to the file
+            if return_for_disk_backend:
+                sc.input = _get_disk_path_from_docking_result(inp)
+
+            results.append(sc)
+
+        return results
+
+    @_dispatch.register
+    def _dispatch(self, inputs: list[Complex], **kwargs):
+        """
+        Dispatch for Complexes
+        """
+        results = []
+        for inp in inputs:
+            _, fint_score = compute_fint_score(
+                inp.target.to_oemol(), inp.ligand.to_oemol(), self.target
+            )
+            results.append(
+                Score.from_score_and_complex(
+                    fint_score, self.score_type, self.units, inp
+                )
+            )
+        return results
+
+    @_dispatch.register
+    def _dispatch(self, inputs: list[Path], **kwargs):
+        """
+        Dispatch for PDB files from disk
+        """
+        # assuming reading PDB files from disk
+        complexes = [
+            Complex.from_pdb(
+                p,
+                ligand_kwargs={"compound_name": f"{p.stem}_ligand"},
+                target_kwargs={"target_name": f"{p.stem}_target"},
+            )
+            for p in inputs
+        ]
+
+        return self._dispatch(complexes, **kwargs)

asapdiscovery-dataviz/asapdiscovery/dataviz/tests/test_fintscore.py

@@ -1,9 +1,12 @@
 from pathlib import Path
-
+import pytest
 from asapdiscovery.data.backend.openeye import load_openeye_pdb
+from asapdiscovery.dataviz.fint_score import FINTScorer
 from asapdiscovery.dataviz.plip import compute_fint_score
 from asapdiscovery.data.readers.molfile import MolFileFactory
 from asapdiscovery.data.testing.test_resources import fetch_test_file
+from asapdiscovery.docking.openeye import POSITDockingResults
+from asapdiscovery.data.schema.complex import Complex
 
 
 def test_fint_score():
@@ -32,3 +35,43 @@ def test_fint_score():
     # should both fall between 0 and 1
     assert 0 <= fint_score[0] <= 1.0
     assert 0 <= fint_score[1] <= 1.0
+
+
+@pytest.fixture()
+def results_simple():
+    return [
+        POSITDockingResults.from_json_file(
+            fetch_test_file("docking_results_simple.json")
+        )
+    ]
+
+
+@pytest.fixture()
+def results_simple_nolist(results_simple):
+    return results_simple[0]
+
+
+@pytest.fixture()
+def complex_simple():
+    return Complex.from_pdb(
+        fetch_test_file("Mpro-P0008_0A_ERI-UCB-ce40166b-17_prepped_receptor_0.pdb"),
+        ligand_kwargs={"compound_name": "test"},
+        target_kwargs={"target_name": "test", "target_hash": "mock_hash"},
+    )
+
+
+@pytest.fixture()
+def pdb_simple():
+    return fetch_test_file("Mpro-P0008_0A_ERI-UCB-ce40166b-17_prepped_receptor_0.pdb")
+
+
+@pytest.mark.parametrize(
+    "data_fixture", ["results_simple_nolist", "complex_simple", "pdb_simple"]
+)
+@pytest.mark.parametrize("return_df", [True, False])
+@pytest.mark.parametrize("use_dask", [True, False])
+def test_FINT_scorer(use_dask, return_df, data_fixture, request):
+    data = request.getfixturevalue(data_fixture)
+    scorer = FINTScorer(target="SARS-CoV-2-Mpro")
+    scores = scorer.score([data], use_dask=use_dask, return_df=return_df)
+    assert len(scores) == 1