"""
IEntity definition. IEntity is the base of most of our Remote server entitiies like Experiment, Simulation, WorkItems, and Suites.
Copyright 2021, Bill & Melinda Gates Foundation. All rights reserved.
"""
from abc import ABCMeta
from dataclasses import dataclass, field
from logging import getLogger
from os import PathLike
from pathlib import Path
from typing import NoReturn, List, Any, Dict, Union, TYPE_CHECKING
from idmtools.core import EntityStatus, ItemType, NoPlatformException
from idmtools.core.interfaces.iitem import IItem
from idmtools.core.id_file import read_id_file, write_id_file
from idmtools.services.platforms import PlatformPersistService
if TYPE_CHECKING: # pragma: no cover
from idmtools.entities.iplatform import IPlatform
logger = getLogger(__name__)
[docs]
@dataclass
class IEntity(IItem, metaclass=ABCMeta):
"""
Interface for all entities in the system.
"""
#: ID of the platform
platform_id: str = field(default=None, compare=False, metadata={"md": True})
#: Platform
_platform: 'IPlatform' = field(default=None, compare=False, metadata={"pickle_ignore": False}) # noqa E821
#: Parent id
parent_id: str = field(default=None, metadata={"md": True})
#: Parent object
_parent: 'IEntity' = field(default=None, compare=False, metadata={"pickle_ignore": False})
#: Status of item
status: EntityStatus = field(default=None, compare=False, metadata={"pickle_ignore": False})
#: Tags for item
tags: Dict[str, Any] = field(default_factory=lambda: {}, metadata={"md": True})
#: Item Type(Experiment, Suite, Asset, etc)
item_type: ItemType = field(default=None, compare=False)
#: Platform Representation of Entity
_platform_object: Any = field(default=None, compare=False, metadata={"pickle_ignore": True})
[docs]
def post_creation(self, platform: 'IPlatform') -> None:
"""
Post creation hook for object.
Returns:
None
"""
self.status = EntityStatus.CREATED
super().post_creation(platform)
[docs]
@classmethod
def from_id_file(cls, filename: Union[PathLike, str], platform: 'IPlatform' = None,
**kwargs) -> 'IEntity': # noqa E821:
"""
Load from a file that container the id.
Args:
filename: Filename to load
platform: Platform object to load id from. This can be loaded from file if saved there.
**kwargs: Platform extra arguments
Returns:
Entity loaded from id file
Raises:
EnvironmentError if item type is None.
"""
item_id, item_type_in_file, platform_block, extra_args = read_id_file(filename)
if platform is None:
if platform_block:
from idmtools.core.platform_factory import Platform
platform = Platform(platform_block, **kwargs)
else:
platform = cls.get_current_platform_or_error()
if cls.item_type is None:
raise EnvironmentError(
"ItemType is None. This is most likely a badly derived IEntity that doesn't run set the default item type on the class")
return platform.get_item(item_id, cls.item_type, **kwargs)
[docs]
@classmethod
def from_id(cls, item_id: str, platform: 'IPlatform' = None, **kwargs) -> 'IEntity': # noqa E821
"""
Load an item from an id.
Args:
item_id: Id of item
platform: Platform. If not supplied, we check the current context
**kwargs: Optional platform args
Returns:
IEntity of object
"""
if platform is None:
platform = cls.get_current_platform_or_error()
if cls.item_type is None:
raise EnvironmentError(
"ItemType is None. This is most likely a badly derived IEntity that doesn't run set the default item type on the class")
return platform.get_item(item_id, cls.item_type, **kwargs)
@property
def parent(self):
"""
Return parent object for item.
Returns:
Parent entity if set
"""
if not self._parent:
if not self.parent_id:
return None
if not self.platform:
raise NoPlatformException("The object has no platform set...")
self._parent = self.platform.get_parent(self.uid, self.item_type)
return self._parent
@parent.setter
def parent(self, parent: 'IEntity'): # noqa E821
"""
Sets the parent object for item.
Args:
parent: Parent object
Returns:
None
"""
if parent:
self._parent = parent
self.parent_id = parent.uid
else:
self.parent_id = self._parent = None
@property
def platform(self) -> 'IPlatform': # noqa E821
"""
Get objects platform object.
Returns:
Platform
"""
if not self._platform and self.platform_id:
self._platform = PlatformPersistService.retrieve(self.platform_id)
return self._platform
@platform.setter
def platform(self, platform: 'IPlatform'): # noqa E821
"""
Sets object platform.
Args:
platform: Platform to set
Returns:
None
"""
if platform:
self.platform_id = platform.uid
self._platform = platform
else:
self._platform = self.platform_id = None
@property
def done(self):
"""
Returns if a item is done.
For an item to be done, it should be in either failed or succeeded state.
Returns:
True if status is succeeded or failed
"""
return self.status in (EntityStatus.SUCCEEDED, EntityStatus.FAILED)
@property
def succeeded(self):
"""
Returns if an item has succeeded.
Returns:
True if status is SUCCEEDED
"""
return self.status == EntityStatus.SUCCEEDED
@property
def failed(self):
"""
Returns is a item has failed.
Returns:
True if status is failed
"""
return self.status == EntityStatus.FAILED
def __hash__(self):
"""
Returns hash for object. For entities, the hash is the id.
Returns:
Hash id
"""
return id(self.uid)
def _check_for_platform_from_context(self, platform) -> 'IPlatform':
"""
Try to determine platform of current object from self or current platform.
Args:
platform: Passed in platform object
Raises:
NoPlatformException: when no platform is on current context
Returns:
Platform object
"""
if self.platform is None:
# check context for current platform
if platform is None:
platform = self.get_current_platform_or_error()
self.platform = platform
return self.platform
[docs]
def to_id_file(self, filename: Union[str, PathLike], save_platform: bool = False, platform_args: Dict = None):
"""
Write a id file.
Args:
filename: Filename to create
save_platform: Save platform to the file as well
platform_args: Platform Args
Returns:
None
"""
write_id_file(filename, self, save_platform, platform_args)
[docs]
def get_directory(self) -> Path:
"""
Retrieve the directory path associated with the current item.
This method returns the local platform-specific directory for the item,
such as a simulation, experiment, or suite. It is only supported for
non-COMPS platforms and for items of type SIMULATION, EXPERIMENT, or SUITE.
Raises:
RuntimeError: If the current platform is COMPSPlatform, or if the item
type is not one of SIMULATION, EXPERIMENT, or SUITE.
AttributeError: If the item does not have an associated platform object.
Returns:
pathlib.Path: The path to the item's working directory on the current platform.
"""
platform = self.get_current_platform_or_error()
if not hasattr(platform, 'job_directory'):
raise RuntimeError(f'Not support get_directory for {platform.__class__.__name__}')
if self.item_type not in (ItemType.SIMULATION, ItemType.EXPERIMENT, ItemType.SUITE):
raise RuntimeError('Only support Suite/Experiment/Simulation for get_directory() for now.')
return platform.get_directory(self)
@property
def directory(self) -> Path:
"""
The get directory for the current item. This is a convenience alias for get_directory.
Returns:
pathlib.Path: The path to the item's working directory on the current platform.
"""
return self.get_directory()
IEntityList = List[IEntity]
