schrodinger.application.transforms.retrosynth module¶

Retrosynthesis transform for predicting synthesis feasibility via LDML.

This module provides Apache Beam transforms for running retrosynthesis predictions on molecular structures using the LiveDesign ML API.

Authentication¶

These transforms rely on the LDMLClient singleton for LDML API access. Use setup_client() (from ldml) to configure the singleton. There are two ways to authenticate:

Explicit credentials: Pass ldml_url and jsession_id to the transform config. setup_client() will call LDMLClient.createFromConfig() directly.
Stored credentials: Pass only ldml_url. setup_client() looks up a refresh token from the credentials store (see schrodinger.application.livedesign.credentials) and authenticates via LiveDesign.

The transforms do not disconnect the LDMLClient on teardown, since the singleton may be shared with other transforms in the same pipeline.

schrodinger.application.transforms.retrosynth.get_ldml_credentials(host: str | None = None, username: str | None = None, password: str | None = None, token: str | None = None) → dict[str, str]¶

Obtain LDML credentials for use with RetroSynthConfig.

If login parameters are provided, a new LiveDesign connection is established first. Otherwise, the current session is used (e.g. from an active Maestro login).

For SSO-enabled servers, use host with token (a refresh token from the SSO login flow). For servers with direct login, use host with username and password.

Example:

# From an existing session:
creds = get_ldml_credentials()

# With username/password:
creds = get_ldml_credentials(
    host='https://livedesign.example.com',
    username='user', password='pass')

# With SSO token:
creds = get_ldml_credentials(
    host='https://livedesign.example.com',
    token='<refresh_token>')

config = RetroSynthConfig(project_name='my_project', **creds)

Parameters:

host – LiveDesign server URL.
username – LiveDesign username (for direct login).
password – LiveDesign password (for direct login).
token – SSO refresh token (for SSO-enabled servers).

Returns:

Dict with ldml_url and jsession_id keys.

Raises:

LDMLClientError – If login fails or credentials cannot be obtained.
ValueError – If login parameters are incomplete.

class schrodinger.application.transforms.retrosynth.RetroSynthConfig(*, ldml_url: HttpUrl, ld_url: pydantic.networks.HttpUrl | None = None, jsession_id: str | None = None, polling_interval: int = 600, max_retries: int = 5, project_name: str, batch_size: Optional[int] = None)¶

Bases: LDMLConfig

Configuration for retrosynthesis transforms.

Parameters:

project_name – LDML retrosynthesis project name.
batch_size – Max structures per LDML job. When None (default), all structures are submitted in a single job.

project_name: str¶

batch_size: Optional[int]¶

