idmtools_platform_comps.utils.file_filter_workitem module

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

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

exception idmtools_platform_comps.utils.file_filter_workitem.CrossEnvironmentFilterNotSupport

Bases: Exception

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

exception idmtools_platform_comps.utils.file_filter_workitem.AtLeastOneItemToWatch

Bases: Exception

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

class idmtools_platform_comps.utils.file_filter_workitem.FileFilterWorkItem(_uid: ~uuid.UUID = 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: ~uuid.UUID = None, _platform: IPlatform = None, parent_id: ~uuid.UUID = 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[UUID] = 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>)

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

Builds our command line for the SSMT Job.

Returns:

Command string

clear_exclude_patterns()

Clear Exclude Patterns will remove all current rules.

Returns:

None

pre_creation(platform: IPlatform) None

Pre-Creation.

Parameters:

platform – Platform

Returns:

None

total_items_watched() int

Returns the number of items watched.

Returns:

Total number of items watched

run_after_by_id(item_id: str, item_type: ItemType, platform: Optional[COMPSPlatform] = None)

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

from_items(item: Union[Experiment, Simulation, IWorkflowItem, List[Union[Experiment, Simulation, IWorkflowItem]]])

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: Optional[int] = None, refresh_interval=None, platform: Optional[COMPSPlatform] = None) None

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

Fetches the error from a WorkItem.

Parameters:

print_error – Should error be printed. If false, error will be returned

Returns:

Error info

__init__(_uid: ~uuid.UUID = 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: ~uuid.UUID = None, _platform: IPlatform = None, parent_id: ~uuid.UUID = 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[UUID] = 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