plinder.data.annotations.get_similarity_scores#

plinder.data.annotations.get_similarity_scores.score_thresholds_metadata(minimum_threshold: float, minimum_thresholds: dict[str, float]) → bytes[source]#

Encode score-filter thresholds for Parquet completion metadata.

plinder.data.annotations.get_similarity_scores.score_metrics_metadata(metrics: Collection[str] | None) → bytes | None[source]#

Encode an optional retained-metric set for cache validation.

plinder.data.annotations.get_similarity_scores.score_cache_is_current(path: Path, *, ligand_3d_mode: bytes, minimum_threshold: float, minimum_thresholds: dict[str, float], holo_protein_scores_mode: bytes | None = None, score_metrics: Collection[str] | None = None) → bool[source]#

Return whether a per-query score file matches the active filters.

plinder.data.annotations.get_similarity_scores.write_ecfp4_fingerprint_table(ligands: DataFrame, output_path: Path) → None[source]#

Write the fixed ECFP4/1024 fingerprint table with explicit metadata.

plinder.data.annotations.get_similarity_scores.write_mhfp6_fingerprints(unique_ligands: DataFrame, output_dir: Path) → None[source]#

Write the MHFP6/2048 MinHash table over the shared ligand_smiles_id set.

unique_ligands is the already-deduplicated, ligand_smiles_id-indexed frame produced for ECFP4, so the two chemical metrics stay node-aligned. Each MinHash vector is stored as a compact uint32 byte blob, mirroring the ECFP4 binary-text column.

plinder.data.annotations.get_similarity_scores.load_ligands_from_index(*, annotation: DataFrame) → DataFrame[source]#

Extract post-ingest ligand-similarity inputs from annotation rows.

plinder.data.annotations.get_similarity_scores.load_ligands_from_annotation_table(*, data_dir: Path) → DataFrame[source]#

Load proper holo ligand instances from the collated annotation table.

plinder.data.annotations.get_similarity_scores.annotate_cofactor_similarity(ligands: DataFrame, *, cofactor_smiles: dict[str, str], threshold: float = 90.0, radius: int = 2, nbits: int = 1024) → DataFrame[source]#

Annotate unique ligand SMILES against CCD cofactor structures.

plinder.data.annotations.get_similarity_scores.compute_ligand_fingerprints(*, data_dir: Path, cofactor_similarity_threshold: float = 90.0, retain_score_shards: bool = False) → None[source]#

Fingerprint the unique canonical SMILES in a completed ligand ingest.

plinder.data.annotations.get_similarity_scores.ligand_scores(*, ligand_ids: list[int], data_dir: Path, output_path: Path, number_id_col: str = 'ligand_smiles_id', minimum_similarity: float = 30.0) → None[source]#

Write all BulkTanimoto edges above minimum_similarity.

plinder.data.annotations.get_similarity_scores.mhfp6_ligand_scores(*, ligand_ids: list[int], data_dir: Path, output_path: Path, number_id_col: str = 'ligand_smiles_id', minimum_similarity: float = 30.0) → None[source]#

Write all MHFP6 MinHash-Jaccard edges above minimum_similarity.

plinder.data.annotations.get_similarity_scores.build_ligand_similarity_annotations(*, unique_ligands: DataFrame) → DataFrame[source]#

Return public unique-SMILES annotations without fingerprint bytes.

plinder.data.annotations.get_similarity_scores.fingerprint_basis_sha256(path: Path, *, number_id_col: str, fingerprint_col: str) → str[source]#

Hash the ID-ordered fingerprint vectors that define chemical scores.

plinder.data.annotations.get_similarity_scores.ligand_score_manifest_payload(*, fingerprint_path: Path, fingerprint_col: str, metric: str, minimum_similarity: float, number_id_col: str) → dict[str, str | float][source]#

Describe the exact fingerprint basis and cutoff behind score shards.

plinder.data.annotations.get_similarity_scores.annotate_ligand_similarity(*, data_dir: Path, minimum_similarity: float = 30.0, number_id_col: str = 'ligand_smiles_id') → Path[source]#

Write unique-SMILES identifiers and cofactor annotations for the index.

