emod_task

EMODTask dataclass

Bases: ITask

Central orchestration class for configuring and running EMOD simulations.

EMODTask extends idmtools.ITask and wires together all the components needed for an EMOD experiment: configuration parameters, campaign interventions, demographics, migration files, climate data, reporters, and the Eradication binary. It manages both experiment-level assets (shared across all simulations) and simulation-level assets (unique per simulation).

Two factory methods are provided for creating tasks:

• from_defaults — builds everything from schema defaults using user-supplied callback functions (config_builder, campaign_builder, demographics_builder, report_builder). This is the primary entry point and supports parameter sweeps.
• from_files — loads pre-existing JSON files (config, campaign, demographics, reports) directly, for working with hand-authored or legacy configurations.

During the idmtools lifecycle, pre_creation finalizes the configuration by merging migrations, wiring reporters into config, setting campaign filenames, and building the command line. Asset gathering methods (gather_common_assets, gather_transient_assets) collect the files that idmtools submits to the execution platform (COMPS, SLURM, containers).

Attributes:

Name	Type	Description
eradication_path	str	Path to the Eradication binary. Can also be set via the [emodpy] section of the idmtools config file. When set, the binary is added to the experiment's common assets.
demographics	DemographicsFiles	Experiment-level demographics files shared by all simulations.
migrations	MigrationFiles	Experiment-level migration files shared by all simulations.
reporters	Reporters	Reporter definitions (e.g. InsetChart, EventReporter) that produce custom_reports.json.
climate	ClimateFiles	Experiment-level climate input files (air temperature, rainfall, humidity).
config	dict	EMOD configuration parameters dictionary. Built from schema defaults via emod-api when using from_defaults, or loaded from a JSON file via from_files. Serialized to config.json for each simulation.
config_file_name	str	Filename used when writing the configuration to a simulation asset.
campaign	EMODCampaign	Campaign object holding intervention events. Serialized to campaign.json for each simulation.
simulation_demographics	DemographicsFiles	Simulation-level demographics files (e.g. overlays) that vary per simulation.
simulation_migrations	MigrationFiles	Simulation-level migration files that vary per simulation. Merged with experiment-level migrations in pre_creation.
schema_path	str	Path to the EMOD schema.json file used to build default config and campaign objects.
use_embedded_python	bool	When True, adds --python-script-path to the EMOD command line to enable the embedded Python interpreter within the simulation.
py_path_list	list	List of paths prepended to sys.path in the embedded Python interpreter. Defaults to ["./Assets/python"]; additional paths can be added via add_py_path.
is_linux	bool	Set to True during pre_creation when the target platform is Linux.
implicit_configs	list	Accumulated implicit configuration functions from demographics and campaign builders that are executed during from_defaults to apply side-effect config changes.
sif_filename	str	Filename of the Singularity image (.sif) used on COMPS to create the execution environment.
sif_path		Filesystem path to the Singularity image, used on SLURM/File/Process platforms.

Source code in emodpy/emod_task.py

