Source code for idmtools.analysis.tags_analyzer

Example of a tags analyzer to get all the tags from your experiment simulations into one csv file.

Copyright 2021, Bill & Melinda Gates Foundation. All rights reserved.
# First, import some necessary system and idmtools packages.
import os
from typing import Dict, Any

import pandas as pd
from idmtools.entities import IAnalyzer

# Create a class for the analyzer
from idmtools.entities.ianalyzer import ANALYZABLE_ITEM

[docs]class TagsAnalyzer(IAnalyzer): """ Provides an analyzer for CSV output. Examples: .. literalinclude:: ../../examples/analyzers/ """ # Arg option for analyzer init are uid, working_dir, parse (True to leverage the :class:`OutputParser`; # False to get the raw data in the :meth:`select_simulation_data`), and filenames # In this case, we want uid, working_dir, and parse=True
[docs] def __init__(self, uid=None, working_dir=None, parse=True, output_path="output_tag"): """ Initialize our Tags Analyzer. Args: uid: working_dir: parse: output_path: See Also: :class:`~idmtools.entities.ianalyzer.IAnalyzer`. """ super().__init__(uid, working_dir, parse) self.exp_id = None self.output_path = output_path
[docs] def initialize(self): """ Initialize the item before mapping data. Here we create a directory for the output. Returns: None """ self.output_path = os.path.join(self.working_dir, self.output_path) # Create the output path if not os.path.exists(self.output_path): os.makedirs(self.output_path)
# Map is called to get for each simulation a data object (all the metadata of the simulations) and simulation object
[docs] def map(self, data: Dict[str, Any], simulation: ANALYZABLE_ITEM): """ Map our data for our Workitems/Simulations. In this case, we just extract the tags and build a dataframe from those. Args: data: List of files. This should be empty for us. simulation: Item to extract Returns: Data frame with the tags built. """ df = pd.DataFrame(columns=list(simulation.tags.keys())) # Create a dataframe with the simulation tag keys df.loc[str(simulation.uid)] = list(simulation.tags.values()) # Get a list of the sim tag values = 'SimId' # Label the index keys you create with the names option return df
# In reduce, we are printing the simulation and result data filtered in map
[docs] def reduce(self, all_data: Dict[ANALYZABLE_ITEM, pd.DataFrame]): """ Reduce the dictionary of items->Tags dataframe to a single dataframe and write to a csv file. Args: all_data: Map of Item->Tags dataframe Returns: None """ results = pd.concat(list(all_data.values()), axis=0) # Combine a list of all the sims tag values # Make a directory labeled the exp id to write the csv results to first_sim = list(all_data.keys())[0] # get first Simulation exp_id = # Set the exp id from the first sim data output_folder = os.path.join(self.output_path, exp_id) os.makedirs(output_folder, exist_ok=True) results.to_csv(os.path.join(output_folder, self.__class__.__name__ + '.csv'))