plinder.data.annotations.get_similarity_scores.align_molecules(reference: Mol, mobile: Mol, max_preiters: int = 100, max_postiters: int = 100) → tuple[float, float][source]#

Shape-align mobile onto reference and return shape/color scores.

plinder.data.annotations.get_similarity_scores.get_feature_map_score(mol_1: Mol, mol_2: Mol, score_mode: int = 0, features_1: Sequence[Any] | None = None, features_2: Sequence[Any] | None = None) → float[source]#

Calculate the normalized pharmacophore feature overlap.

plinder.data.annotations.get_similarity_scores.get_sucos_score(mol_1: Mol, mol_2: Mol, score_mode: int = 0, features_1: Sequence[Any] | None = None, features_2: Sequence[Any] | None = None) → float[source]#

Calculate SuCOS from feature overlap and shape protrusion distance.

plinder.data.annotations.get_similarity_scores.ligand_pharmacophore_records(sdf: bytes, *, label: str) → list[dict[str, Any]] | None[source]#

Detect feature atom memberships and weights once per canonical SDF.

plinder.data.annotations.get_similarity_scores.load_sdf_molecule(sdf_file: Path) → Mol | None[source]#

Load one SDF, falling back to permissive parsing for bad valence.

plinder.data.annotations.get_similarity_scores.load_sdf_molecule_block(sdf: bytes, *, label: str) → Mol | None[source]#

Load one packed SDF record, with a permissive bad-valence fallback.

plinder.data.annotations.get_similarity_scores.prepare_ligand_for_3d_scoring(molecule: Mol) → Mol | None[source]#

Clone a ligand after checking that it contains an explicit heavy atom.

plinder.data.annotations.get_similarity_scores.canonical_ligand_sdf_path(data_dir: Path, *, pdb_id: str, asym_id: str) → Path[source]#

Return the canonical ASU SDF path for one ligand occurrence.

plinder.data.annotations.get_similarity_scores.is_ligand_shape_comparable(sdf_file: Path) → bool[source]#

Test whether one canonical SDF supports shape, color, and SuCOS scoring.

plinder.data.annotations.get_similarity_scores.annotate_ligand_3d_score_ability(ligands: DataFrame, *, data_dir: Path) → DataFrame[source]#

Annotate each ligand occurrence using its canonical ASU conformation.

plinder.data.annotations.get_similarity_scores.get_sequence_similarity(seq_str1: str, seq_str2: str) → tuple[float, float][source]#

Calculate the protein similarity score of an alignment.

If the alignment contains more than two protein sequences, all pairwise scores are counted.

Parameters:
seq_str1str

First protein sequence

seq_str2str

Second protein sequence

Returns:
tuple[float, float]

Sequence identity and sequence similarity score.

plinder.data.annotations.get_similarity_scores.get_sequence_similarity_helper(seq_str1: str, seq_str2: str) → float[source]#
plinder.data.annotations.get_similarity_scores.run_alignment(aln_type: str, query_db: Path, target_db: Path, search_target_db: Path, search_db: Path, aln_file: Path, alignment_config: FoldseekConfig | MMSeqsConfig, cluster_alignment_db: Path | None = None, tmp_dir: Path = PosixPath('/home/runner/work/plinder/plinder/tmp'), remove_tmp: bool = True, threads: int = 1, include_target_pdb_id: bool = True, query_ids_only: bool = False, id_column: str = 'query') → set[str] | None[source]#
plinder.data.annotations.get_similarity_scores.combine_scores(q_t_scores: dict[str, float], q_t_mappings: dict[str, list[tuple[str, str]]], protein_chain_mapper: str = '') → dict[str, str | float][source]#
class plinder.data.annotations.get_similarity_scores.Scorer(entries: 'dict[str, EntryView]', source_to_full_db_file: 'dict[str, Path]', db_dir: 'Path', scores_dir: 'Path', alignment_chain_lookup: 'Path | None' = None, ligand_sdf_resolver: 'Callable[[LigandView], Path | None] | None' = None, foldseek_config: 'FoldseekConfig' = <factory>, mmseqs_config: 'MMSeqsConfig' = <factory>, protein_chain_mappers: 'list[str]' = <factory>, minimum_threshold: 'float' = 0.0, minimum_thresholds: 'dict[str, float]' = <factory>, max_query_protein_chains: 'int' = 30, max_query_proper_ligand_chains: 'int' = 30, shape_score_threads: 'int' = 1, include_pli_fident: 'bool' = False)[source]#

