Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion chemap/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,8 +3,8 @@


__all__ = [
"DatasetLoader",
"FingerprintConfig",
"compute_fingerprints",
"DatasetLoader",
"mol_from_smiles",
]
20 changes: 16 additions & 4 deletions chemap/approx_nn.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
"""

import time
from typing import Any, List, Tuple
from typing import Any
import numba
import numpy as np
from fingerprint_computation import compute_fingerprints_from_smiles
Expand All @@ -30,8 +30,8 @@


def compound_nearest_neighbors(
smiles: List[str], k_pca: int = 500, k_morgan: int = 100
) -> Tuple[Any, Any]:
smiles: list[str], k_pca: int = 500, k_morgan: int = 100
) -> tuple[Any, Any]:
"""
Compute approximate nearest neighbors for a list of SMILES strings.

Expand Down Expand Up @@ -78,8 +78,20 @@ def compound_nearest_neighbors(

def compute_approx_nearest_neighbors(
fingerprints_coarse, fingerprints_fine, k_pca: int = 500, k_morgan: int = 100
) -> Tuple[Any, Any]:
) -> tuple[Any, Any]:
"""Compute approximate nearest neighbors using PCA and Ruzicka similarity.

Parameters
----------
fingerprints_coarse:
Dense fingerprints (e.g., 1024-bit) used for PCA-based dimensionality reduction.
fingerprints_fine:
Sparse fingerprints (e.g., 4096-bit) used for refined Ruzicka-based neighbor search.
k_pca:
Number of neighbors to consider in the PCA-based approximate nearest neighbor search.
k_morgan:
Number of neighbors to consider in the refined Ruzicka-based neighbor search.
"""
t_start = time.time()
print(">" * 20, "Compute PCA vectors")
pca = PCA(n_components=100)
Expand Down
39 changes: 20 additions & 19 deletions chemap/benchmarking/fingerprint_duplicates.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,8 @@
import json
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, List, Mapping, Optional, Sequence, Tuple, Union
from typing import Any
import numpy as np


Expand All @@ -14,14 +15,14 @@ class DuplicatesNPZ:
"""CSR-like encoding of duplicate groups."""
indices: np.ndarray # shape (nnz,), int
indptr: np.ndarray # shape (n_groups+1,), int
n_items: Optional[int] = None # optional size of the original universe
n_items: int | None = None # optional size of the original universe


def encode_duplicates(
duplicates: Sequence[Sequence[int]],
*,
dtype: np.dtype = np.int32,
n_items: Optional[int] = None,
n_items: int | None = None,
) -> DuplicatesNPZ:
"""Encode list-of-lists duplicate groups into CSR-like arrays."""
# Build indptr
Expand Down Expand Up @@ -51,7 +52,7 @@ def encode_duplicates(
)


def decode_duplicates(encoded: DuplicatesNPZ) -> List[List[int]]:
def decode_duplicates(encoded: DuplicatesNPZ) -> list[list[int]]:
"""Decode CSR-like arrays back into list-of-lists."""
indices = np.asarray(encoded.indices)
indptr = np.asarray(encoded.indptr)
Expand All @@ -65,7 +66,7 @@ def decode_duplicates(encoded: DuplicatesNPZ) -> List[List[int]]:
if np.any(indptr[1:] < indptr[:-1]):
raise ValueError("indptr must be non-decreasing")

out: List[List[int]] = []
out: list[list[int]] = []
for i in range(indptr.size - 1):
start = int(indptr[i])
end = int(indptr[i + 1])
Expand All @@ -78,12 +79,12 @@ def decode_duplicates(encoded: DuplicatesNPZ) -> List[List[int]]:
# ---------------------------------------------------------------------------

