Source code for idmtools.entities.simulation

Defines our Simulation object.

The simulation object can be thought as a metadata object. It represents a configuration of a remote job execution metadata.
All simulations have a task. Optionally that have assets. All simulations should belong to an Experiment.

Copyright 2021, Bill & Melinda Gates Foundation. All rights reserved.
from dataclasses import dataclass, field, fields
from logging import getLogger, DEBUG
from typing import List, Union, Mapping, Any, Type, TypeVar, Dict, TYPE_CHECKING
from idmtools.assets import AssetCollection, Asset
from idmtools.core import ItemType, NoTaskFound
from idmtools.core.enums import EntityStatus
from idmtools.core.interfaces.iassets_enabled import IAssetsEnabled
from idmtools.core.interfaces.iitem import IItem
from idmtools.core.interfaces.inamed_entity import INamedEntity
from idmtools.entities.task_proxy import TaskProxy
from idmtools.utils.language import get_qualified_class_name_from_obj

if TYPE_CHECKING:  # pragma: no cover
    from idmtools.entities.itask import ITask
    from idmtools.entities.iplatform import IPlatform
    from idmtools.entities.experiment import Experiment

logger = getLogger(__name__)
user_logger = getLogger('user')

[docs]@dataclass class Simulation(IAssetsEnabled, INamedEntity): """ Class that represents a generic simulation. This class needs to be implemented for each model type with specifics. """ #: Task representing the configuration of the command to be executed task: 'ITask' = field(default=None) # noqa: F821 #: Item Type. Should not be changed from Simulation item_type: ItemType = field(default=ItemType.SIMULATION, compare=False) #: Control whether we should replace the task with a proxy after creation __replace_task_with_proxy: bool = field(default=True, init=False, compare=False) #: Ensure we don't gather assets twice __assets_gathered: bool = field(default=False) #: Extra arguments to pass on creation to platform _platform_kwargs: dict = field(default_factory=dict) @property def experiment(self) -> 'Experiment': # noqa: F821 """ Get experiment parent. Returns: Parent Experiment """ return self.parent @experiment.setter def experiment(self, experiment: 'Experiment'): # noqa: F821 """ Set the parent experiment. Args: experiment: Experiment to set as parent. Returns: None """ self.parent = experiment def __repr__(self): """ String representation of simulation. """ return f"<Simulation: {self.uid} - Exp_id: {self.parent_id}>" def __hash__(self): """ Hash of simulation(id). Returns: Hash of simulation """ return id(self.uid)
[docs] def pre_creation(self, platform: 'IPlatform'): """ Runs before a simulation is created server side. Args: platform: Platform the item is being executed on Returns: None """ # skip the IItem created function IItem.pre_creation(self, platform) if self.task is None: msg = 'Task is required for simulations' user_logger.error(msg) raise NoTaskFound(msg) if logger.isEnabledFor(DEBUG): logger.debug('Calling task pre creation') self.task.pre_creation(self, platform) self.gather_assets() if self.__class__ is not Simulation: # Add a tag to keep the Simulation class name sn = get_qualified_class_name_from_obj(self) if logger.isEnabledFor(DEBUG): logger.debug(f'Setting Simulation Tag "simulation_type" to "{sn}"') self.tags["simulation_type"] = sn # Add a tag to for task if self.task is not None: tn = get_qualified_class_name_from_obj(self.task) if logger.isEnabledFor(DEBUG): logger.debug(f'Setting Simulation Tag "task_type" to "{tn}"') self.tags["task_type"] = tn
[docs] def post_creation(self, platform: 'IPlatform') -> None: """ Called after a simulation is created. Args: platform: Platform simulation is being executed on Returns: None """ if logger.isEnabledFor(DEBUG): logger.debug('Calling task post creation') if self.task is not None and not isinstance(self.task, TaskProxy): self.task.post_creation(self, self.platform) IItem.post_creation(self, platform) if self.__replace_task_with_proxy or (self.parent and self.parent._Experiment__replace_task_with_proxy): if logger.isEnabledFor(DEBUG): logger.debug('Replacing task with proxy') self.task = TaskProxy.from_task(self.task) if self.status is None: self.status = EntityStatus.CREATED
[docs] def pre_getstate(self): """ Return default values for :meth:`pickle_ignore_fields`. Call before pickling. """ from idmtools.assets import AssetCollection from idmtools.core.interfaces.entity_container import EntityContainer return {"assets": AssetCollection(), "simulations": EntityContainer()}
[docs] def gather_assets(self): """ Gather all the assets for the simulation. """ if not self.__assets_gathered: self.task.gather_transient_assets() self.assets.add_assets(self.task.transient_assets, fail_on_duplicate=False) self.__assets_gathered = True
[docs] @classmethod def from_task(cls, task: 'ITask', tags: Dict[str, Any] = None, # noqa E821 asset_collection: AssetCollection = None): """ Create a simulation from a task. Args: task: Task to create from tags: Tags to create on the simulation asset_collection: Simulation Assets Returns: Simulation using the parameters provided """ return Simulation(task=task, tags=dict() if tags is None else tags, assets=asset_collection if asset_collection else AssetCollection())
[docs] def list_static_assets(self, platform: 'IPlatform' = None, **kwargs) -> List[Asset]: """ List assets that have been uploaded to a server already. Args: platform: Optional platform to load assets list from **kwargs: Returns: List of assets Raises: ValueError - If you try to list an assets for an simulation that hasn't been created/loaded from a remote platform. """ if is None: raise ValueError("You can only list static assets on an existing experiment") p = super()._check_for_platform_from_context(platform) return p._simulations.list_assets(self, **kwargs)
[docs] def to_dict(self) -> Dict: """ Do a lightweight conversation to json. Returns: Dict representing json of object """ result = dict() for f in fields(self): if not"_") and not in ['parent']: result[] = getattr(self, result['_uid'] = self.uid result['task'] = self.task.to_dict() if self.task else None return result
# TODO Rename to T simulation once old simulation is one TTSimulation = TypeVar("TTSimulation", bound=Simulation) TTSimulationClass = Type[TTSimulation] TTSimulationBatch = List[TTSimulation] TTAllSimulationData = Mapping[TTSimulation, Any] TTSimulationList = List[Union[TTSimulation, str]]