Bases: object

entries: dict[str, EntryView]#
source_to_full_db_file: dict[str, Path]#
db_dir: Path#
scores_dir: Path#
alignment_chain_lookup: Path | None = None#
ligand_sdf_resolver: Callable[[LigandView], Path | None] | None = None#
foldseek_config: FoldseekConfig#
mmseqs_config: MMSeqsConfig#
protein_chain_mappers: list[str]#
minimum_threshold: float = 0.0#
minimum_thresholds: dict[str, float]#
max_query_protein_chains: int = 30#
max_query_proper_ligand_chains: int = 30#
shape_score_threads: int = 1#
include_pli_fident: bool = False#
system_is_query_scoreable(system: SystemView) → bool#

Return whether a holo system is small enough to score as a query.

static system_is_target_scoreable(system: SystemView) → bool#

Return whether a system may contribute scores as a target.

Query limits exist to bound the work initiated by an entry. Applying them to targets as well would remove large systems from the score graph entirely, so every protein-containing holo system remains available as a target. Only proper ligands are emitted later by get_scores_holo().

resolve_ligand_sdf(data_dir: Path, ligand: LigandView) → Path | None#

Resolve the canonical ASU ligand SDF for a ligand annotation.

get_ligand_pair_shape_scores(data_dir: Path, query_ligand: LigandView, target_ligand: LigandView, pocket_qcov: float) → dict[str, float]#

Shape-score one ligand pair after the pocket-coverage gate.

score_canonical_ligand_pairs(data_dir: Path, pairs: DataFrame) → DataFrame#

Calculate base 3D metrics once for each canonical directed pair.

make_dbs() → None#
get_config(search_db: str, aln_type: str) → FoldseekConfig | MMSeqsConfig#
run_alignments(entry_ids: list[str], search_db: str, output_folder: Path, overwrite: bool = False, threads: int = 1, alignment_types: Sequence[str] | None = None, query_chain_auth_ids: Mapping[str, Collection[str]] | None = None, output_metadata_by_entry: Mapping[str, Mapping[bytes, bytes]] | None = None, target_database_dir: Path | None = None, query_database_dir: Path | None = None, result_database_dir: Path | None = None, write_empty_results: bool = True) → set[str]#

Search selected entries and return those with at least one hit.

The optional database roots let update jobs search the current query database against a small changed-target database without touching the release’s existing raw alignments.

map_alignment_files(data_dir: Path, pdb_id: str, search_db: str, overwrite: bool = True, scratch_dir: Path | None = None, mapped_db_dir: Path | None = None) → list[Path]#

Map raw backend alignments without calculating system scores.

get_score_df(data_dir: Path, pdb_id: str, search_db: str, overwrite: bool = True, map_alignments: bool = True, scratch_dir: Path | None = None, source_to_aln_file: dict[str, Path] | None = None, defer_ligand_3d: bool = False, query_entry_alignments: DataFrame | None = None, score_metrics: Collection[str] | None = None) → Path#

Convert aligmnent results to mapped alignment results. Then aggregate the mapped alignment results into the scores dataset.

repair_score_df_targets(data_dir: Path, pdb_id: str, *, affected_target_entries: set[str], search_db: str = 'holo', scratch_dir: Path | None = None, allow_missing: bool = False, query_system_ids: set[str] | None = None, query_ligand_ids: set[str] | None = None, target_system_ids: set[str] | None = None, target_ligand_ids: set[str] | None = None) → Path#

Replace target-entry rows, optionally creating a bounded query.

load_alignments(source_to_aln_file: dict[str, Path], search_db: str = 'holo', query_entry_ids: set[str] | None = None, target_entry_ids: set[str] | None = None) → DataFrame#
map_alignment_df(df_file: Path, aln_type: str, search_db: str) → DataFrame#
map_row(parts: Series, aln_type: str, search_db: str) → Series#