@dataclass
class EMODTask(ITask):
    """
    Central orchestration class for configuring and running EMOD simulations.

    EMODTask extends `idmtools.ITask` and wires together all the components needed for an EMOD
    experiment: configuration parameters, campaign interventions, demographics, migration files,
    climate data, reporters, and the Eradication binary. It manages both experiment-level assets
    (shared across all simulations) and simulation-level assets (unique per simulation).

    Two factory methods are provided for creating tasks:

    - `from_defaults` — builds everything from schema defaults using user-supplied callback
      functions (config_builder, campaign_builder, demographics_builder, report_builder). This is
      the primary entry point and supports parameter sweeps.
    - `from_files` — loads pre-existing JSON files (config, campaign, demographics, reports)
      directly, for working with hand-authored or legacy configurations.

    During the idmtools lifecycle, `pre_creation` finalizes the configuration by merging
    migrations, wiring reporters into config, setting campaign filenames, and building the command
    line. Asset gathering methods (`gather_common_assets`, `gather_transient_assets`)
    collect the files that idmtools submits to the execution platform (COMPS, SLURM, containers).

    Attributes:
        eradication_path (str): Path to the Eradication binary. Can also be set via the
            `[emodpy]` section of the idmtools config file. When set, the binary is added to
            the experiment's common assets.
        demographics (DemographicsFiles): Experiment-level demographics files shared by all
            simulations.
        migrations (MigrationFiles): Experiment-level migration files shared by all simulations.
        reporters (Reporters): Reporter definitions (e.g. InsetChart, EventReporter) that produce
            custom_reports.json.
        climate (ClimateFiles): Experiment-level climate input files (air temperature, rainfall,
            humidity).
        config (dict): EMOD configuration parameters dictionary. Built from schema defaults via
            `emod-api` when using `from_defaults`, or loaded from a JSON file via `from_files`.
            Serialized to config.json for each simulation.
        config_file_name (str): Filename used when writing the configuration to a simulation asset.
        campaign (EMODCampaign): Campaign object holding intervention events. Serialized to
            campaign.json for each simulation.
        simulation_demographics (DemographicsFiles): Simulation-level demographics files
            (e.g. overlays) that vary per simulation.
        simulation_migrations (MigrationFiles): Simulation-level migration files that vary per
            simulation. Merged with experiment-level migrations in `pre_creation`.
        schema_path (str): Path to the EMOD schema.json file used to build default config and
            campaign objects.
        use_embedded_python (bool): When True, adds `--python-script-path` to the EMOD command
            line to enable the embedded Python interpreter within the simulation.
        py_path_list (list): List of paths prepended to `sys.path` in the embedded Python
            interpreter. Defaults to `["./Assets/python"]`; additional paths can be added via
            `add_py_path`.
        is_linux (bool): Set to True during `pre_creation` when the target platform is Linux.
        implicit_configs (list): Accumulated implicit configuration functions from demographics
            and campaign builders that are executed during `from_defaults` to apply side-effect
            config changes.
        sif_filename (str): Filename of the Singularity image (.sif) used on COMPS to create the
            execution environment.
        sif_path: Filesystem path to the Singularity image, used on SLURM/File/Process platforms.
    """
    eradication_path: str = field(default=None, compare=False, metadata={"md": True})
    demographics: DemographicsFiles = field(default_factory=lambda: DemographicsFiles(''))
    migrations: MigrationFiles = field(default_factory=lambda: MigrationFiles('migrations'))
    reporters: Reporters = field(default_factory=lambda: Reporters())
    climate: ClimateFiles = field(default_factory=lambda: ClimateFiles())
    config: dict = field(default_factory=lambda: {})
    config_file_name: str = "config.json"
    campaign: EMODCampaign = field(default_factory=lambda: EMODCampaign(name="Empty Campaign", events=[], use_defaults=True))
    simulation_demographics: DemographicsFiles = field(default_factory=lambda: DemographicsFiles())
    simulation_migrations: MigrationFiles = field(default_factory=lambda: MigrationFiles())
    schema_path: str = None
    use_embedded_python: bool = False
    py_path_list: list = field(default_factory=lambda: [])
    is_linux: bool = False
    implicit_configs: list = field(default_factory=lambda: [])
    sif_filename: str = None
    sif_path = None

    def __post_init__(self):
        """Initialize derived state after dataclass field assignment.

        Sets the default executable name to "Eradication", adds "./Assets/python" to the
        embedded Python path list, and resolves the Eradication binary path — either from
        the explicitly provided `eradication_path` or from the `[emodpy]` section of the
        idmtools config file.
        """
        super().__post_init__()
        self.executable_name = "Eradication"
        self.py_path_list.append("./Assets/python")
        if self.eradication_path:
            self.executable_name = os.path.basename(self.eradication_path)
            self.eradication_path = os.path.abspath(self.eradication_path)
        else:
            eradication_path = IdmConfigParser().get_option("emodpy", "eradication_path")
            if eradication_path:
                self.eradication_path = eradication_path
                self.executable_name = os.path.basename(self.eradication_path)

    def create_campaign_from_callback(self, builder: Callable, verbose: bool = False, bootstrapped: bool = False) -> None:
        """
        This function is responsible for generating and configuring a campaign using a provided
        builder function. It also handles the custom events that are generated by the campaign by
        adding them to the config.

        Args:
            builder: Function that creates and adds interventions to the campaign module. The function needs to take
                the campaign module as the first required argument, which can then be followed by parameters that you
                want to modify within the interventions. These additional parameters are then available to be modified
                in a sweep. The function must return the campaign module at the end.

                Example::

                    def campaign_builder(campaign, another_param=0.3):
                        from emodpy.campaign.individual_intervention import CommonInterventionParameters, SimpleVaccine, VaccineType
                        from emodpy.campaign.distributor import add_intervention_scheduled
                        import emodpy.campaign.waning_config as waning_config

                        this_waning_config = waning_config.BoxExponential(25, 60, 0.89)
                        common_intervention_parameters = CommonInterventionParameters(cost=0.5,
                                                                                      dont_allow_duplicates=True)
                        vaccine = SimpleVaccine(campaign,
                                                waning_config=this_waning_config,
                                                vaccine_take=another_param,
                                                vaccine_type=VaccineType.TransmissionBlocking,
                                                common_intervention_parameters=common_intervention_parameters)
                        add_intervention_scheduled(campaign, intervention_list=[vaccine], start_day=2)
                        return campaign

            verbose: If True, prints debug information about the generated file.
            bootstrapped: Set to True if the campaign builder will build a campaign from scratch itself. False if it
                will accept an initialized campaign from this function instead and then modify it. Default False.

        Returns:
            None
        """
        if builder is None:
            return

        if bootstrapped:
            campaign = builder()
        else:
            default_campaign = self.build_default_campaign(schema_path=self.schema_path)
            campaign = builder(default_campaign)

        if getattr(campaign, '__name__', None) != 'emod_api.campaign':
            # verify the campaign is an emod_api.campaign module
            raise ValueError("Something went wrong with campaign_builder, "
                             "please make sure that the campaign_builder function returns the campaign module.")

        if "implicits" in dir(campaign) and campaign.implicits:
            self.implicit_configs.extend(campaign.implicits)

        # TODO: this is very bad. This is necessary due to the fact that emod-api campaigns are modules with
        # global module scope, NOT objects! They must be serialize/deserialized to prevent different campaigns
        # from mucking with each other.
        campaign_dict = json.loads(json.dumps(campaign.campaign_dict))
        self.campaign = EMODCampaign.load_from_dict(campaign_dict)
        if dev_mode:
            campaign.save()

        if "Custom_Individual_Events" in self.config.parameters:  # not present in EMOD-Generic
            # adding to the events that might already be there due to user explicitly adding them
            self.config.parameters.Custom_Individual_Events = list(set(
                self.config.parameters.Custom_Individual_Events + campaign.validate_custom_individual_events()))
            self.config.parameters.Custom_Coordinator_Events = list(set(
                self.config.parameters.Custom_Coordinator_Events + campaign.validate_custom_coordinator_events()))
            self.config.parameters.Custom_Node_Events = list(set(
                self.config.parameters.Custom_Node_Events + campaign.validate_custom_node_events()))
        else: # this just runs the validation that the listened to events are also broadcast
            campaign.validate_custom_individual_events()
            campaign.validate_custom_coordinator_events()
            campaign.validate_custom_node_events()

        # This might be a great place to reset the campaign module so users don't have to.
        campaign.reset()

    def create_demographics_from_callback(self, builder: Callable,
                                          from_sweep: bool = False,
                                          verbose: bool = False) -> None:
        """
        Creates a demographics file using a builder function and manages its storage.

        Args:
            builder: A function that generates the demographics object.
            from_sweep (bool): If True, the demographics file is stored in a temporary location.
            verbose (bool): If True, prints debug information about the generated file.

        Returns:
            None
        """
        # If no builder function is provided, exit early.
        if builder is None:
            return

        # Determine the storage path for the demographics file.
        if from_sweep:
            demog_path = tempfile.NamedTemporaryFile(delete=False).name + '.json'
        else:
            # Create a uniquely named directory to prevent conflicts when running multiple scripts.
            timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
            random_string = ''.join(random.choices(string.ascii_letters + string.digits, k=3))
            demog_folder = f"demographics_{timestamp}_{random_string}"
            # Ensure the directory exists.
            os.makedirs(demog_folder, exist_ok=True)
            # Keep the demographics file name consistent with the original name, so that the comps platform can
            # recognize the demographics file and not recreate AssetCollection if the file already exists.
            demog_path = os.path.join(demog_folder, "demographics.json")

        if verbose:
            print(f"Generating demographics file {demog_path}.")

        # Generate and save the demographics file.
        demographics = builder()
        if not demographics or not isinstance(demographics, Demographics):
            raise ValueError("Something went wrong with demographics_builder, "
                             "please make sure that the demographics_builder function returns a Demographics object.")

        demographics.to_file(path=demog_path)

        # Process associated migration files and add them to the asset collection.
        for mig_path in demographics.migration_files:
            if verbose:
                user_logger.info("Adding migration file and json to assets.")
            self.transient_assets.add_asset(str(mig_path))
            self.transient_assets.add_asset(str(mig_path) + ".json")

        # Add the generated demographics file to the appropriate asset collection.
        if from_sweep:
            self.transient_assets.add_asset(demog_path)
        else:
            self.common_assets.add_asset(demog_path)

        # Set the demographics file name for the simulation.
        demog_files = [pathlib.PurePath(demog_path).name]
        demographics.set_demographics_filenames(filenames=demog_files)

        # Apply implicit parameters before the demographics object is destroyed.
        for fn in demographics.implicits:
            if fn:
                self.config = fn(self.config)

    def handle_implicit_configs(self) -> None:
        """
        Execute the implicit config functions created by the demographics builder.

        Returns:
            None
        """
        if dev_mode:
            logger.debug(f"Executing {len(self.implicit_configs)} implicit config functions from demographics and "
                         f"migration.")
        for fn in self.implicit_configs:
            if fn:
                self.config = fn(self.config)

    def _validate_reporter_listening_events(self) -> None:
        if not self.reporters or not self.reporters.schema_path:
            return

        has_custom_events = "Custom_Individual_Events" in self.config.parameters

        builtin = self.reporters.get_builtin_events()

        checks = [
            ("individual", self.reporters.listening_individual_events,
             "Custom_Individual_Events"),
            ("node", self.reporters.listening_node_events,
             "Custom_Node_Events"),
            ("coordinator", self.reporters.listening_coordinator_events,
             "Custom_Coordinator_Events"),
        ]
        for level, listening_events, config_key in checks:
            if not listening_events:
                continue
            builtin_set = set(builtin.get(level, []))
            if has_custom_events:
                custom_set = set(getattr(self.config.parameters, config_key, []))
            else:
                custom_set = set()
            known = builtin_set | custom_set
            unknown = [e for e in listening_events if e not in known]
            if unknown:
                raise ValueError(
                    f"Reporter(s) are listening for {level}-level event(s) that are not broadcast by any of the "
                    f"campaign modules or simulation (built-in): {unknown}. Either remove these events from the "
                    f"reporter or set up your campaign to broadcast these events. Known {level}-level events are: {sorted(known)}")

    @staticmethod
    def build_default_config(schema_path: Union[str, Path]) -> ReadOnlyDict:
        """
        Build the default config object from the schema.

        Args:
            schema_path: Path to the schema file.

        Returns:
            ReadOnlyDict: The default config based on the schema.
        """
        default_config = dfs.get_default_config_from_schema(path_to_schema=schema_path, as_rod=True)
        # needed by emodpy_hiv.country_model.build_config
        default_config["schema_path"] = schema_path
        return default_config

    @staticmethod
    def build_default_campaign(schema_path: Union[str, Path]) -> api_campaign:
        """
        Build the default (empty) campaign and set its schema_path.

        Args:
            schema_path: Path to the schema file.

        Returns:
            Fresh initialized campaign module with schema_path set
        """
        api_campaign.set_schema(schema_path_in=schema_path)
        return api_campaign

    @classmethod
    def from_defaults(cls,
                      schema_path: str,
                      eradication_path: str = None,
                      config_builder: Callable = None,
                      campaign_builder: Callable = None,
                      demographics_builder: Callable = None,
                      report_builder: Callable[[Reporters], Reporters] = None,
                      embedded_python_scripts_path: Union[str, Path, List[Union[str, Path]]] = None,
                      serialized_population_files: Union[str, List[str]] = None,
                      bootstrapped: bool = False) -> "EMODTask":
        """
        Create a task from emod-api defaults and functions to update them.

        Args:
            schema_path: Path to processed schema.json, including the filename.
            eradication_path: Path to Eradication binary, including the filename. You can also add Eradication as an
                asset.
            config_builder: Function that sets parameters for config. The function must have config object as the
                parameter and return the config object. Inside the function, the config object will be modified in
                the following way: config.parameters.<parameter_name> = <value>, with <parameter_name> being an
                attribute.

                Example::

                    def config_builder(config):
                        import emodpy.emod_enum as emod_enum
                        config.parameters.Incubation_Period_Distribution = emod_enum.DistributionType.CONSTANT_DISTRIBUTION
                        config.parameters.Incubation_Period_Constant = 5
                        config.parameters.Infectious_Period_Distribution = emod_enum.DistributionType.CONSTANT_DISTRIBUTION
                        config.parameters.Infectious_Period_Constant = 5
                        config.parameters.Simulation_Duration = 80
                        return config

            campaign_builder: Function that creates and adds interventions to the campaign module. The function needs
                to take the campaign module as the first required argument, which can then be followed by parameters
                that you  want to modify within the interventions. These additional parameters are then available to
                be modified in a sweep. The function must return the campaign module at the end.

                Example::

                    def campaign_builder(campaign, additional_param=0.3):
                        from emodpy.campaign.individual_intervention import CommonInterventionParameters, SimpleVaccine, VaccineType
                        from emodpy.campaign.distributor import add_intervention_scheduled
                        import emodpy.campaign.waning_config as waning_config

                        this_waning_config = waning_config.BoxExponential(box_duration=25,
                                                                          decay_time_constant=60,
                                                                          initial_effect=0.89)
                        common_intervention_parameters = CommonInterventionParameters(cost=0.5,
                                                                                      dont_allow_duplicates=True)
                        vaccine = SimpleVaccine(campaign,
                                                waning_config=this_waning_config,
                                                vaccine_take=additional_param,
                                                vaccine_type=VaccineType.TransmissionBlocking,
                                                common_intervention_parameters=common_intervention_parameters)
                        add_intervention_scheduled(campaign, intervention_list=[vaccine], start_day=2)
                        return campaign



            demographics_builder: Function that builds the demographics configuration and optional migration
                configuration.

                Example::

                    def demographics_builder():
                        import emodpy.demographics.Demographics as Demographics

                        demographics_object = Demographics.from_template_node(pop=500)
                        # other code setting demographics parameters
                        return demographics_object

            report_builder: Function that creates reports to be used in the experiment. The function must have Reporters
                object as the parameter and return that object. It is assumed that all the reporters come from the
                reporters that are part of EMOD main code. EMOD also supports reporters as custom plug-in dlls,
                however, not through the current emodpy system.

                Example::

                        def report_builder(reporters):
                            from emodpy.reporters.reporters import ReportEventCounter, ReportFilter

                            report_filter = ReportFilter(start_day=0, end_day=365, filename_suffix="LifeEvents")
                            reporters.add(ReportEventCounter(reporters_object=reporters,
                                               report_filter=report_filter,
                                               event_list=["DiseaseDeaths", "Births", "HappyBirthday"]))
                            return reporters


            embedded_python_scripts_path: Path to folder with python scripts or a list of paths to specific python
                scripts. When path to folder, all python scripts in the folder will be added to the experiment.
                embedded_python_scripts_path may also be a list of such items, each added independently.
                Note: We no longer support passing in a function for this parameter.
            serialized_population_files: Path to folder containing .dtk serialized population files or specific
                .dtk file, including the filename. All .dtk files in the folder will be added and used in the
                experiment.
            bootstrapped: Set to True if the campaign builder will build a campaign from scratch itself. False if it
                will accept an initialized campaign from this function instead and then modify it. Default False.

        Returns:
            EMODTask
        """
        task = cls(eradication_path=eradication_path, schema_path=schema_path)
        # We do not regenerate the schema from the Eradication binary because we can't guarantee this code is running
        # on a matching platform, so we use a schema file.
        default_config = cls.build_default_config(schema_path=task.schema_path)
        task.available_config_parameters = list(default_config['parameters'].keys())

        task.config = dfs.get_config_from_default_and_params(config=default_config, set_fn=config_builder)
        if not isinstance(task.config, ReadOnlyDict):
            raise ValueError("Something went wrong with config_builder, please make sure "
                             "the config_builder function returns a config object.")

        # Let's do the demographics building here...
        if demographics_builder:
            task.create_demographics_from_callback(demographics_builder)

        if campaign_builder:
            task.create_campaign_from_callback(builder=campaign_builder, bootstrapped=bootstrapped)

        if embedded_python_scripts_path:
            task.add_embedded_python_scripts_from_path(path=embedded_python_scripts_path)

        if report_builder:
            task.reporters = Reporters(schema_path=task.schema_path)
            returned = report_builder(task.reporters)
            if not returned or not isinstance(returned, Reporters):
                raise ValueError("Something went wrong with report_builder, please make sure "
                                 "the report_builder function returns a Reporters object.")
            task._validate_reporter_listening_events()

        if serialized_population_files:
            task.add_serialized_population_files_from_path(serialized_population_files)

        task.handle_implicit_configs()

        return task

    @classmethod
    def from_files(cls,
                   eradication_path: str = None,
                   config_path: str = None,
                   campaign_path: str = None,
                   demographics_paths: Union[str, list] = None,
                   custom_reports_path: str = None,
                   embedded_python_scripts_path: Union[str, list[str]] = None,
                   serialized_population_files: Union[str, list[str]] = None,
                   asset_path: str = None) -> "EMODTask":
        """
        Load custom EMOD files when creating [EMODTask][].

        Args:
            eradication_path: Path to Eradication binary, including the filename.
            config_path: Path to the configuration file, including the filename.
            campaign_path: Path to the campaign file, including the filename.
            demographics_paths: Path or a list of paths to the folder containing demographics files or path to a
                specific demographics file, including the filename. When path to folder, all .json files in the folder
                will be added to the experiment as demographics files.
            custom_reports_path: Path to the custom reports file, including the filename. It is assumed that all the
                reporters in the file come from the reporters that are part of EMOD main code. EMOD also supports
                reporters as custom plug-in dlls, however, not through the current emodpy system.
            embedded_python_scripts_path: Path to folder with python scripts or path to a specific python script,
                including the filename. When path to folder, all python scripts in the folder will be added to the
                experiment.
                Note: We no longer support passing in a function for this parameter.
            serialized_population_files: Path to folder containing .dtk serialized population files or specific
                .dtk file, including the filename. All .dtk files in the folder will be added and used in the
                experiment.
            asset_path: Path to migration and/or climate files.

        Returns:
            An initialized experiment
        """
        # Create the experiment
        task = cls(eradication_path=eradication_path)

        task.config = load_json_file(config_path)["parameters"]

        if campaign_path:
            task.campaign = EMODCampaign.load_from_file(campaign_path)

        if demographics_paths:
            if isinstance(demographics_paths, list):
                for path in demographics_paths:
                    task.demographics.add_demographics_from_files(path)
            else:
                task.demographics.add_demographics_from_files(demographics_paths)
            task.demographics.set_task_config(task)  # adds the demographics to the config

        if custom_reports_path:
            custom_reports_path = os.path.abspath(custom_reports_path)
            task.transient_assets.add_asset(custom_reports_path)
            # need to do this explicitly because this is usually added by the Reporters object,
            # but we are not using it here.
            custom_reports_filename = os.path.basename(custom_reports_path)
            task.config["Custom_Reports_Filename"] = custom_reports_filename

        if embedded_python_scripts_path is not None:
            task.add_embedded_python_scripts_from_path(path=embedded_python_scripts_path)

        if serialized_population_files is not None:
            task.add_serialized_population_files_from_path(serialized_population_files)

        if asset_path and config_path:
            # Look for climate
            task.climate.read_config_file(config_path, asset_path)
            # Look for migrations
            task.migrations.read_config_file(config_path, asset_path)

        return task

    def pre_creation(self, parent: Union[Simulation, IWorkflowItem], platform: 'IPlatform'):
        """
        Call before a task is executed. This ensures our configuration is properly done

        """
        # Set the demographics
        # self.demographics.set_task_config(self)
        # self.simulation_demographics.set_task_config(self, extend=True)

        # Set the migrations
        self.simulation_migrations.merge_with(self.migrations)
        # self.simulation_migrations.set_task_config(self)

        # Set the climate
        # self.climate.set_task_config(self)

        # Set the reporters
        # this only runs for when using from_defaults,
        # because with "from_files" we bypass Reporters object creation
        if self.reporters.builtin_reporters:
            self.config.parameters.Custom_Reports_Filename = "custom_reports.json"
        for reporter in self.reporters.config_reporters:
            for i in reporter.parameters:
                setattr(self.config.parameters, i, reporter.parameters[i])

        # Set the campaign filename
        if self.campaign:
            if type(self.config) is dict:  # when "from_files" was used
                self.config["Campaign_Filename"] = "campaign.json"
                self.config["Enable_Interventions"] = 1  # implicit?
            else:    # when using from_defaults
                self.config.parameters.Campaign_Filename = "campaign.json"
                self.config.parameters.Enable_Interventions = 1  # implicit

        # Gather the custom coordinator, individual, and node events
        self.set_command_line()
        super().pre_creation(parent, platform)
        if not platform.is_windows_platform():
            # print( "Target is LINUX!" )
            self.is_linux = True

    def set_command_line(self) -> None:
        """
        Build and set the command line object.

        Returns:

        """
        # In COMPS, it's rare to have to specify multiple paths because 'we' control the environment and
        # can put everything in Assets. The multiple input paths is useful for local command-line usage where
        # the input files are spread across different locations. Note that with symlinks it's trivial to put
        # all the files in one path without copying. The only exception here is when we are using dtk_pre_process
        # to create a (demographics) input file and this can not be in Assets.
        # input_path = "./Assets" # this works on windows

        # Both "./Assets\;." and "./Assets\\;." work but the former confuses the linter because it is expecting
        # a known escape code, e.g. "\n" - escaping the backslash with "\\" escapes the escape code (got that?).
        input_path = "./Assets\\;."

        # Create the command line according to self. location of the model
        if self.sif_filename:
            self.command = CommandLine(
                "singularity",
                "exec",
                f"Assets/{self.sif_filename}",
                f"Assets/{self.executable_name}",
                "--config",
                f"{self.config_file_name}",
                "--dll-path",   # for Generic-Ongoing Eradication
                "./Assets",
            )
        else:
            self.command = CommandLine(f"Assets/{self.executable_name}", "--config",
                                       f"{self.config_file_name}", "--dll-path", "./Assets")

        if self.use_embedded_python:
            list_sep = ";"
            self.command._options.update({"--python-script-path": list_sep.join(self.py_path_list)})

        # We do this here because CommandLine tries to be smart and quote input_path, but it isn't quite right...
        self.command.add_raw_argument("--input-path")
        self.command.add_raw_argument(input_path)

    def add_py_path(self, path_to_add) -> None:
        """
        Add path to list of python paths prepended to sys.path in embedded interpreter

        Returns:

        """
        self.py_path_list.append(path_to_add)

    def set_sif(self, path_to_sif: Union[Path, str], platform: IPlatform) -> None:
        """
        Set the singularity image file to use for creating the EMOD execution environment.

        Args:
            path_to_sif:

                non-COMPSPlatforms: The file system path to the sif file for creating the EMOD execution environment.
                COMPSPlatform: Either:

                    a) The same sif file path as with other platforms
                    b) A .id file containing a COMPS AssetCollection id containing the desired sif file, formatted e.g.::

                        8df53802-53f3-ec11-a9f9-b88303911bc1::Asset Collection

                ContainerPlatform: No sif file specification is used.

            platform: Platform object to use for this task.

        Returns:
            None
        """
        platform_type = platform.__class__.__name__
        path_to_sif = Path(path_to_sif)
        suffix = path_to_sif.suffix
        is_sif_file = path_to_sif.suffix == '.sif'
        is_id_file = path_to_sif.suffix == '.id'

        # path_to_sif = str(path_to_sif)  # code below is written to require strings
        platforms = ['SlurmPlatform', 'FilePlatform', 'ProcessPlatform']
        if platform_type in platforms:
            # only .sif is valid for these platforms
            if is_id_file:
                platforms_str = ', '.join(platforms)
                raise ValueError(f"Cannot use a .id file for sif specification on these platforms: {platforms_str}")
            elif is_sif_file:
                self.sif_path = path_to_sif
            else:
                raise ValueError(f"Unknown sif file with extension: {suffix}")
        elif platform_type == 'COMPSPlatform':
            # .sif and .id files are valid for COMPSPlatform
            if is_id_file:
                ac = AssetCollection.from_id_file(path_to_sif)
                sif_asset = [acf for acf in ac.assets if Path(acf.filename).suffix == ".sif"][0]
                self.sif_filename = sif_asset.filename
            elif is_sif_file:
                sif_asset = path_to_sif
                self.sif_filename = path_to_sif.name
            else:
                raise ValueError(f"Unknown sif file with extension: {suffix}")
            self.common_assets.add_asset(sif_asset)
        elif platform_type == 'ContainerPlatform':
            warnings.warn('.sif or .id file specified for platform of type ContainerPlatform . '
                          'This is unnecessary and is being ignored.', RuntimeWarning)
        else:
            raise ValueError(f"Unknown platform type for setting sif file: {platform_type}")

    def gather_common_assets(self) -> AssetCollection:
        """
        Gather Experiment Level Assets
        Returns:

        """
        # check whether there are any .sif or .img files in the common assets diretories...
        # Add Eradication.exe to assets
        if self.eradication_path:
            self.common_assets.add_asset(Asset(absolute_path=self.eradication_path,
                                               filename=self.executable_name),
                                         fail_on_duplicate=False)

        # Add demographics to assets
        if self.demographics.assets:
            self.common_assets.extend(self.demographics.gather_assets())

        # Add the migrations
        if self.migrations.assets:
            self.common_assets.extend(self.migrations.gather_assets())

        # Add the climate
        if self.climate.assets:
            self.common_assets.extend(self.climate.gather_assets())

        return self.common_assets

    def _enforce_non_schema_coherence(self) -> None:
        """
        This function enforces business logic that can't be encoded in the schema.
        Rules:
            Start_Time + Simulation_Duration >= Minimum_End_Time

        Returns:
            None
        """
        if "Miminum_End_Time" in self.config:  # only present when we enable Enable_Termination_On_Zero_Total_Infectivity
            if (self.config['Start_Time'] + self.config['Simulation_Duration']) < self.config['Minimum_End_Time']:
                raise ValueError(f"{self.config['Start_Time']} + {self.config['Simulation_Duration']} "
                                 f"(Start_Time + Simulation_Duration) < "
                                 f"{self.config['Minimum_End_Time']} (Minimum_End_Time)")

    def gather_transient_assets(self) -> AssetCollection:
        """
        Gather assets that are per simulation
        Returns:
            AssetCollection
        """

        # This config code needs to be rewritten
        # task.config contains emod-api version of config i.e., with schema. Needs to be finalized and written.
        if logger.isEnabledFor(DEBUG) and dev_mode:
            logger.debug("DEBUG: Calling finalize.")

        # Add config and campaign to assets as needed

        if self.config:
            if type(self.config) is dict:  # old/basic style, when "from_files" is used
                self.config = {"parameters": self.config}
            else:
                self.config.parameters.finalize()
            self._enforce_non_schema_coherence()
            if dev_mode:
                with open(self.config_file_name, "w") as fp:
                    json.dump(self.config, fp, sort_keys=True, indent=4)
            asset = Asset(filename=self.config_file_name, content=json.dumps(self.config, sort_keys=True))
            self.transient_assets.add_asset(asset=asset, fail_on_duplicate=False)

        if self.campaign:
            asset = Asset(filename="campaign.json", content=self.campaign.json)
            self.transient_assets.add_asset(asset=asset, fail_on_duplicate=False)

            if dev_mode:
                import emod_api.peek_camp as base_peek_camp
                original = sys.stdout
                sys.stdout = open('campaign.ccdl', 'w')
                base_peek_camp.decode("campaign.json", self.config_file_name)
                sys.stdout = original
                data = open("campaign.ccdl").readlines()
                data.sort()
                for i in range(len(data)):
                    print(data[i].strip())

        # Add custom_reporters.json if needed
        if self.reporters.builtin_reporters:
            asset = Asset(filename="custom_reports.json", content=self.reporters.json)
            self.transient_assets.add_asset(asset=asset, fail_on_duplicate=False)

        # Add demographics files to assets
        if self.simulation_demographics.assets:
            self.transient_assets.extend(self.simulation_demographics.gather_assets())

        # Add the migrations
        if self.simulation_migrations.assets:
            self.transient_assets.extend(self.simulation_migrations.gather_assets())

        return self.transient_assets

    def copy_simulation(self, base_simulation: 'Simulation') -> 'Simulation':
        """
        Called when making copies of a simulation. We deep copy parts of the simulation to ensure we don't
        accidentally share objects between simulations.

        Args:
            base_simulation: Base Simulation

        Returns:
            New Simulation
        """
        simulation = copy.deepcopy(base_simulation)

        # Copy the experiment demographics and set them as persisted to prevent change
        demog_copy = copy.deepcopy(self.demographics)
        demog_copy.set_all_persisted()
        simulation.task.demographics.extend(demog_copy)

        # Copy the climate
        climate_copy = copy.deepcopy(self.climate)
        climate_copy.set_all_persisted()
        simulation.task.climate = climate_copy

        # Tale care of the migrations
        migration_copy = copy.deepcopy(self.migrations)
        migration_copy.set_all_persisted()
        simulation.task.simulation_migrations.merge_with(migration_copy)

        # Handle the custom reporters
        reporters_copy = copy.deepcopy(self.reporters)
        reporters_copy.set_all_persisted()
        simulation.task.reporters = reporters_copy

        return simulation

    def set_parameter(self, name: str, value: any) -> dict:
        """
        Set a value in the EMOD config.json file. This will be deprecated in the future in favour of emod_api.config.

        Args:
            name: Name of parameter to set
            value: Value to set

        Returns:
            Tags to set
        """
        if not self.config:
            raise ValueError("task.config is empty. Please load a config file or dictionary"
                             " or create a new one using from_defaults.")
        if type(self.config) is dict:  # old style, when "from_files" is used
            self.config[name] = value  # no checks against the schema
        elif hasattr(self.config.parameters, name):
            setattr(self.config.parameters, name, value)
        else:
            raise ValueError(f"Parameter '{name}' not a valid parameter based on schema.")

        return {name: value}

    @staticmethod
    def set_parameter_sweep_callback(simulation: Simulation, param: str, value: Any) -> Dict[str, Any]:
        """
        Convenience callback for sweeps

        Args:
            simulation: Simulation we are updating
            param: Parameter
            value: Value

        Returns:
            Tags to set on simulation
        """
        if not hasattr(simulation.task, 'set_parameter'):
            raise ValueError("set_parameter_sweep_callback can only be used on tasks with a set_parameter")
        return simulation.task.set_parameter(param, value)

    @classmethod
    def set_parameter_partial(cls, parameter: str) -> partial[Simulation]:
        """
        Convenience callback for sweeps

        Args:
            parameter: Parameter to set

        Returns:
            Partial function
        """
        return partial(cls.set_parameter_sweep_callback, param=parameter)

    def get_parameter(self, name: str, default: Optional[Any] = None) -> Any:
        """
        Get a parameter in the simulation.

        Args:
            name: The name of the parameter.
            default: Optional, the default value.

        Returns:
            The value of the parameter.
        """
        return self.config.get(name, default)

    @staticmethod
    def _add_files_from_path(path: str, valid_extension: str, process_file_callback: Callable) -> List[str]:
        """
        Generalized function to add files from a path (file or directory) based on a valid extension.

        Args:
            path (str): Path to a file or directory.
            valid_extension (str): Extension of the files to process (e.g., ".dtk" or ".py").
            process_file_callback (Callable[[str], None]): Function to handle valid files.

        Returns:
            List[str]: List of filenames added.
        """
        path = os.path.abspath(path)

        files_added = []
        if os.path.isfile(path):
            if path.endswith(valid_extension):
                files_added.append(os.path.basename(path))
                process_file_callback(path)
            else:
                raise ValueError(f"File {path} is not a {valid_extension} file.")
        elif os.path.isdir(path):  # it's a directory
            for entry_name in os.listdir(path):
                full_path = os.path.join(path, entry_name)
                if os.path.isfile(full_path) and entry_name.endswith(valid_extension):
                    files_added.append(entry_name)
                    process_file_callback(full_path)
            if not files_added:
                raise ValueError(f"No {valid_extension} files found in {path}")
        else:
            raise ValueError(f"Path {path} is not a file or a folder.")

        return files_added

    def add_embedded_python_scripts_from_path(self, path: Union[str, Path, List[Union[str, Path]]]) -> None:
        """
        Adds embedded python scripts from the path to the common assets

        Args:
            path: Relative or absolute path to the file (including the file name) or to a folder containing
                python scripts. Please note, all the python scripts in a specified folder will be added to the
                simulation. path may also be a list of such items, each added independently.

        """
        def process_valid_python_files(file_path: str):
            python_file_asset = Asset(file_path, relative_path="python")
            # if adding python script again, maybe we want to overwrite the other one
            self.common_assets.add_asset(python_file_asset,
                                         fail_on_duplicate=False,
                                         fail_on_deep_comparison=False)

        path = path if isinstance(path, list) else [path]
        for p in path:
            self._add_files_from_path(p, ".py", process_valid_python_files)
        self.use_embedded_python = True

    def add_serialized_population_files_from_path(self, path: str) -> None:
        """
        Adds serialized population files from the path to the common assets

        Args:
            path: Relative or absolute path to the file (including the file name) or to the folder containing
                the .dtk files. Please note, all the .dtk files in the folder will be added and be used in the
                simulations.
        """
        def process_dtk_file(file_path: str):
            self.common_assets.add_asset(file_path)

        files_added = self._add_files_from_path(path, ".dtk", process_dtk_file)
        # Set serialized population path
        if isinstance(self.config, dict):
            self.config["Serialized_Population_Filenames"] = files_added
            self.config["Serialized_Population_Path"] = "Assets"
        else:
            self.config.parameters.Serialized_Population_Filenames = files_added
            self.config.parameters.Serialized_Population_Path = "Assets"

    def reload_from_simulation(self, simulation: 'Simulation'):
        pass