def save_duplicates_npz(
filepath: Union[str, Path],
filepath: str | Path,
duplicates: Sequence[Sequence[int]],
*,
n_items: Optional[int] = None,
n_items: int | None = None,
dtype: np.dtype = np.int32,
metadata: Optional[Mapping[str, Any]] = None,
metadata: Mapping[str, Any] | None = None,
metadata_suffix: str = ".json",
) -> Path:
"""Save duplicates to a compressed NPZ file (+ optional JSON metadata sidecar)."""
Expand All @@ -107,26 +108,26 @@ def save_duplicates_npz(


def load_duplicates_npz(
filepath: Union[str, Path],
filepath: str | Path,
*,
load_metadata: bool = False,
metadata_suffix: str = ".json",
) -> Tuple[List[List[int]], Optional[Dict[str, Any]]]:
) -> tuple[list[list[int]], dict[str, Any] | None]:
"""Load duplicates from NPZ (and optional JSON metadata if present)."""
path = Path(filepath)
with np.load(path, allow_pickle=False) as z:
indices = z["indices"]
indptr = z["indptr"]
n_items_arr = z.get("n_items", None)

n_items: Optional[int] = None
n_items: int | None = None
if n_items_arr is not None:
v = int(np.asarray(n_items_arr).reshape(-1)[0])
n_items = None if v < 0 else v

duplicates = decode_duplicates(DuplicatesNPZ(indices=indices, indptr=indptr, n_items=n_items))

meta: Optional[Dict[str, Any]] = None
meta: dict[str, Any] | None = None
if load_metadata:
meta_path = path.with_suffix(path.suffix + metadata_suffix)
if meta_path.exists():
Expand All @@ -145,17 +146,17 @@ class PrecomputedDuplicates:
"""One precomputed experiment entry loaded from disk."""
name: str
path_npz: Path
duplicates: List[List[int]]
metadata: Optional[Dict[str, Any]] = None
duplicates: list[list[int]]
metadata: dict[str, Any] | None = None


def load_precomputed_duplicates_folder(
folder: Union[str, Path],
folder: str | Path,
*,
pattern: str = "*_duplicates.npz",
name_from_filename: Optional[Any] = None,
name_from_filename: Any | None = None,
load_metadata: bool = False,
) -> List[PrecomputedDuplicates]:
) -> list[PrecomputedDuplicates]:
"""Load all precomputed duplicate results from a folder.

Parameters
Expand All @@ -178,11 +179,11 @@ def load_precomputed_duplicates_folder(
raise FileNotFoundError(f"Folder does not exist: {folder_path}")

files = sorted(folder_path.glob(pattern))
out: List[PrecomputedDuplicates] = []
out: list[PrecomputedDuplicates] = []

def default_name(p: Path) -> str:
stem = p.stem # for "x_duplicates.npz" => "x_duplicates"
return stem[:-11] if stem.endswith("_duplicates") else stem
return stem.removesuffix("_duplicates")

name_fn = name_from_filename or default_name

Expand Down
6 changes: 4 additions & 2 deletions chemap/benchmarking/utils.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,3 @@
from typing import List
import numpy as np


Expand All @@ -11,7 +10,7 @@ def compute_duplicate_max_mass_differences(
"""
Compute all maximum mass differences between duplicates.
"""
max_diffs: List[float] = []
max_diffs: list[float] = []
for group in duplicates:
idx = np.asarray(group, dtype=int)
group_masses = masses[idx]
Expand All @@ -22,6 +21,9 @@ def compute_duplicate_max_mass_differences(


def compute_compound_max_mass_differences(masses):
"""
Compute the maximum mass difference for each compound in a group of masses.
"""
all_max_diffs = []
min_mass = masses.min()
max_mass = masses.max()
Expand Down
5 changes: 3 additions & 2 deletions chemap/data_loader.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@


class DatasetLoader:
"""Class to load datasets from local files or web sources."""
def __init__(self, cache_dir="./data_cache"):
self.cache_dir = cache_dir

Expand Down Expand Up @@ -51,10 +52,10 @@ def load_collection(self, source: str, **kwargs) -> list:
-------------
ValueError if DOI not present.
"""
doi_pattern = r'(10\.\d{4,9}/[-._;()/:a-zA-Z0-9]+)'
doi_pattern = r"(10\.\d{4,9}/[-._;()/:a-zA-Z0-9]+)"

if not source.startswith("doi") or not bool(re.search(doi_pattern, source)):
ValueError(f"Could not detect DOI in source {source}.")
raise ValueError(f"Could not detect DOI in source {source}.")

return self._from_registry(source, **kwargs)

Expand Down
Loading
Loading