Map selected positions for callers operating on one pandas row.

get_protein_scores_pair(aln: DataFrame) → dict[str, float]#
get_protein_scores(query_target_entry_alignments: DataFrame, query_system: SystemView, target_protein_chains: list[str], query_system_length: int, query_protein_chains: list[str] | None = None) → tuple[dict[str, list[tuple[str, str]]], dict[str, float], dict[tuple[str, str], DataFrame], str]#
get_pocket_pli_scores(alns: dict[tuple[str, str], DataFrame], query_system: SystemView, target_system: SystemView | None = None) → tuple[dict[str, float], dict[str, float]]#
get_ligand_pair_pocket_pli_scores(query_target_entry_alignments: DataFrame, query_ligand: LigandView, target_ligand: LigandView) → tuple[dict[str, float], dict[str, float], dict[str, list[tuple[str, str]]]]#

Score a ligand pair after maximizing its pocket coverage per backend.

get_ligand_pocket_scores(alns: dict[tuple[str, str], DataFrame], query_ligand: LigandView) → dict[str, float]#

Calculate directed pocket identity for a ligand against apo/pred.

get_protein_receptor_chains(pdb_id: str, receptor_chains: list[str]) → list[str]#

Return only polypeptide instance chains from a receptor collection.

get_protein_chain_length(pdb_id: str, protein_chains: list[str]) → int#

Return the total SEQRES length for an instance-chain collection.

get_scores(search_db: str, query_system: SystemView, query_entry_alignments: DataFrame, data_dir: Path | None = None, ligand_3d_candidates: list[dict[str, str | float | None]] | None = None, ligand_pair_scores: list[dict[str, str | int]] | None = None, query_ligand_ids: set[str] | None = None, target_system_ids: set[str] | None = None, target_ligand_ids: set[str] | None = None, include_holo_protein_scores: bool = False, pocket_pair_cache: dict[str, dict[str, set[str]]] | None = None, query_pocket_representatives: dict[str, LigandView] | None = None, target_ligand_cache: dict[str, list[LigandView]] | None = None, pocket_score_cache: dict[tuple[str, str, str], tuple[dict[str, float], dict[str, float], dict[str, list[tuple[str, str]]]]] | None = None, target_alignment_cache: dict[str, DataFrame] | None = None) → Generator[dict[str, str | float | None], None, None]#
get_scores_holo(query_system: SystemView, query_entry_alignments: DataFrame, data_dir: Path | None = None, ligand_3d_candidates: list[dict[str, str | float | None]] | None = None, ligand_pair_scores: list[dict[str, str | int]] | None = None, query_ligand_ids: set[str] | None = None, target_system_ids: set[str] | None = None, target_ligand_ids: set[str] | None = None, include_protein_scores: bool = False, pocket_pair_cache: dict[str, dict[str, set[str]]] | None = None, query_pocket_representatives: dict[str, LigandView] | None = None, target_ligand_cache: dict[str, list[LigandView]] | None = None, pocket_score_cache: dict[tuple[str, str, str], tuple[dict[str, float], dict[str, float], dict[str, list[tuple[str, str]]]]] | None = None, target_alignment_cache: dict[str, DataFrame] | None = None) → Generator[dict[str, str | float | None], None, None]#
get_column_mapr() → dict[str, list[str]]#
aggregate_scores(pdb_id: str, search_db: str = 'holo', data_dir: Path | None = None, source_to_aln_file: dict[str, Path] | None = None, query_entry_alignments: DataFrame | None = None, query_system_ids: set[str] | None = None, query_ligand_ids: set[str] | None = None, target_system_ids: set[str] | None = None, target_ligand_ids: set[str] | None = None, ligand_3d_candidates: list[dict[str, str | float | None]] | None = None, ligand_pair_scores: list[dict[str, str | int]] | None = None, include_holo_protein_scores: bool = False) → DataFrame | None#
get_scores_apo_pred(query_system: SystemView, query_entry_alignments: DataFrame, query_ligand_ids: set[str] | None = None, target_system_ids: set[str] | None = None) → Generator[dict[str, str | float | None], None, None]#