idmtools_platform_comps.utils.file_filter_workitem module#

idmtools FileFilterWorkItem is a interface for SSMT command to act on files using filters in WorkItems.

exception idmtools_platform_comps.utils.file_filter_workitem.CrossEnvironmentFilterNotSupport[source]#

Bases: Exception

Defines cross environment error for when a user tried to filter across multiple comps environments.

doc_link: str = 'platforms/comps/errors.html#errors'#

exception idmtools_platform_comps.utils.file_filter_workitem.AtLeastOneItemToWatch[source]#

Bases: Exception

Defines error for when there are not items being watched by FileFilterWorkItem.

doc_link: str = 'platforms/comps/errors.html#errors'#

class idmtools_platform_comps.utils.file_filter_workitem.FileFilterWorkItem(_uid: str = None, _IItem__pre_creation_hooks: ~typing.List[~typing.Callable[[IItem, IPlatform], None]] = <factory>, _IItem__post_creation_hooks: ~typing.List[~typing.Callable[[IItem, IPlatform], None]] = <factory>, platform_id: str = None, _platform: IPlatform = None, parent_id: str = None, _parent: IEntity = None, status: ~idmtools.core.enums.EntityStatus = None, tags: ~typing.Dict[str, ~typing.Any] = <factory>, _platform_object: ~typing.Any = None, _IRunnableEntity__pre_run_hooks: ~typing.List[~typing.Callable[[IRunnableEntity, IPlatform], None]] = <factory>, _IRunnableEntity__post_run_hooks: ~typing.List[~typing.Callable[[IRunnableEntity, IPlatform], None]] = <factory>, name: str = 'idmtools workflow item', assets: ~idmtools.assets.asset_collection.AssetCollection = <factory>, item_name: dataclasses.InitVar[str] = None, asset_collection_id: dataclasses.InitVar[str] = None, transient_assets: ~idmtools.assets.asset_collection.AssetCollection = <factory>, asset_files: dataclasses.InitVar[FileList] = None, user_files: dataclasses.InitVar[FileList] = None, task: ~idmtools.entities.itask.ITask = None, related_experiments: list = <factory>, related_simulations: list = <factory>, related_suites: list = <factory>, related_work_items: list = <factory>, related_asset_collections: list = <factory>, work_item_type: str = None, work_order: dict = <factory>, plugin_key: str = '1.0.0.0_RELEASE', docker_image: str = None, command: dataclasses.InitVar[str] = None, file_patterns: ~typing.List[str] = <factory>, exclude_patterns: ~typing.List[str] = <factory>, include_assets: bool = False, simulation_prefix_format_str: str = '{simulation.id}', work_item_prefix_format_str: str = None, no_simulation_prefix: bool = False, verbose: bool = False, pre_run_functions: ~typing.List[~typing.Callable] = <factory>, entity_filter_function: ~typing.Callable[[~COMPS.Data.CommissionableEntity.CommissionableEntity], bool] = None, filename_format_function: ~typing.Callable[[str], str] = None, dry_run: bool = False, _ssmt_script: str = None, _ssmt_depends: ~typing.List[str] = <factory>)[source]#

Bases: SSMTWorkItem, ABC

Defines our filtering workitem base that is used by assetize outputs and download work items.

file_patterns: List[str]#

//docs.python.org/3.7/library/glob.html for details on the patterns

Type:: List of glob patterns. See https

exclude_patterns: List[str]#

include_assets: bool = False#: Include Assets directories. This allows patterns to also include items from the assets directory

simulation_prefix_format_str: str = '{simulation.id}'#: Formatting pattern for directory names. Simulations tend to have similar outputs so the workitem puts those in directories using the simulation id by default as the directory name

work_item_prefix_format_str: str = None#: WorkFlowItem outputs will not have a folder prefix by default. If you are filtering multiple work items, you may want to set this to “{workflow_item.id}”

no_simulation_prefix: bool = False#: Simulations outputs will not have a folder. Useful when you are filtering a single simulation

verbose: bool = False#: Enable verbose

pre_run_functions: List[Callable]#: Python Functions that will be ran before Filtering script. The function must be named

entity_filter_function: Callable[[CommissionableEntity], bool] = None#: Python Function to filter entities. This Function should receive a Comps CommissionableEntity. True means include item, false is don’t