__post_init__()

Initialize derived state after dataclass field assignment.

Sets the default executable name to "Eradication", adds "./Assets/python" to the embedded Python path list, and resolves the Eradication binary path — either from the explicitly provided eradication_path or from the [emodpy] section of the idmtools config file.

Source code in emodpy/emod_task.py

def __post_init__(self):
    """Initialize derived state after dataclass field assignment.

    Sets the default executable name to "Eradication", adds "./Assets/python" to the
    embedded Python path list, and resolves the Eradication binary path — either from
    the explicitly provided `eradication_path` or from the `[emodpy]` section of the
    idmtools config file.
    """
    super().__post_init__()
    self.executable_name = "Eradication"
    self.py_path_list.append("./Assets/python")
    if self.eradication_path:
        self.executable_name = os.path.basename(self.eradication_path)
        self.eradication_path = os.path.abspath(self.eradication_path)
    else:
        eradication_path = IdmConfigParser().get_option("emodpy", "eradication_path")
        if eradication_path:
            self.eradication_path = eradication_path
            self.executable_name = os.path.basename(self.eradication_path)

add_embedded_python_scripts_from_path(path)

Adds embedded python scripts from the path to the common assets

Parameters:

Name	Type	Description	Default
path	Union[str, Path, List[Union[str, Path]]]	Relative or absolute path to the file (including the file name) or to a folder containing python scripts. Please note, all the python scripts in a specified folder will be added to the simulation. path may also be a list of such items, each added independently.	required

