schrodinger.application.transforms.fep module¶

class schrodinger.application.transforms.fep.GraphCoder¶

Bases: Coder

encode(graph: Graph) → bytes¶: Encodes the given object into a byte string.

is_deterministic()¶

Whether this coder is guaranteed to encode values deterministically.

A deterministic coder is required for key coders in GroupByKey operations to produce consistent results.

For example, note that the default coder, the PickleCoder, is not deterministic: the ordering of picked entries in maps may vary across executions since there is no defined order, and such a coder is not in general suitable for usage as a key coder in GroupByKey operations, since each instance of the same key may be encoded differently.

Returns:: Whether coder is deterministic.

decode(encoded: bytes) → Graph¶: Decodes the given byte string into the corresponding object.

to_type_hint() → type¶

class schrodinger.application.transforms.fep.FEPProtocol(opls_dir: Optional[str] = None, restraints_file: Optional[str] = None, ff: str = 'OPLS4', equilibration_time: int = 20, simulation_time: int = 5000, ensemble: str = 'muVT', water_model: str = 'SPC', salt_concentration: float = 0.0, membrane_relaxation: bool = False, membrane_type: Optional[str] = None, torsion_scaling: bool = False, num_windows: int = 12, num_core_hopping_windows: int = 16, num_charged_windows: int = 24)¶

Bases: object

opls_dir: Optional[str] = None¶

restraints_file: Optional[str] = None¶

ff: str = 'OPLS4'¶

equilibration_time: int = 20¶

simulation_time: int = 5000¶

ensemble: str = 'muVT'¶

water_model: str = 'SPC'¶

salt_concentration: float = 0.0¶

membrane_relaxation: bool = False¶

membrane_type: Optional[str] = None¶

torsion_scaling: bool = False¶

num_windows: int = 12¶

num_core_hopping_windows: int = 16¶

num_charged_windows: int = 24¶

getPaths() → Dict[str, Path]¶

validate()¶

classmethod FromFile(path: Union[str, Path])¶

classmethod FromFEPArgsString(fep_args: Optional[str]) → FEPProtocol¶

Parse an FEP+ args string into an FEPProtocol.

Uses the spec definitions in parameters.py to map CLI arg names (e.g. “-time”, “-lambda-windows”) to FEPProtocol fields. Unrecognized args are skipped.

Parameters:: fep_args – FEP+ args string (e.g. “-time 2000 -lambda-windows 10”)

__init__(opls_dir: Optional[str] = None, restraints_file: Optional[str] = None, ff: str = 'OPLS4', equilibration_time: int = 20, simulation_time: int = 5000, ensemble: str = 'muVT', water_model: str = 'SPC', salt_concentration: float = 0.0, membrane_relaxation: bool = False, membrane_type: Optional[str] = None, torsion_scaling: bool = False, num_windows: int = 12, num_core_hopping_windows: int = 16, num_charged_windows: int = 24) → None¶

class schrodinger.application.transforms.fep.FEPWebServicesSettings(project_name: str = 'project', schrodinger_path: str = 'OB/2026-3/release')¶

Bases: object

The settings for the FEP job to be run using Web Services.

project_name: str = 'project'¶

schrodinger_path: str = 'OB/2026-3/release'¶

validate()¶

Validate the settings for use in step.

Parameters:: step – stepper._BaseStep
Return type:: list[TaskError or TaskWarning]

__init__(project_name: str = 'project', schrodinger_path: str = 'OB/2026-3/release') → None¶

class schrodinger.application.transforms.fep.FEPJobServerSettings(host: str, subhost: str, cpus: int = 1, max_jobs: int = 0, max_simulataneous_graphs: int = 0, retries: int = 3, queue_args: str = '')¶

Bases: object

host: str¶

subhost: str¶

cpus: int = 1¶

max_jobs: int = 0¶

max_simulataneous_graphs: int = 0¶

retries: int = 3¶

queue_args: str = ''¶

validate()¶

__init__(host: str, subhost: str, cpus: int = 1, max_jobs: int = 0, max_simulataneous_graphs: int = 0, retries: int = 3, queue_args: str = '') → None¶

class schrodinger.application.transforms.fep.FEPOutputInclusionSettings(fmpdb: bool = True)¶

Bases: object

What type of information, in addition to the graph output, that should be returned and written to the output directory.

fmpdb: bool = True¶

__init__(fmpdb: bool = True) → None¶

class schrodinger.application.transforms.fep.FEPRunResult(jobname: str, graph: Optional[Graph] = None, fmpdb: Optional[DataFile] = None)¶

Bases: object

The results of the FEP calculation.

jobname: str¶

