"""
The direct simulation estimation layer.
"""
import logging
from os import path
from propertyestimator.layers import (
PropertyCalculationLayer,
register_calculation_layer,
)
from propertyestimator.workflow import Workflow, WorkflowGraph
[docs]@register_calculation_layer()
class SimulationLayer(PropertyCalculationLayer):
"""A calculation layer which aims to calculate physical properties
directly from molecular simulation.
.. warning :: This class is experimental and should not be used in a production environment.
"""
@staticmethod
def _build_workflow_graph(
working_directory,
properties,
force_field_path,
parameter_gradient_keys,
options,
):
""" Construct a graph of the protocols needed to calculate a set of properties.
Parameters
----------
working_directory: str
The local directory in which to store all local,
temporary calculation data from this graph.
properties : list of PhysicalProperty
The properties to attempt to compute.
force_field_path : str
The path to the force field parameters to use in the workflow.
parameter_gradient_keys: list of ParameterGradientKey
A list of references to all of the parameters which all observables
should be differentiated with respect to.
options: PropertyEstimatorOptions
The options to run the workflows with.
"""
workflow_graph = WorkflowGraph(working_directory)
for property_to_calculate in properties:
property_type = type(property_to_calculate).__name__
if property_type not in options.workflow_schemas:
logging.warning(
"The property calculator does not support {} "
"workflows.".format(property_type)
)
continue
if SimulationLayer.__name__ not in options.workflow_schemas[property_type]:
continue
schema = options.workflow_schemas[property_type][SimulationLayer.__name__]
workflow_options = options.workflow_options[property_type].get(
SimulationLayer.__name__
)
global_metadata = Workflow.generate_default_metadata(
property_to_calculate,
force_field_path,
parameter_gradient_keys,
workflow_options,
)
workflow = Workflow(property_to_calculate, global_metadata)
workflow.schema = schema
from propertyestimator.properties import CalculationSource
workflow.physical_property.source = CalculationSource(
fidelity=SimulationLayer.__name__, provenance={}
)
workflow_graph.add_workflow(workflow)
return workflow_graph
[docs] @staticmethod
def schedule_calculation(
calculation_backend,
storage_backend,
layer_directory,
data_model,
callback,
synchronous=False,
):
# Store a temporary copy of the force field for protocols to easily access.
force_field_source = storage_backend.retrieve_force_field(
data_model.force_field_id
)
force_field_path = path.join(
layer_directory, "force_field_{}".format(data_model.force_field_id)
)
with open(force_field_path, "w") as file:
file.write(force_field_source.json())
workflow_graph = SimulationLayer._build_workflow_graph(
layer_directory,
data_model.queued_properties,
force_field_path,
data_model.parameter_gradient_keys,
data_model.options,
)
simulation_futures = workflow_graph.submit(calculation_backend)
PropertyCalculationLayer._await_results(
calculation_backend,
storage_backend,
layer_directory,
data_model,
callback,
simulation_futures,
synchronous,
)