Source code in emodpy/emod_task.py

def add_embedded_python_scripts_from_path(self, path: Union[str, Path, List[Union[str, Path]]]) -> None:
    """
    Adds embedded python scripts from the path to the common assets

    Args:
        path: Relative or absolute path to the file (including the file name) or to a folder containing
            python scripts. Please note, all the python scripts in a specified folder will be added to the
            simulation. path may also be a list of such items, each added independently.

    """
    def process_valid_python_files(file_path: str):
        python_file_asset = Asset(file_path, relative_path="python")
        # if adding python script again, maybe we want to overwrite the other one
        self.common_assets.add_asset(python_file_asset,
                                     fail_on_duplicate=False,
                                     fail_on_deep_comparison=False)

    path = path if isinstance(path, list) else [path]
    for p in path:
        self._add_files_from_path(p, ".py", process_valid_python_files)
    self.use_embedded_python = True

add_py_path(path_to_add)

Add path to list of python paths prepended to sys.path in embedded interpreter

Returns:

Source code in emodpy/emod_task.py

def add_py_path(self, path_to_add) -> None:
    """
    Add path to list of python paths prepended to sys.path in embedded interpreter

    Returns:

    """
    self.py_path_list.append(path_to_add)

add_serialized_population_files_from_path(path)

Adds serialized population files from the path to the common assets