graph: Optional[Graph] = None¶

fmpdb: Optional[DataFile] = None¶

__init__(jobname: str, graph: Optional[Graph] = None, fmpdb: Optional[DataFile] = None) → None¶

class schrodinger.application.transforms.fep.WriteFEPResults(output_dir: Union[str, Path])¶

Bases: _LocalOnlyPTransform, WriteFEPResults

Write the FEP results to the output directory.

Note: this will happily overwrite existing files.

schrodinger.application.transforms.fep.get_fmp_filename(jobname: str) → str¶

Get the fmp filename from the jobname.

Parameters:: jobname – the jobname for the FEP calculation
Returns:: the fmp filename

schrodinger.application.transforms.fep.get_out_fmp_filename(jobname: str) → str¶

Get the output fmp filename from the jobname.

Parameters:: jobname – the jobname for the FEP calculation
Returns:: the output fmp filename

schrodinger.application.transforms.fep.get_out_fmpdb_filename(jobname: str) → str¶

Get the output fmpdb filename from the jobname.

Parameters:: jobname – the jobname for the FEP calculation
Returns:: the output fmpdb filename

class schrodinger.application.transforms.fep.FEPInput(graph: Graph, protocol: FEPProtocol, jobname: Optional[str] = '', graphname: Optional[str] = '')¶

Bases: object

The input for the RunFEP transform.

The default jobname is ‘FEP’ unless the transform is already running under jobcontrol, in which case jobcontol’s jobname, will be used. The default graphname is equal to the jobname

graph: Graph¶

protocol: FEPProtocol¶

jobname: Optional[str] = ''¶

graphname: Optional[str] = ''¶

dependent_files: Dict[str, DataFile]¶

__init__(graph: Graph, protocol: FEPProtocol, jobname: Optional[str] = '', graphname: Optional[str] = '') → None¶

class schrodinger.application.transforms.fep.FEPWSSubmitResult(graph_id: str, jobname: str)¶

Bases: object

Handle linking a web services graph ID to its job.

Produced by SubmitFEPWS and consumed by CollectFEPWS.

graph_id: str¶

jobname: str¶

__init__(graph_id: str, jobname: str) → None¶

class schrodinger.application.transforms.fep.RunFEP(execution_settings: Union[FEPJobServerSettings, FEPWebServicesSettings], output_inclusion_settings: Optional[FEPOutputInclusionSettings] = None)¶

Bases: PTransform

Run the FEP calculations using either Web Services or JobServer.

It is assumed that the jobname for the FEPInputs are unique.

Note: a side effect of this transform is that a job directory with the jobname will be created in the working directory.

Example usage for Web Services runs:

graph = Graph.deserialize('in.fmp')
protocol = fep.FEPProtocol()
execution_settings = fep.FEPWebServicesSettings(
    project_name='dev', schrodinger_path='OB/2024-1/release')
fep_inputs = [fep.FEPInput(graph, protocol, jobname='graphdb_test')]
with beam.Pipeline() as p:
    (p
     | beam.Create(fep_inputs)
     | fep.RunFEP(execution_settings, FEPOutputInclusionSettings())
     | fep.WriteFEPResults(working_dir)
    )

To run the FEP calculations on a jobserver, only the execution_settings needs to be different, e.g.,:

execution_settings = fep.FEPJobServerSettings('cpu_host', 'gpu_host')

with a protocol yaml file for either usage something along the lines of:

opls_dir: custom_2024_1.opls
ff: OPLS4
restraints_file: restraints.txt
equilibration_time: 10  # ps
simulation_time: 500    # ps
ensemble:
water_model: SPC
torsion_scaling: false
num_windows: 12
num_core_hopping_windows: 16
num_charged_windows: 24

Parameters:

execution_settings – the settings for either Web Services or JobServer
output_inclusion_settings – the settings for what to return

__init__(execution_settings: Union[FEPJobServerSettings, FEPWebServicesSettings], output_inclusion_settings: Optional[FEPOutputInclusionSettings] = None)¶

Parameters:

execution_settings – the settings for either Web Services or JobServer
output_inclusion_settings – the settings for what to return. Default is to return what the default FEPOutputInclusionSettings stipulates.

expand(fep_inputs)¶

class schrodinger.application.transforms.fep.SubmitFEPWS(execution_settings: FEPWebServicesSettings)¶

Bases: PTransform

Submit FEP calculations to web services.

Takes a PCollection of FEPInput and yields FEPWSSubmitResult handles. Submission is fast; no polling occurs here.

__init__(execution_settings: FEPWebServicesSettings)¶

expand(fep_inputs)¶