filename_format_function: Callable[[str], str] = None#: Function to pass a custom function that is called on the name. This can be used to do advanced mapping or renaming of files

dry_run: bool = False#: Enables running jobs without creating executing. It instead produces a file list of what would be includes in the final filter

create_command() → str[source]#

Builds our command line for the SSMT Job.

Returns:: Command string

clear_exclude_patterns()[source]#

Clear Exclude Patterns will remove all current rules.

Returns:: None

pre_creation(platform: IPlatform) → None[source]#

Pre-Creation.

Parameters:: platform – Platform
Returns:: None

total_items_watched() → int[source]#

Returns the number of items watched.

Returns:: Total number of items watched

run_after_by_id(item_id: str, item_type: ItemType, platform: COMPSPlatform | None = None)[source]#

Runs the workitem after an existing item finishes.

Parameters:

item_id – ItemId
item_type – ItemType
platform – Platform

Returns:

None

Raises:

ValueError - If item_type is not an experiment, simulation, or workflow item –

Add items to load assets from.

Parameters:: item – Item or list of items to watch.
Returns:: None
Raises:: ValueError - If any items specified are not an Experiment, Simulation or WorkItem –

Notes

We should add suite support in the future if possible. This should be done in client side by converting suite to list of experiments.

wait(wait_on_done_progress: bool = True, timeout: int | None = None, refresh_interval=None, platform: COMPSPlatform | None = None) → None[source]#

Waits on Filter Workitem to finish. This first waits on any dependent items to finish(Experiment/Simulation/WorkItems).

Parameters:

wait_on_done_progress – When set to true, a progress bar will be shown from the item
timeout – Timeout for waiting on item. If none, wait will be forever
refresh_interval – How often to refresh progress
platform – Platform

Returns:

AssetCollection created if item succeeds

fetch_error(print_error: bool = True) → Dict[source]#

Fetches the error from a WorkItem.

Parameters:: print_error – Should error be printed. If false, error will be returned
Returns:: Error info

__init__(_uid: str = None, _IItem__pre_creation_hooks: ~typing.List[~typing.Callable[[IItem, IPlatform], None]] = <factory>, _IItem__post_creation_hooks: ~typing.List[~typing.Callable[[IItem, IPlatform], None]] = <factory>, platform_id: str = None, _platform: IPlatform = None, parent_id: str = None, _parent: IEntity = None, status: ~idmtools.core.enums.EntityStatus = None, tags: ~typing.Dict[str, ~typing.Any] = <factory>, _platform_object: ~typing.Any = None, _IRunnableEntity__pre_run_hooks: ~typing.List[~typing.Callable[[IRunnableEntity, IPlatform], None]] = <factory>, _IRunnableEntity__post_run_hooks: ~typing.List[~typing.Callable[[IRunnableEntity, IPlatform], None]] = <factory>, name: str = 'idmtools workflow item', assets: ~idmtools.assets.asset_collection.AssetCollection = <factory>, item_name: dataclasses.InitVar[str] = None, asset_collection_id: dataclasses.InitVar[str] = None, transient_assets: ~idmtools.assets.asset_collection.AssetCollection = <factory>, asset_files: dataclasses.InitVar[FileList] = None, user_files: dataclasses.InitVar[FileList] = None, task: ~idmtools.entities.itask.ITask = None, related_experiments: list = <factory>, related_simulations: list = <factory>, related_suites: list = <factory>, related_work_items: list = <factory>, related_asset_collections: list = <factory>, work_item_type: str = None, work_order: dict = <factory>, plugin_key: str = '1.0.0.0_RELEASE', docker_image: str = None, command: dataclasses.InitVar[str] = None, file_patterns: ~typing.List[str] = <factory>, exclude_patterns: ~typing.List[str] = <factory>, include_assets: bool = False, simulation_prefix_format_str: str = '{simulation.id}', work_item_prefix_format_str: str = None, no_simulation_prefix: bool = False, verbose: bool = False, pre_run_functions: ~typing.List[~typing.Callable] = <factory>, entity_filter_function: ~typing.Callable[[~COMPS.Data.CommissionableEntity.CommissionableEntity], bool] = None, filename_format_function: ~typing.Callable[[str], str] = None, dry_run: bool = False, _ssmt_script: str = None, _ssmt_depends: ~typing.List[str] = <factory>) → None#