model_config: ClassVar[ConfigDict] = {'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class schrodinger.application.transforms.retrosynth.RetroSynthResult(input_structure: Structure, success: bool, job_id: str | None = None, corporate_id: str | None = None, synthesis_difficulty_score: int | None = None, synthesis_forward_relative_score: float | None = None, synthesis_forward_score: float | None = None, synthesis_is_solved: bool | None = None, synthesis_num_routes: int | None = None, synthesis_number_of_steps: int | None = None, synthesis_score: float | None = None, error_message: str | None = None)¶

Bases: object

Result of running retrosynthesis on a single structure.

Parameters:

input_structure – The input structure, unchanged.
success – Whether retrosynthesis succeeded.
job_id – LDML retrosynthesis job ID.
corporate_id – Corporate ID used to identify this structure in the LDML job. Together with job_id, this can be used to retrieve cloud-stored results later.
synthesis_difficulty_score – Difficulty score.
synthesis_forward_relative_score – Forward relative score.
synthesis_forward_score – Forward score.
synthesis_is_solved – Whether the synthesis is solved.
synthesis_num_routes – Number of routes found.
synthesis_number_of_steps – Number of synthesis steps.
synthesis_score – Overall synthesis score.
error_message – Error message if processing failed.

input_structure: Structure¶

success: bool¶

job_id: str | None = None¶

corporate_id: str | None = None¶

synthesis_difficulty_score: int | None = None¶

synthesis_forward_relative_score: float | None = None¶

synthesis_forward_score: float | None = None¶

synthesis_is_solved: bool | None = None¶

synthesis_num_routes: int | None = None¶

synthesis_number_of_steps: int | None = None¶

synthesis_score: float | None = None¶

error_message: str | None = None¶

structure_properties() → dict[str, int | float | bool | str | None]¶

Return retrosynthesis results as a structure property dict.

Keys are fully qualified property names (e.g. r_ldml_synthesis_score, s_ldml_job_id) suitable for setting directly on a Structure. Includes synthesis scores and job identifiers; values are None when no result was returned for that attribute.

Returns:: Mapping of property names to values (or None).

__init__(input_structure: Structure, success: bool, job_id: str | None = None, corporate_id: str | None = None, synthesis_difficulty_score: int | None = None, synthesis_forward_relative_score: float | None = None, synthesis_forward_score: float | None = None, synthesis_is_solved: bool | None = None, synthesis_num_routes: int | None = None, synthesis_number_of_steps: int | None = None, synthesis_score: float | None = None, error_message: str | None = None) → None¶

class schrodinger.application.transforms.retrosynth.RetroSynthSubmission(id_to_input_structure: dict[str, Structure], job_id: str | None = None, error_message: str | None = None)¶

Bases: object

Intermediate result from submitting a retrosynthesis job.

Bridges the submit and collect phases, carrying enough state for the collect phase to poll, fetch results, and match them back to the original input structures.

Parameters:

id_to_input_structure – Mapping of corporate ID to original input structure.
job_id – LDML job ID, or None if submission failed.
error_message – Error message if submission failed.

id_to_input_structure: dict[str, Structure]¶

job_id: str | None = None¶

error_message: str | None = None¶

__init__(id_to_input_structure: dict[str, Structure], job_id: str | None = None, error_message: str | None = None) → None¶

class schrodinger.application.transforms.retrosynth.SubmitRetroSynth(**kwargs)¶

Bases: PTransformWithConfig

A PTransform that submits a retrosynthesis job and returns a submission.

Collects input structures into a batch, assigns corporate IDs, and submits the batch to LDML. Returns a RetroSynthSubmission that can be passed to CollectRetroSynth to poll and retrieve results.

See RetroSynthConfig for available configuration options.

config_class¶: alias of RetroSynthConfig

class schrodinger.application.transforms.retrosynth.CollectRetroSynth(**kwargs)¶

Bases: PTransformWithConfig

A PTransform that polls and collects retrosynthesis results.

Takes a RetroSynthSubmission from SubmitRetroSynth, polls for job completion, fetches results, and yields a RetroSynthResult for each input structure.

See RetroSynthConfig for available configuration options.

config_class¶: alias of RetroSynthConfig

class schrodinger.application.transforms.retrosynth.RunRetroSynth(**kwargs)¶

Bases: PTransformWithConfig

A PTransform that runs retrosynthesis and returns results.

Returns a RetroSynthResult for every input structure, indicating whether processing succeeded and containing synthesis scores.

Example usage:

>>> from schrodinger import adapter
>>> st = adapter.to_structure('c1ccccc1')
>>> with beam.Pipeline() as p:
...     results = (p
...         | beam.Create([st])
...         | RunRetroSynth(project_name='my_project')
...         | beam.Map(lambda r: (r.success, r.synthesis_score))
...         | beam.LogElements())

See RetroSynthConfig for available configuration options.

config_class¶: alias of RetroSynthConfig

class schrodinger.application.transforms.retrosynth.AnnotateRetroSynth(**kwargs)¶

Bases: PTransformWithConfig

A PTransform that annotates structures with retrosynthesis properties.

This is a convenience wrapper around RunRetroSynth that sets synthesis properties directly on the input structures. On success, properties are replaced with fresh values. On failure, existing properties are preserved unchanged.

Example usage:

>>> from schrodinger import adapter
>>> st = adapter.to_structure('c1ccccc1')
>>> with beam.Pipeline() as p:
...     annotated = (p
...         | beam.Create([st])
...         | AnnotateRetroSynth(project_name='my_project')
...         | beam.LogElements())

See RetroSynthConfig for available configuration options.

config_class¶: alias of RetroSynthConfig