class schrodinger.application.transforms.fep.CollectFEPWS(execution_settings: FEPWebServicesSettings, output_inclusion_settings: Optional[FEPOutputInclusionSettings] = None)¶

Bases: PTransform

Collect FEP results from web services.

Takes a PCollection of FEPWSSubmitResult handles, polls for completion, and yields FEPRunResult objects.

__init__(execution_settings: FEPWebServicesSettings, output_inclusion_settings: Optional[FEPOutputInclusionSettings] = None)¶

expand(submit_results)¶

schrodinger.application.transforms.fep.make_fepplus_cmd(fmp_filename: str, protocol_yaml_path: Path, jobserver_settings: FEPJobServerSettings, jobname: str) → List[str]¶

Prepare FEP submission scripts and return the command to run fep_plus.

Parameters:

fmp_filename – the .fmp file name
protocol_yaml_path – the protocol yaml file path

Returns:

the arg list to run fep_plus

class schrodinger.application.transforms.fep.FEPMapSettings(core_smarts: Optional[str] = None, topology: Optional[str] = None, receptor_hotatoms_asl: str = '', ligand_hotatoms_rule_complex: str = 'default', ligand_hotatoms_rule_solvent: str = 'default', disable_supernodes: bool = False, generate_neutral_intermediates: bool = False)¶

Bases: object

Settings that control FEP map generation (topology, atom mapping, etc.).

Parameters:

core_smarts – SMARTS pattern for atom mapping
topology – Graph topology (e.g., ‘star’, ‘normal’)
receptor_hotatoms_asl – ASL for receptor hot atoms
ligand_hotatoms_rule_complex – Hot atoms rule for complex
ligand_hotatoms_rule_solvent – Hot atoms rule for solvent
disable_supernodes – Whether to disable supernodes
generate_neutral_intermediates – Whether to generate neutral intermediates

core_smarts: Optional[str] = None¶

topology: Optional[str] = None¶

receptor_hotatoms_asl: str = ''¶

ligand_hotatoms_rule_complex: str = 'default'¶

ligand_hotatoms_rule_solvent: str = 'default'¶

disable_supernodes: bool = False¶

generate_neutral_intermediates: bool = False¶

__init__(core_smarts: Optional[str] = None, topology: Optional[str] = None, receptor_hotatoms_asl: str = '', ligand_hotatoms_rule_complex: str = 'default', ligand_hotatoms_rule_solvent: str = 'default', disable_supernodes: bool = False, generate_neutral_intermediates: bool = False) → None¶

class schrodinger.application.transforms.fep.FEPMapperInput(env_structures: ~typing.List[~schrodinger.structure._structure.Structure], reference_ligands: ~typing.List[~schrodinger.structure._structure.Structure], ligands: ~typing.List[~schrodinger.structure._structure.Structure], jobname: str = 'FEPMapper', map_settings: ~schrodinger.application.transforms.fep.FEPMapSettings = <factory>)¶

Bases: object

Input for the FEP Mapper transform.

Parameters:

env_structures – Environmental structures (receptor, membrane, etc.)
reference_ligands – Reference ligands to include as biased nodes
ligands – Ligands to include in the FEP map
jobname – Job name for output files (default: ‘FEPMapper’)
map_settings – Settings for FEP map generation

env_structures: List[Structure]¶

reference_ligands: List[Structure]¶

ligands: List[Structure]¶

jobname: str = 'FEPMapper'¶

map_settings: FEPMapSettings¶

__init__(env_structures: ~typing.List[~schrodinger.structure._structure.Structure], reference_ligands: ~typing.List[~schrodinger.structure._structure.Structure], ligands: ~typing.List[~schrodinger.structure._structure.Structure], jobname: str = 'FEPMapper', map_settings: ~schrodinger.application.transforms.fep.FEPMapSettings = <factory>) → None¶

class schrodinger.application.transforms.fep.FEPMapper(label: Optional[str] = None)¶

Bases: PTransform

Generate an FEP perturbation map from a PCollection of FEPMapperInput objects.

This transform takes a PCollection of FEPMapperInput objects (containing structures and mapper settings) and generates an optimized FEP graph using fep_mapper.py.

Example usage:

mapper_inputs = (p
    | 'Create inputs' >> beam.Create([
        FEPMapperInput(
            env_structures=[receptor_st],
            reference_ligands=[ref_ligand],
            ligands=[ligand1, ligand2, ligand3],
            jobname='my_fep_map',
            map_settings=FEPMapSettings(topology='star'),
        )
    ])
)

graphs = mapper_inputs | 'Generate FEP Map' >> FEPMapper()

expand(mapper_inputs)¶