Parameters:

Name	Type	Description	Default
path	str	Relative or absolute path to the file (including the file name) or to the folder containing the .dtk files. Please note, all the .dtk files in the folder will be added and be used in the simulations.	required

Source code in emodpy/emod_task.py

def add_serialized_population_files_from_path(self, path: str) -> None:
    """
    Adds serialized population files from the path to the common assets

    Args:
        path: Relative or absolute path to the file (including the file name) or to the folder containing
            the .dtk files. Please note, all the .dtk files in the folder will be added and be used in the
            simulations.
    """
    def process_dtk_file(file_path: str):
        self.common_assets.add_asset(file_path)

    files_added = self._add_files_from_path(path, ".dtk", process_dtk_file)
    # Set serialized population path
    if isinstance(self.config, dict):
        self.config["Serialized_Population_Filenames"] = files_added
        self.config["Serialized_Population_Path"] = "Assets"
    else:
        self.config.parameters.Serialized_Population_Filenames = files_added
        self.config.parameters.Serialized_Population_Path = "Assets"

build_default_campaign(schema_path) staticmethod

Build the default (empty) campaign and set its schema_path.

Parameters:

Name	Type	Description	Default
schema_path	Union[str, Path]	Path to the schema file.	required

Returns:

Type	Description
campaign	Fresh initialized campaign module with schema_path set

Source code in emodpy/emod_task.py

@staticmethod
def build_default_campaign(schema_path: Union[str, Path]) -> api_campaign:
    """
    Build the default (empty) campaign and set its schema_path.

    Args:
        schema_path: Path to the schema file.

    Returns:
        Fresh initialized campaign module with schema_path set
    """
    api_campaign.set_schema(schema_path_in=schema_path)
    return api_campaign

build_default_config(schema_path) staticmethod

Build the default config object from the schema.

Parameters:

Name	Type	Description	Default
schema_path	Union[str, Path]	Path to the schema file.	required

Returns:

Name	Type	Description
ReadOnlyDict	ReadOnlyDict	The default config based on the schema.

Source code in emodpy/emod_task.py

@staticmethod
def build_default_config(schema_path: Union[str, Path]) -> ReadOnlyDict:
    """
    Build the default config object from the schema.

    Args:
        schema_path: Path to the schema file.

    Returns:
        ReadOnlyDict: The default config based on the schema.
    """
    default_config = dfs.get_default_config_from_schema(path_to_schema=schema_path, as_rod=True)
    # needed by emodpy_hiv.country_model.build_config
    default_config["schema_path"] = schema_path
    return default_config

copy_simulation(base_simulation)

Called when making copies of a simulation. We deep copy parts of the simulation to ensure we don't accidentally share objects between simulations.

Parameters:

Name	Type	Description	Default
base_simulation	Simulation	Base Simulation	required

Returns:

Type	Description
Simulation	New Simulation

Source code in emodpy/emod_task.py

def copy_simulation(self, base_simulation: 'Simulation') -> 'Simulation':
    """
    Called when making copies of a simulation. We deep copy parts of the simulation to ensure we don't
    accidentally share objects between simulations.

    Args:
        base_simulation: Base Simulation

    Returns:
        New Simulation
    """
    simulation = copy.deepcopy(base_simulation)

    # Copy the experiment demographics and set them as persisted to prevent change
    demog_copy = copy.deepcopy(self.demographics)
    demog_copy.set_all_persisted()
    simulation.task.demographics.extend(demog_copy)

    # Copy the climate
    climate_copy = copy.deepcopy(self.climate)
    climate_copy.set_all_persisted()
    simulation.task.climate = climate_copy

    # Tale care of the migrations
    migration_copy = copy.deepcopy(self.migrations)
    migration_copy.set_all_persisted()
    simulation.task.simulation_migrations.merge_with(migration_copy)

    # Handle the custom reporters
    reporters_copy = copy.deepcopy(self.reporters)
    reporters_copy.set_all_persisted()
    simulation.task.reporters = reporters_copy

    return simulation

create_campaign_from_callback(builder, verbose=False, bootstrapped=False)

This function is responsible for generating and configuring a campaign using a provided builder function. It also handles the custom events that are generated by the campaign by adding them to the config.

Parameters:

Name	Type	Description	Default
builder	Callable	Function that creates and adds interventions to the campaign module. The function needs to take the campaign module as the first required argument, which can then be followed by parameters that you want to modify within the interventions. These additional parameters are then available to be modified in a sweep. The function must return the campaign module at the end. Example:: def campaign_builder(campaign, another_param=0.3): from emodpy.campaign.individual_intervention import CommonInterventionParameters, SimpleVaccine, VaccineType from emodpy.campaign.distributor import add_intervention_scheduled import emodpy.campaign.waning_config as waning_config this_waning_config = waning_config.BoxExponential(25, 60, 0.89) common_intervention_parameters = CommonInterventionParameters(cost=0.5, dont_allow_duplicates=True) vaccine = SimpleVaccine(campaign, waning_config=this_waning_config, vaccine_take=another_param, vaccine_type=VaccineType.TransmissionBlocking, common_intervention_parameters=common_intervention_parameters) add_intervention_scheduled(campaign, intervention_list=[vaccine], start_day=2) return campaign	required
verbose	bool	If True, prints debug information about the generated file.	False
bootstrapped	bool	Set to True if the campaign builder will build a campaign from scratch itself. False if it will accept an initialized campaign from this function instead and then modify it. Default False.	False

Returns:

Type	Description
None	None

Source code in emodpy/emod_task.py

def create_campaign_from_callback(self, builder: Callable, verbose: bool = False, bootstrapped: bool = False) -> None:
    """
    This function is responsible for generating and configuring a campaign using a provided
    builder function. It also handles the custom events that are generated by the campaign by
    adding them to the config.

    Args:
        builder: Function that creates and adds interventions to the campaign module. The function needs to take
            the campaign module as the first required argument, which can then be followed by parameters that you
            want to modify within the interventions. These additional parameters are then available to be modified
            in a sweep. The function must return the campaign module at the end.

            Example::

                def campaign_builder(campaign, another_param=0.3):
                    from emodpy.campaign.individual_intervention import CommonInterventionParameters, SimpleVaccine, VaccineType
                    from emodpy.campaign.distributor import add_intervention_scheduled
                    import emodpy.campaign.waning_config as waning_config

                    this_waning_config = waning_config.BoxExponential(25, 60, 0.89)
                    common_intervention_parameters = CommonInterventionParameters(cost=0.5,
                                                                                  dont_allow_duplicates=True)
                    vaccine = SimpleVaccine(campaign,
                                            waning_config=this_waning_config,
                                            vaccine_take=another_param,
                                            vaccine_type=VaccineType.TransmissionBlocking,
                                            common_intervention_parameters=common_intervention_parameters)
                    add_intervention_scheduled(campaign, intervention_list=[vaccine], start_day=2)
                    return campaign

        verbose: If True, prints debug information about the generated file.
        bootstrapped: Set to True if the campaign builder will build a campaign from scratch itself. False if it
            will accept an initialized campaign from this function instead and then modify it. Default False.

    Returns:
        None
    """
    if builder is None:
        return

    if bootstrapped:
        campaign = builder()
    else:
        default_campaign = self.build_default_campaign(schema_path=self.schema_path)
        campaign = builder(default_campaign)

    if getattr(campaign, '__name__', None) != 'emod_api.campaign':
        # verify the campaign is an emod_api.campaign module
        raise ValueError("Something went wrong with campaign_builder, "
                         "please make sure that the campaign_builder function returns the campaign module.")

    if "implicits" in dir(campaign) and campaign.implicits:
        self.implicit_configs.extend(campaign.implicits)

    # TODO: this is very bad. This is necessary due to the fact that emod-api campaigns are modules with
    # global module scope, NOT objects! They must be serialize/deserialized to prevent different campaigns
    # from mucking with each other.
    campaign_dict = json.loads(json.dumps(campaign.campaign_dict))
    self.campaign = EMODCampaign.load_from_dict(campaign_dict)
    if dev_mode:
        campaign.save()

    if "Custom_Individual_Events" in self.config.parameters:  # not present in EMOD-Generic
        # adding to the events that might already be there due to user explicitly adding them
        self.config.parameters.Custom_Individual_Events = list(set(
            self.config.parameters.Custom_Individual_Events + campaign.validate_custom_individual_events()))
        self.config.parameters.Custom_Coordinator_Events = list(set(
            self.config.parameters.Custom_Coordinator_Events + campaign.validate_custom_coordinator_events()))
        self.config.parameters.Custom_Node_Events = list(set(
            self.config.parameters.Custom_Node_Events + campaign.validate_custom_node_events()))
    else: # this just runs the validation that the listened to events are also broadcast
        campaign.validate_custom_individual_events()
        campaign.validate_custom_coordinator_events()
        campaign.validate_custom_node_events()

    # This might be a great place to reset the campaign module so users don't have to.
    campaign.reset()

create_demographics_from_callback(builder, from_sweep=False, verbose=False)

Creates a demographics file using a builder function and manages its storage.

Parameters:

Name	Type	Description	Default
builder	Callable	A function that generates the demographics object.	required
from_sweep	bool	If True, the demographics file is stored in a temporary location.	False
verbose	bool	If True, prints debug information about the generated file.	False

Returns:

Type	Description
None	None

Source code in emodpy/emod_task.py

