Source code for idmtools.entities.iworkflow_item

Defines our IWorkflowItem interface.

Copyright 2021, Bill & Melinda Gates Foundation. All rights reserved.

import warnings
from abc import ABC
from dataclasses import dataclass, field, fields, InitVar
from typing import NoReturn, Dict, Any, TYPE_CHECKING
from idmtools.assets import AssetCollection
from idmtools.assets.file_list import FileList
from idmtools.core import ItemType
from idmtools.core.interfaces.iassets_enabled import IAssetsEnabled
from idmtools.core.interfaces.inamed_entity import INamedEntity
from idmtools.core.interfaces.irunnable_entity import IRunnableEntity
from idmtools.entities.itask import ITask

if TYPE_CHECKING:  # pragma: no cover
    from idmtools.entities.iplatform import IPlatform

[docs]@dataclass class IWorkflowItem(IAssetsEnabled, INamedEntity, IRunnableEntity, ABC): """ Interface of idmtools work item. """ #: Name of the workflow step name: str = field(default=None) #: Legacy name for workflow items item_name: InitVar[str] = None #: Legacy name. Set assets now asset_collection_id: InitVar[str] = None #: Tags associated with the work item tags: Dict[str, Any] = field(default_factory=lambda: {}) #: Common Assets for the workitem transient_assets: AssetCollection = field(default_factory=AssetCollection) #: Legacy var. Going forward use assets asset_files: InitVar[FileList] = None #: Legacy var. Going forward use assets user_files: InitVar[FileList] = None # Task for object. All workflow items must implement by 1.7.0 task: ITask = field(default=None) related_experiments: list = field(default_factory=list) related_simulations: list = field(default_factory=list) related_suites: list = field(default_factory=list) related_work_items: list = field(default_factory=list) related_asset_collections: list = field(default_factory=list) work_item_type: str = field(default=None) item_type: 'ItemType' = field(default=ItemType.WORKFLOW_ITEM, compare=False, init=False) def __post_init__(self, item_name: str, asset_collection_id: str, asset_files: FileList, user_files: FileList): """ Constructor. Args: item_name: Item name asset_collection_id: AssetCollectionId asset_files: AssetFiles user_files: UserFiles Returns: None """ if item_name is not None and not isinstance(item_name, property): = item_name if user_files and not isinstance(user_files, property): # user property since it can convert file lists self.user_files = user_files if asset_files and not isinstance(asset_files, property): # user property since it can convert file lists self.asset_files = asset_files if asset_collection_id and not isinstance(asset_collection_id, property): # user property since it will inform user of changes self.asset_collection_id = asset_collection_id if self.task is None: warnings.warn("In 1.7.0, all work items require a Task") self.tags = self.tags or dict() def __repr__(self): """ String representation of workflow items. """ return f"<WorkItem {self.uid}>"
[docs] def gather_assets(self) -> NoReturn: """ Function called at runtime to gather all assets in the collection. """ pass
[docs] def add_file(self, af): """ Methods used to add new file. Args: af: file to add Returns: None """ self.user_files.add_asset_file(af)
[docs] def clear_user_files(self): """ Clear all existing user files. Returns: None """ self.transient_assets.clear()
[docs] def pre_creation(self, platform: 'IPlatform') -> None: """ Called before the actual creation of the entity. """ if is None: raise ValueError("Name is required") files_to_be_removed = ('comps_log.log', 'idmtools.log') super().pre_creation(platform) if self.task: self.task.pre_creation(self, platform) self.assets.add_assets(self.task.common_assets) self.transient_assets.add_assets(self.task.transient_assets) self.transient_assets.files = [f for f in self.transient_assets.assets if f.filename.lower() not in files_to_be_removed]
def __check_for_platform(self, platform: 'IPlatform'): # noqa: F821 """ Check for platform. If platform is non, we try to load the current platform instead. Args: platform: Platform that can be pre-defined. Returns: None Raises: ValueError - If platform cannot be found """ from idmtools.core.context import CURRENT_PLATFORM if platform is not None: self.platform = platform if self.platform is None: if CURRENT_PLATFORM is None: raise ValueError("Platform is required to run item") self.platform = CURRENT_PLATFORM
[docs] def to_dict(self) -> Dict: """ Convert IWorkflowItem to a dictionary. Returns: Dictionary of WorkflowItem """ result = dict() for f in fields(self): if not"_") and not in ['parent']: result[] = getattr(self, result['_uid'] = self.uid return result