def create_demographics_from_callback(self, builder: Callable,
                                      from_sweep: bool = False,
                                      verbose: bool = False) -> None:
    """
    Creates a demographics file using a builder function and manages its storage.

    Args:
        builder: A function that generates the demographics object.
        from_sweep (bool): If True, the demographics file is stored in a temporary location.
        verbose (bool): If True, prints debug information about the generated file.

    Returns:
        None
    """
    # If no builder function is provided, exit early.
    if builder is None:
        return

    # Determine the storage path for the demographics file.
    if from_sweep:
        demog_path = tempfile.NamedTemporaryFile(delete=False).name + '.json'
    else:
        # Create a uniquely named directory to prevent conflicts when running multiple scripts.
        timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
        random_string = ''.join(random.choices(string.ascii_letters + string.digits, k=3))
        demog_folder = f"demographics_{timestamp}_{random_string}"
        # Ensure the directory exists.
        os.makedirs(demog_folder, exist_ok=True)
        # Keep the demographics file name consistent with the original name, so that the comps platform can
        # recognize the demographics file and not recreate AssetCollection if the file already exists.
        demog_path = os.path.join(demog_folder, "demographics.json")

    if verbose:
        print(f"Generating demographics file {demog_path}.")

    # Generate and save the demographics file.
    demographics = builder()
    if not demographics or not isinstance(demographics, Demographics):
        raise ValueError("Something went wrong with demographics_builder, "
                         "please make sure that the demographics_builder function returns a Demographics object.")

    demographics.to_file(path=demog_path)

    # Process associated migration files and add them to the asset collection.
    for mig_path in demographics.migration_files:
        if verbose:
            user_logger.info("Adding migration file and json to assets.")
        self.transient_assets.add_asset(str(mig_path))
        self.transient_assets.add_asset(str(mig_path) + ".json")

    # Add the generated demographics file to the appropriate asset collection.
    if from_sweep:
        self.transient_assets.add_asset(demog_path)
    else:
        self.common_assets.add_asset(demog_path)

    # Set the demographics file name for the simulation.
    demog_files = [pathlib.PurePath(demog_path).name]
    demographics.set_demographics_filenames(filenames=demog_files)

    # Apply implicit parameters before the demographics object is destroyed.
    for fn in demographics.implicits:
        if fn:
            self.config = fn(self.config)

from_defaults(schema_path, eradication_path=None, config_builder=None, campaign_builder=None, demographics_builder=None, report_builder=None, embedded_python_scripts_path=None, serialized_population_files=None, bootstrapped=False) classmethod

Create a task from emod-api defaults and functions to update them.

Parameters:

Name	Type	Description	Default
schema_path	str	Path to processed schema.json, including the filename.	required
eradication_path	str	Path to Eradication binary, including the filename. You can also add Eradication as an asset.	None
config_builder	Callable	Function that sets parameters for config. The function must have config object as the parameter and return the config object. Inside the function, the config object will be modified in the following way: config.parameters. = , with being an attribute. Example:: def config_builder(config): import emodpy.emod_enum as emod_enum config.parameters.Incubation_Period_Distribution = emod_enum.DistributionType.CONSTANT_DISTRIBUTION config.parameters.Incubation_Period_Constant = 5 config.parameters.Infectious_Period_Distribution = emod_enum.DistributionType.CONSTANT_DISTRIBUTION config.parameters.Infectious_Period_Constant = 5 config.parameters.Simulation_Duration = 80 return config	None
campaign_builder	Callable	Function that creates and adds interventions to the campaign module. The function needs to take the campaign module as the first required argument, which can then be followed by parameters that you want to modify within the interventions. These additional parameters are then available to be modified in a sweep. The function must return the campaign module at the end. Example:: def campaign_builder(campaign, additional_param=0.3): from emodpy.campaign.individual_intervention import CommonInterventionParameters, SimpleVaccine, VaccineType from emodpy.campaign.distributor import add_intervention_scheduled import emodpy.campaign.waning_config as waning_config this_waning_config = waning_config.BoxExponential(box_duration=25, decay_time_constant=60, initial_effect=0.89) common_intervention_parameters = CommonInterventionParameters(cost=0.5, dont_allow_duplicates=True) vaccine = SimpleVaccine(campaign, waning_config=this_waning_config, vaccine_take=additional_param, vaccine_type=VaccineType.TransmissionBlocking, common_intervention_parameters=common_intervention_parameters) add_intervention_scheduled(campaign, intervention_list=[vaccine], start_day=2) return campaign	None
demographics_builder	Callable	Function that builds the demographics configuration and optional migration configuration. Example:: def demographics_builder(): import emodpy.demographics.Demographics as Demographics demographics_object = Demographics.from_template_node(pop=500) # other code setting demographics parameters return demographics_object	None
report_builder	Callable[[Reporters], Reporters]	Function that creates reports to be used in the experiment. The function must have Reporters object as the parameter and return that object. It is assumed that all the reporters come from the reporters that are part of EMOD main code. EMOD also supports reporters as custom plug-in dlls, however, not through the current emodpy system. Example:: def report_builder(reporters): from emodpy.reporters.reporters import ReportEventCounter, ReportFilter report_filter = ReportFilter(start_day=0, end_day=365, filename_suffix="LifeEvents") reporters.add(ReportEventCounter(reporters_object=reporters, report_filter=report_filter, event_list=["DiseaseDeaths", "Births", "HappyBirthday"])) return reporters	None
embedded_python_scripts_path	Union[str, Path, List[Union[str, Path]]]	Path to folder with python scripts or a list of paths to specific python scripts. When path to folder, all python scripts in the folder will be added to the experiment. embedded_python_scripts_path may also be a list of such items, each added independently. Note: We no longer support passing in a function for this parameter.	None
serialized_population_files	Union[str, List[str]]	Path to folder containing .dtk serialized population files or specific .dtk file, including the filename. All .dtk files in the folder will be added and used in the experiment.	None
bootstrapped	bool	Set to True if the campaign builder will build a campaign from scratch itself. False if it will accept an initialized campaign from this function instead and then modify it. Default False.	False

Returns:

Type	Description
EMODTask	EMODTask

Source code in emodpy/emod_task.py

@classmethod
def from_defaults(cls,
                  schema_path: str,
                  eradication_path: str = None,
                  config_builder: Callable = None,
                  campaign_builder: Callable = None,
                  demographics_builder: Callable = None,
                  report_builder: Callable[[Reporters], Reporters] = None,
                  embedded_python_scripts_path: Union[str, Path, List[Union[str, Path]]] = None,
                  serialized_population_files: Union[str, List[str]] = None,
                  bootstrapped: bool = False) -> "EMODTask":
    """
    Create a task from emod-api defaults and functions to update them.

    Args:
        schema_path: Path to processed schema.json, including the filename.
        eradication_path: Path to Eradication binary, including the filename. You can also add Eradication as an
            asset.
        config_builder: Function that sets parameters for config. The function must have config object as the
            parameter and return the config object. Inside the function, the config object will be modified in
            the following way: config.parameters.<parameter_name> = <value>, with <parameter_name> being an
            attribute.

            Example::

                def config_builder(config):
                    import emodpy.emod_enum as emod_enum
                    config.parameters.Incubation_Period_Distribution = emod_enum.DistributionType.CONSTANT_DISTRIBUTION
                    config.parameters.Incubation_Period_Constant = 5
                    config.parameters.Infectious_Period_Distribution = emod_enum.DistributionType.CONSTANT_DISTRIBUTION
                    config.parameters.Infectious_Period_Constant = 5
                    config.parameters.Simulation_Duration = 80
                    return config

        campaign_builder: Function that creates and adds interventions to the campaign module. The function needs
            to take the campaign module as the first required argument, which can then be followed by parameters
            that you  want to modify within the interventions. These additional parameters are then available to
            be modified in a sweep. The function must return the campaign module at the end.

            Example::

                def campaign_builder(campaign, additional_param=0.3):
                    from emodpy.campaign.individual_intervention import CommonInterventionParameters, SimpleVaccine, VaccineType
                    from emodpy.campaign.distributor import add_intervention_scheduled
                    import emodpy.campaign.waning_config as waning_config

                    this_waning_config = waning_config.BoxExponential(box_duration=25,
                                                                      decay_time_constant=60,
                                                                      initial_effect=0.89)
                    common_intervention_parameters = CommonInterventionParameters(cost=0.5,
                                                                                  dont_allow_duplicates=True)
                    vaccine = SimpleVaccine(campaign,
                                            waning_config=this_waning_config,
                                            vaccine_take=additional_param,
                                            vaccine_type=VaccineType.TransmissionBlocking,
                                            common_intervention_parameters=common_intervention_parameters)
                    add_intervention_scheduled(campaign, intervention_list=[vaccine], start_day=2)
                    return campaign



        demographics_builder: Function that builds the demographics configuration and optional migration
            configuration.

            Example::

                def demographics_builder():
                    import emodpy.demographics.Demographics as Demographics

                    demographics_object = Demographics.from_template_node(pop=500)
                    # other code setting demographics parameters
                    return demographics_object

        report_builder: Function that creates reports to be used in the experiment. The function must have Reporters
            object as the parameter and return that object. It is assumed that all the reporters come from the
            reporters that are part of EMOD main code. EMOD also supports reporters as custom plug-in dlls,
            however, not through the current emodpy system.

            Example::

                    def report_builder(reporters):
                        from emodpy.reporters.reporters import ReportEventCounter, ReportFilter

                        report_filter = ReportFilter(start_day=0, end_day=365, filename_suffix="LifeEvents")
                        reporters.add(ReportEventCounter(reporters_object=reporters,
                                           report_filter=report_filter,
                                           event_list=["DiseaseDeaths", "Births", "HappyBirthday"]))
                        return reporters


        embedded_python_scripts_path: Path to folder with python scripts or a list of paths to specific python
            scripts. When path to folder, all python scripts in the folder will be added to the experiment.
            embedded_python_scripts_path may also be a list of such items, each added independently.
            Note: We no longer support passing in a function for this parameter.
        serialized_population_files: Path to folder containing .dtk serialized population files or specific
            .dtk file, including the filename. All .dtk files in the folder will be added and used in the
            experiment.
        bootstrapped: Set to True if the campaign builder will build a campaign from scratch itself. False if it
            will accept an initialized campaign from this function instead and then modify it. Default False.

    Returns:
        EMODTask
    """
    task = cls(eradication_path=eradication_path, schema_path=schema_path)
    # We do not regenerate the schema from the Eradication binary because we can't guarantee this code is running
    # on a matching platform, so we use a schema file.
    default_config = cls.build_default_config(schema_path=task.schema_path)
    task.available_config_parameters = list(default_config['parameters'].keys())

    task.config = dfs.get_config_from_default_and_params(config=default_config, set_fn=config_builder)
    if not isinstance(task.config, ReadOnlyDict):
        raise ValueError("Something went wrong with config_builder, please make sure "
                         "the config_builder function returns a config object.")

    # Let's do the demographics building here...
    if demographics_builder:
        task.create_demographics_from_callback(demographics_builder)

    if campaign_builder:
        task.create_campaign_from_callback(builder=campaign_builder, bootstrapped=bootstrapped)

    if embedded_python_scripts_path:
        task.add_embedded_python_scripts_from_path(path=embedded_python_scripts_path)

    if report_builder:
        task.reporters = Reporters(schema_path=task.schema_path)
        returned = report_builder(task.reporters)
        if not returned or not isinstance(returned, Reporters):
            raise ValueError("Something went wrong with report_builder, please make sure "
                             "the report_builder function returns a Reporters object.")
        task._validate_reporter_listening_events()

    if serialized_population_files:
        task.add_serialized_population_files_from_path(serialized_population_files)

    task.handle_implicit_configs()

    return task

from_files(eradication_path=None, config_path=None, campaign_path=None, demographics_paths=None, custom_reports_path=None, embedded_python_scripts_path=None, serialized_population_files=None, asset_path=None) classmethod

Load custom EMOD files when creating [EMODTask][].

Parameters:

Name	Type	Description	Default
eradication_path	str	Path to Eradication binary, including the filename.	None
config_path	str	Path to the configuration file, including the filename.	None
campaign_path	str	Path to the campaign file, including the filename.	None
demographics_paths	Union[str, list]	Path or a list of paths to the folder containing demographics files or path to a specific demographics file, including the filename. When path to folder, all .json files in the folder will be added to the experiment as demographics files.	None
custom_reports_path	str	Path to the custom reports file, including the filename. It is assumed that all the reporters in the file come from the reporters that are part of EMOD main code. EMOD also supports reporters as custom plug-in dlls, however, not through the current emodpy system.	None
embedded_python_scripts_path	Union[str, list[str]]	Path to folder with python scripts or path to a specific python script, including the filename. When path to folder, all python scripts in the folder will be added to the experiment. Note: We no longer support passing in a function for this parameter.	None
serialized_population_files	Union[str, list[str]]	Path to folder containing .dtk serialized population files or specific .dtk file, including the filename. All .dtk files in the folder will be added and used in the experiment.	None
asset_path	str	Path to migration and/or climate files.	None

Returns:

Type	Description
EMODTask	An initialized experiment

Source code in emodpy/emod_task.py

@classmethod
def from_files(cls,
               eradication_path: str = None,
               config_path: str = None,
               campaign_path: str = None,
               demographics_paths: Union[str, list] = None,
               custom_reports_path: str = None,
               embedded_python_scripts_path: Union[str, list[str]] = None,
               serialized_population_files: Union[str, list[str]] = None,
               asset_path: str = None) -> "EMODTask":
    """
    Load custom EMOD files when creating [EMODTask][].

    Args:
        eradication_path: Path to Eradication binary, including the filename.
        config_path: Path to the configuration file, including the filename.
        campaign_path: Path to the campaign file, including the filename.
        demographics_paths: Path or a list of paths to the folder containing demographics files or path to a
            specific demographics file, including the filename. When path to folder, all .json files in the folder
            will be added to the experiment as demographics files.
        custom_reports_path: Path to the custom reports file, including the filename. It is assumed that all the
            reporters in the file come from the reporters that are part of EMOD main code. EMOD also supports
            reporters as custom plug-in dlls, however, not through the current emodpy system.
        embedded_python_scripts_path: Path to folder with python scripts or path to a specific python script,
            including the filename. When path to folder, all python scripts in the folder will be added to the
            experiment.
            Note: We no longer support passing in a function for this parameter.
        serialized_population_files: Path to folder containing .dtk serialized population files or specific
            .dtk file, including the filename. All .dtk files in the folder will be added and used in the
            experiment.
        asset_path: Path to migration and/or climate files.

    Returns:
        An initialized experiment
    """
    # Create the experiment
    task = cls(eradication_path=eradication_path)

    task.config = load_json_file(config_path)["parameters"]

    if campaign_path:
        task.campaign = EMODCampaign.load_from_file(campaign_path)

    if demographics_paths:
        if isinstance(demographics_paths, list):
            for path in demographics_paths:
                task.demographics.add_demographics_from_files(path)
        else:
            task.demographics.add_demographics_from_files(demographics_paths)
        task.demographics.set_task_config(task)  # adds the demographics to the config

    if custom_reports_path:
        custom_reports_path = os.path.abspath(custom_reports_path)
        task.transient_assets.add_asset(custom_reports_path)
        # need to do this explicitly because this is usually added by the Reporters object,
        # but we are not using it here.
        custom_reports_filename = os.path.basename(custom_reports_path)
        task.config["Custom_Reports_Filename"] = custom_reports_filename

    if embedded_python_scripts_path is not None:
        task.add_embedded_python_scripts_from_path(path=embedded_python_scripts_path)

    if serialized_population_files is not None:
        task.add_serialized_population_files_from_path(serialized_population_files)

    if asset_path and config_path:
        # Look for climate
        task.climate.read_config_file(config_path, asset_path)
        # Look for migrations
        task.migrations.read_config_file(config_path, asset_path)

    return task

gather_common_assets()

Gather Experiment Level Assets Returns:

Source code in emodpy/emod_task.py

def gather_common_assets(self) -> AssetCollection:
    """
    Gather Experiment Level Assets
    Returns:

    """
    # check whether there are any .sif or .img files in the common assets diretories...
    # Add Eradication.exe to assets
    if self.eradication_path:
        self.common_assets.add_asset(Asset(absolute_path=self.eradication_path,
                                           filename=self.executable_name),
                                     fail_on_duplicate=False)

    # Add demographics to assets
    if self.demographics.assets:
        self.common_assets.extend(self.demographics.gather_assets())

    # Add the migrations
    if self.migrations.assets:
        self.common_assets.extend(self.migrations.gather_assets())

    # Add the climate
    if self.climate.assets:
        self.common_assets.extend(self.climate.gather_assets())

    return self.common_assets

gather_transient_assets()

Gather assets that are per simulation Returns: AssetCollection

Source code in emodpy/emod_task.py

def gather_transient_assets(self) -> AssetCollection:
    """
    Gather assets that are per simulation
    Returns:
        AssetCollection
    """

    # This config code needs to be rewritten
    # task.config contains emod-api version of config i.e., with schema. Needs to be finalized and written.
    if logger.isEnabledFor(DEBUG) and dev_mode:
        logger.debug("DEBUG: Calling finalize.")

    # Add config and campaign to assets as needed

    if self.config:
        if type(self.config) is dict:  # old/basic style, when "from_files" is used
            self.config = {"parameters": self.config}
        else:
            self.config.parameters.finalize()
        self._enforce_non_schema_coherence()
        if dev_mode:
            with open(self.config_file_name, "w") as fp:
                json.dump(self.config, fp, sort_keys=True, indent=4)
        asset = Asset(filename=self.config_file_name, content=json.dumps(self.config, sort_keys=True))
        self.transient_assets.add_asset(asset=asset, fail_on_duplicate=False)

    if self.campaign:
        asset = Asset(filename="campaign.json", content=self.campaign.json)
        self.transient_assets.add_asset(asset=asset, fail_on_duplicate=False)

        if dev_mode:
            import emod_api.peek_camp as base_peek_camp
            original = sys.stdout
            sys.stdout = open('campaign.ccdl', 'w')
            base_peek_camp.decode("campaign.json", self.config_file_name)
            sys.stdout = original
            data = open("campaign.ccdl").readlines()
            data.sort()
            for i in range(len(data)):
                print(data[i].strip())

    # Add custom_reporters.json if needed
    if self.reporters.builtin_reporters:
        asset = Asset(filename="custom_reports.json", content=self.reporters.json)
        self.transient_assets.add_asset(asset=asset, fail_on_duplicate=False)

    # Add demographics files to assets
    if self.simulation_demographics.assets:
        self.transient_assets.extend(self.simulation_demographics.gather_assets())

    # Add the migrations
    if self.simulation_migrations.assets:
        self.transient_assets.extend(self.simulation_migrations.gather_assets())

    return self.transient_assets

get_parameter(name, default=None)

Get a parameter in the simulation.

Parameters:

Name	Type	Description	Default
name	str	The name of the parameter.	required
default	Optional[Any]	Optional, the default value.	None

Returns:

Type	Description
Any	The value of the parameter.

Source code in emodpy/emod_task.py

def get_parameter(self, name: str, default: Optional[Any] = None) -> Any:
    """
    Get a parameter in the simulation.

    Args:
        name: The name of the parameter.
        default: Optional, the default value.

    Returns:
        The value of the parameter.
    """
    return self.config.get(name, default)

handle_implicit_configs()

Execute the implicit config functions created by the demographics builder.

Returns:

Type	Description
None	None

Source code in emodpy/emod_task.py

def handle_implicit_configs(self) -> None:
    """
    Execute the implicit config functions created by the demographics builder.

    Returns:
        None
    """
    if dev_mode:
        logger.debug(f"Executing {len(self.implicit_configs)} implicit config functions from demographics and "
                     f"migration.")
    for fn in self.implicit_configs:
        if fn:
            self.config = fn(self.config)

pre_creation(parent, platform)

Call before a task is executed. This ensures our configuration is properly done

Source code in emodpy/emod_task.py

def pre_creation(self, parent: Union[Simulation, IWorkflowItem], platform: 'IPlatform'):
    """
    Call before a task is executed. This ensures our configuration is properly done

    """
    # Set the demographics
    # self.demographics.set_task_config(self)
    # self.simulation_demographics.set_task_config(self, extend=True)

    # Set the migrations
    self.simulation_migrations.merge_with(self.migrations)
    # self.simulation_migrations.set_task_config(self)

    # Set the climate
    # self.climate.set_task_config(self)

    # Set the reporters
    # this only runs for when using from_defaults,
    # because with "from_files" we bypass Reporters object creation
    if self.reporters.builtin_reporters:
        self.config.parameters.Custom_Reports_Filename = "custom_reports.json"
    for reporter in self.reporters.config_reporters:
        for i in reporter.parameters:
            setattr(self.config.parameters, i, reporter.parameters[i])

    # Set the campaign filename
    if self.campaign:
        if type(self.config) is dict:  # when "from_files" was used
            self.config["Campaign_Filename"] = "campaign.json"
            self.config["Enable_Interventions"] = 1  # implicit?
        else:    # when using from_defaults
            self.config.parameters.Campaign_Filename = "campaign.json"
            self.config.parameters.Enable_Interventions = 1  # implicit

    # Gather the custom coordinator, individual, and node events
    self.set_command_line()
    super().pre_creation(parent, platform)
    if not platform.is_windows_platform():
        # print( "Target is LINUX!" )
        self.is_linux = True

set_command_line()

Build and set the command line object.

Returns:

Source code in emodpy/emod_task.py

def set_command_line(self) -> None:
    """
    Build and set the command line object.

    Returns:

    """
    # In COMPS, it's rare to have to specify multiple paths because 'we' control the environment and
    # can put everything in Assets. The multiple input paths is useful for local command-line usage where
    # the input files are spread across different locations. Note that with symlinks it's trivial to put
    # all the files in one path without copying. The only exception here is when we are using dtk_pre_process
    # to create a (demographics) input file and this can not be in Assets.
    # input_path = "./Assets" # this works on windows

    # Both "./Assets\;." and "./Assets\\;." work but the former confuses the linter because it is expecting
    # a known escape code, e.g. "\n" - escaping the backslash with "\\" escapes the escape code (got that?).
    input_path = "./Assets\\;."

    # Create the command line according to self. location of the model
    if self.sif_filename:
        self.command = CommandLine(
            "singularity",
            "exec",
            f"Assets/{self.sif_filename}",
            f"Assets/{self.executable_name}",
            "--config",
            f"{self.config_file_name}",
            "--dll-path",   # for Generic-Ongoing Eradication
            "./Assets",
        )
    else:
        self.command = CommandLine(f"Assets/{self.executable_name}", "--config",
                                   f"{self.config_file_name}", "--dll-path", "./Assets")

    if self.use_embedded_python:
        list_sep = ";"
        self.command._options.update({"--python-script-path": list_sep.join(self.py_path_list)})

    # We do this here because CommandLine tries to be smart and quote input_path, but it isn't quite right...
    self.command.add_raw_argument("--input-path")
    self.command.add_raw_argument(input_path)

set_parameter(name, value)

Set a value in the EMOD config.json file. This will be deprecated in the future in favour of emod_api.config.

Parameters:

Name	Type	Description	Default
name	str	Name of parameter to set	required
value	any	Value to set	required

Returns:

Type	Description
dict	Tags to set

Source code in emodpy/emod_task.py

def set_parameter(self, name: str, value: any) -> dict:
    """
    Set a value in the EMOD config.json file. This will be deprecated in the future in favour of emod_api.config.

    Args:
        name: Name of parameter to set
        value: Value to set

    Returns:
        Tags to set
    """
    if not self.config:
        raise ValueError("task.config is empty. Please load a config file or dictionary"
                         " or create a new one using from_defaults.")
    if type(self.config) is dict:  # old style, when "from_files" is used
        self.config[name] = value  # no checks against the schema
    elif hasattr(self.config.parameters, name):
        setattr(self.config.parameters, name, value)
    else:
        raise ValueError(f"Parameter '{name}' not a valid parameter based on schema.")

    return {name: value}

set_parameter_partial(parameter) classmethod

Convenience callback for sweeps

Parameters:

Name	Type	Description	Default
parameter	str	Parameter to set	required

Returns:

Type	Description
partial[Simulation]	Partial function

Source code in emodpy/emod_task.py

@classmethod
def set_parameter_partial(cls, parameter: str) -> partial[Simulation]:
    """
    Convenience callback for sweeps

    Args:
        parameter: Parameter to set

    Returns:
        Partial function
    """
    return partial(cls.set_parameter_sweep_callback, param=parameter)

set_parameter_sweep_callback(simulation, param, value) staticmethod

Convenience callback for sweeps

Parameters:

Name	Type	Description	Default
simulation	Simulation	Simulation we are updating	required
param	str	Parameter	required
value	Any	Value	required

Returns:

Type	Description
Dict[str, Any]	Tags to set on simulation

Source code in emodpy/emod_task.py

@staticmethod
def set_parameter_sweep_callback(simulation: Simulation, param: str, value: Any) -> Dict[str, Any]:
    """
    Convenience callback for sweeps

    Args:
        simulation: Simulation we are updating
        param: Parameter
        value: Value

    Returns:
        Tags to set on simulation
    """
    if not hasattr(simulation.task, 'set_parameter'):
        raise ValueError("set_parameter_sweep_callback can only be used on tasks with a set_parameter")
    return simulation.task.set_parameter(param, value)

set_sif(path_to_sif, platform)

Set the singularity image file to use for creating the EMOD execution environment.

Parameters:

Name	Type	Description	Default
path_to_sif	Union[Path, str]	non-COMPSPlatforms: The file system path to the sif file for creating the EMOD execution environment. COMPSPlatform: Either: a) The same sif file path as with other platforms b) A .id file containing a COMPS AssetCollection id containing the desired sif file, formatted e.g.:: 8df53802-53f3-ec11-a9f9-b88303911bc1::Asset Collection ContainerPlatform: No sif file specification is used.	required
platform	IPlatform	Platform object to use for this task.	required

Returns:

Type	Description
None	None

Source code in emodpy/emod_task.py

def set_sif(self, path_to_sif: Union[Path, str], platform: IPlatform) -> None:
    """
    Set the singularity image file to use for creating the EMOD execution environment.

    Args:
        path_to_sif:

            non-COMPSPlatforms: The file system path to the sif file for creating the EMOD execution environment.
            COMPSPlatform: Either:

                a) The same sif file path as with other platforms
                b) A .id file containing a COMPS AssetCollection id containing the desired sif file, formatted e.g.::

                    8df53802-53f3-ec11-a9f9-b88303911bc1::Asset Collection

            ContainerPlatform: No sif file specification is used.

        platform: Platform object to use for this task.

    Returns:
        None
    """
    platform_type = platform.__class__.__name__
    path_to_sif = Path(path_to_sif)
    suffix = path_to_sif.suffix
    is_sif_file = path_to_sif.suffix == '.sif'
    is_id_file = path_to_sif.suffix == '.id'

    # path_to_sif = str(path_to_sif)  # code below is written to require strings
    platforms = ['SlurmPlatform', 'FilePlatform', 'ProcessPlatform']
    if platform_type in platforms:
        # only .sif is valid for these platforms
        if is_id_file:
            platforms_str = ', '.join(platforms)
            raise ValueError(f"Cannot use a .id file for sif specification on these platforms: {platforms_str}")
        elif is_sif_file:
            self.sif_path = path_to_sif
        else:
            raise ValueError(f"Unknown sif file with extension: {suffix}")
    elif platform_type == 'COMPSPlatform':
        # .sif and .id files are valid for COMPSPlatform
        if is_id_file:
            ac = AssetCollection.from_id_file(path_to_sif)
            sif_asset = [acf for acf in ac.assets if Path(acf.filename).suffix == ".sif"][0]
            self.sif_filename = sif_asset.filename
        elif is_sif_file:
            sif_asset = path_to_sif
            self.sif_filename = path_to_sif.name
        else:
            raise ValueError(f"Unknown sif file with extension: {suffix}")
        self.common_assets.add_asset(sif_asset)
    elif platform_type == 'ContainerPlatform':
        warnings.warn('.sif or .id file specified for platform of type ContainerPlatform . '
                      'This is unnecessary and is being ignored.', RuntimeWarning)
    else:
        raise ValueError(f"Unknown platform type for setting sif file: {platform_type}")

EMODTaskSpecification

Bases: TaskSpecification

Idmtools implemented each platform and task as a Plugin and idmtools is able to identify them and use them dynamically. For example, idmtools workflow is able to crete each Platform with Plaform Factory and each Task with Task Factory. The link defined in setup.py entrypoint is the way to allow idmtools workflow to identify them.

Take EMODTaskSpecification as an example, idmtools uses it internally and indirectly (user usually doesn't create an instance of it, but idmtools work may do in certain case). This EMODTaskSpecification entry in setup.py has been used in idmtools workflow in two places:

CLI command - check idmtools related packages installed (version including plugins): idmtools version - check available Task: idmtools info plugins task

Source code in emodpy/emod_task.py

class EMODTaskSpecification(TaskSpecification):
    """
    Idmtools implemented each platform and task as a Plugin and idmtools is able to identify them and use them
    dynamically. For example, idmtools workflow is able to crete each Platform with Plaform Factory and each Task with
    Task Factory. The link defined in setup.py entrypoint is the way to allow idmtools workflow to identify them.

    Take EMODTaskSpecification as an example, idmtools uses it internally and indirectly (user usually doesn't create
    an instance of it, but idmtools work may do in certain case). This EMODTaskSpecification entry in setup.py has been
    used in idmtools workflow in two places:

    CLI command
    - check idmtools related packages installed (version including plugins): idmtools version
    - check available Task: idmtools info plugins task
    """

    def get(self, configuration: dict) -> EMODTask:
        """
        Return an EMODTask object using provided configuration
        Args:
            configuration: Configuration for Task

        Returns:
            EMODTask for configuration
        """
        return EMODTask(**configuration)

    def get_description(self) -> str:
        """
        Defines a description of the plugin

        Returns:
            Plugin description
        """
        return "Defines a EMODTask command"

    def get_type(self) -> Type[EMODTask]:
        """
        Returns the Task type defined by specification

        Returns:

        """
        return EMODTask

    def get_version(self) -> str:
        """
        Return the version string for EMODTask. This should be the module version so return that

        Returns:
            Version
        """
        from emodpy import __version__
        return __version__

get(configuration)

Return an EMODTask object using provided configuration Args: configuration: Configuration for Task

Returns:

Type	Description
EMODTask	EMODTask for configuration

Source code in emodpy/emod_task.py

def get(self, configuration: dict) -> EMODTask:
    """
    Return an EMODTask object using provided configuration
    Args:
        configuration: Configuration for Task

    Returns:
        EMODTask for configuration
    """
    return EMODTask(**configuration)

get_description()

Defines a description of the plugin

Returns:

Type	Description
str	Plugin description

Source code in emodpy/emod_task.py

def get_description(self) -> str:
    """
    Defines a description of the plugin

    Returns:
        Plugin description
    """
    return "Defines a EMODTask command"

get_type()

Returns the Task type defined by specification

Returns:

Source code in emodpy/emod_task.py

def get_type(self) -> Type[EMODTask]:
    """
    Returns the Task type defined by specification

    Returns:

    """
    return EMODTask

get_version()

Return the version string for EMODTask. This should be the module version so return that

Returns:

Type	Description
str	Version

Source code in emodpy/emod_task.py

def get_version(self) -> str:
    """
    Return the version string for EMODTask. This should be the module version so return that

    Returns:
        Version
    """
    from emodpy import __version__
    return __version__