Preprocessor Base

oqd_heisenberg_ion.common.preprocessor

base

Preprocessor

Preprocessor base class. Handles the preprocessing step before a driver is called Carries a set of keys for which a single value should always be specified. These are: "root_folder", "simulation_folder", "number_of_threads", "output_folder_name"

Raises:

• Exception –

  if a required parameter is not specified

• Exception –

  if a uuid is repeated

• Exception –

  if multiple values are provided for a parameter that should have a single value

Source code in src/oqd_heisenberg_ion/common/preprocessor/base.py

class Preprocessor:
    """
    Preprocessor base class. Handles the preprocessing step before a driver is called
    Carries a set of keys for which a single value should always be specified.
    These are: "root_folder", "simulation_folder", "number_of_threads", "output_folder_name"

    Raises:
        Exception: if a required parameter is not specified
        Exception: if a uuid is repeated
        Exception: if multiple values are provided for a parameter that should have a single value
    """

    keys_single_parameters = {"root_folder", "simulation_folder", "number_of_threads", "output_folder_name"}

    def __init__(self, parameter_set_list):
        """
        constructor for the Preprocessor base class

        Args:
            parameter_set_list (list[dict]): list of unprocessed parameter sets specified as dicts
        """

        self.parameter_set_list = parameter_set_list
        self.num_parameter_sets = len(parameter_set_list)

        self.root_folder = None
        self.simulation_folder = None

        self.driver_inputs = None
        self.processed_configs = []

    def preprocess(self):
        """
        Each subclass must implement a preprocess method
        """

        pass

    def create_output_folder(self):
        """
        creates the output folder to contain the simulation outputs inside the root folder

        Returns:
            (str): class attribute simulation_folder, the path to the simulation output folder
        """

        self.root_folder = os.path.abspath(self.parameter_set_list[0]["root_folder"])
        os.makedirs(self.root_folder, exist_ok=True)

        output_folder_name_provided = self.check_single_input("output_folder_name", True)

        if output_folder_name_provided:
            self.output_folder_name = self.parameter_set_list[0]["output_folder_name"]
        else:
            self.output_folder_name = datetime.now().strftime("%Y_%m_%d_%H_%M_%S")

        self.simulation_folder = os.path.abspath(os.path.join(self.root_folder, self.output_folder_name))

        os.mkdir(self.simulation_folder)

        return self.simulation_folder

    def create_run_folder(self, misc_args):
        """
        creates the run folder for a given parameter set inside the simulation output folder

        Args:
            misc_args (dict): contains the key "uuid" to specify the run_id corresponding to the associated parameter set

        Returns:
            (str): path to the parameter set run folder
        """

        simulation_folder = self.simulation_folder
        run_id = str(misc_args["uuid"])
        run_folder = os.path.join(simulation_folder, run_id)

        os.mkdir(run_folder)

        return run_folder

    def get_run_id(self, misc_args):
        """
        Extracts the uuid if specified, and creates one if not specified in the arguments

        Args:
            misc_args (dict): contains the key "uuid" to specify the run_id corresponding to the associated parameter set

        Returns:
            (str): the uuid associated with the parameter set
        """

        if "uuid" in misc_args:
            run_id = misc_args["uuid"]
        else:
            run_id = uuid.uuid4()

        return run_id

    def check_input_provided(self, key, optional=False):
        """
        checks whether an input has been specified in the parameter set

        Args:
            key (str): _description_
            optional (bool, optional): determines whether value associated with key is optional. Defaults to False.

        Raises:
            Exception: is a required input is not provided

        Returns:
            (bool): determines whether an input was provided for the given key
        """

        input_provided = key in self.parameter_set_list[0]
        if not input_provided and not optional:
            raise Exception(f"Missing required key: {key}")
        else:
            return input_provided

    def check_unique_uuids(self):
        """
        checks if multiple uuids are provided and if they are all unique. Does not throw an error if no uuid is provided

        Raises:
            Exception: if a uuid is repeated
        """

        input_provided = self.check_input_provided("uuid", True)
        if input_provided:
            uuid_set = {self.parameter_set_list[i]["uuid"] for i in range(self.num_parameter_sets)}
            if len(uuid_set) != self.num_parameter_sets:
                raise Exception("Specified uuids are not unique\n")

    def check_single_input(self, key, optional=False):
        """
        checks if multiple inputs are provided for a given key if it should have a single input

        Args:
            key (str): the parameter key
            optional (bool, optional): specifies if the parameter is optional. Defaults to False.

        Raises:
            Exception: if multiple inputs provided for a key that should have a single input

        Returns:
            (bool): determines whether the specified input was provided
        """

        input_provided = self.check_input_provided(key, optional)
        if input_provided:
            val = self.parameter_set_list[0][key]
            for i in range(self.num_parameter_sets):
                if self.parameter_set_list[i][key] != val:
                    raise Exception(f"There should only be one input for key: {key}")

        return input_provided

    def extract_optional_input(self, key, unique=False):
        """
        extracts an optional input if provided and return None otherwise

        Args:
            key (str): parameter key
            unique (bool, optional): determines if the value needs to be unique. Defaults to False.

        Returns:
            (str): the value associated with key if provided, otherwise returns None
        """

        if unique:
            if self.check_single_input(key, True):
                return self.parameter_set_list[0][key]
            else:
                return None
        else:
            if self.check_input_provided(key, True):
                return self.parameter_set_list[0][key]
            else:
                return None

    def write_input_file(self):
        """
        writes a tab-delimited input file for the engine using the parameter set values
        """

        sse_input_file = os.path.join(self.simulation_folder, "inputs.txt")

        with open(sse_input_file, "w") as f:
            for param_group_name, param_group in self.processed_configs[0].items():
                for key in param_group.keys():
                    text_line = key + "\t" + str(self.processed_configs[0][param_group_name][key])
                    for i in range(1, self.num_parameter_sets):
                        if key not in self.keys_single_parameters:
                            text_line += "," + str(self.processed_configs[i][param_group_name][key])
                    text_line += "\n"
                    f.write(text_line)

__init__

constructor for the Preprocessor base class

Parameters:

• parameter_set_list (list[dict]) –

  list of unprocessed parameter sets specified as dicts

Source code in src/oqd_heisenberg_ion/common/preprocessor/base.py

def __init__(self, parameter_set_list):
    """
    constructor for the Preprocessor base class

    Args:
        parameter_set_list (list[dict]): list of unprocessed parameter sets specified as dicts
    """

    self.parameter_set_list = parameter_set_list
    self.num_parameter_sets = len(parameter_set_list)

    self.root_folder = None
    self.simulation_folder = None

    self.driver_inputs = None
    self.processed_configs = []

preprocess

Each subclass must implement a preprocess method

Source code in src/oqd_heisenberg_ion/common/preprocessor/base.py

def preprocess(self):
    """
    Each subclass must implement a preprocess method
    """

    pass

create_output_folder

creates the output folder to contain the simulation outputs inside the root folder

Returns:

• str –

  class attribute simulation_folder, the path to the simulation output folder

Source code in src/oqd_heisenberg_ion/common/preprocessor/base.py

def create_output_folder(self):
    """
    creates the output folder to contain the simulation outputs inside the root folder

    Returns:
        (str): class attribute simulation_folder, the path to the simulation output folder
    """

    self.root_folder = os.path.abspath(self.parameter_set_list[0]["root_folder"])
    os.makedirs(self.root_folder, exist_ok=True)

    output_folder_name_provided = self.check_single_input("output_folder_name", True)

    if output_folder_name_provided:
        self.output_folder_name = self.parameter_set_list[0]["output_folder_name"]
    else:
        self.output_folder_name = datetime.now().strftime("%Y_%m_%d_%H_%M_%S")

    self.simulation_folder = os.path.abspath(os.path.join(self.root_folder, self.output_folder_name))

    os.mkdir(self.simulation_folder)

    return self.simulation_folder

create_run_folder

creates the run folder for a given parameter set inside the simulation output folder

Parameters:

• misc_args (dict) –

  contains the key "uuid" to specify the run_id corresponding to the associated parameter set

Returns:

• str –

  path to the parameter set run folder

Source code in src/oqd_heisenberg_ion/common/preprocessor/base.py

def create_run_folder(self, misc_args):
    """
    creates the run folder for a given parameter set inside the simulation output folder

    Args:
        misc_args (dict): contains the key "uuid" to specify the run_id corresponding to the associated parameter set

    Returns:
        (str): path to the parameter set run folder
    """

    simulation_folder = self.simulation_folder
    run_id = str(misc_args["uuid"])
    run_folder = os.path.join(simulation_folder, run_id)

    os.mkdir(run_folder)

    return run_folder

get_run_id

Extracts the uuid if specified, and creates one if not specified in the arguments

Parameters:

• misc_args (dict) –

  contains the key "uuid" to specify the run_id corresponding to the associated parameter set

Returns:

• str –

  the uuid associated with the parameter set

Source code in src/oqd_heisenberg_ion/common/preprocessor/base.py

def get_run_id(self, misc_args):
    """
    Extracts the uuid if specified, and creates one if not specified in the arguments

    Args:
        misc_args (dict): contains the key "uuid" to specify the run_id corresponding to the associated parameter set

    Returns:
        (str): the uuid associated with the parameter set
    """

    if "uuid" in misc_args:
        run_id = misc_args["uuid"]
    else:
        run_id = uuid.uuid4()

    return run_id

check_input_provided

checks whether an input has been specified in the parameter set

Parameters:

• key (str) –

  description

• optional (bool, default: False ) –

  determines whether value associated with key is optional. Defaults to False.

Raises:

• Exception –

  is a required input is not provided

Returns:

• bool –

  determines whether an input was provided for the given key

Source code in src/oqd_heisenberg_ion/common/preprocessor/base.py

def check_input_provided(self, key, optional=False):
    """
    checks whether an input has been specified in the parameter set

    Args:
        key (str): _description_
        optional (bool, optional): determines whether value associated with key is optional. Defaults to False.

    Raises:
        Exception: is a required input is not provided

    Returns:
        (bool): determines whether an input was provided for the given key
    """

    input_provided = key in self.parameter_set_list[0]
    if not input_provided and not optional:
        raise Exception(f"Missing required key: {key}")
    else:
        return input_provided

check_unique_uuids

checks if multiple uuids are provided and if they are all unique. Does not throw an error if no uuid is provided

Raises:

• Exception –

  if a uuid is repeated

Source code in src/oqd_heisenberg_ion/common/preprocessor/base.py

def check_unique_uuids(self):
    """
    checks if multiple uuids are provided and if they are all unique. Does not throw an error if no uuid is provided

    Raises:
        Exception: if a uuid is repeated
    """

    input_provided = self.check_input_provided("uuid", True)
    if input_provided:
        uuid_set = {self.parameter_set_list[i]["uuid"] for i in range(self.num_parameter_sets)}
        if len(uuid_set) != self.num_parameter_sets:
            raise Exception("Specified uuids are not unique\n")

check_single_input

checks if multiple inputs are provided for a given key if it should have a single input

Parameters:

• key (str) –

  the parameter key

• optional (bool, default: False ) –

  specifies if the parameter is optional. Defaults to False.

Raises:

• Exception –

  if multiple inputs provided for a key that should have a single input

Returns:

• bool –

  determines whether the specified input was provided

Source code in src/oqd_heisenberg_ion/common/preprocessor/base.py

def check_single_input(self, key, optional=False):
    """
    checks if multiple inputs are provided for a given key if it should have a single input

    Args:
        key (str): the parameter key
        optional (bool, optional): specifies if the parameter is optional. Defaults to False.

    Raises:
        Exception: if multiple inputs provided for a key that should have a single input

    Returns:
        (bool): determines whether the specified input was provided
    """

    input_provided = self.check_input_provided(key, optional)
    if input_provided:
        val = self.parameter_set_list[0][key]
        for i in range(self.num_parameter_sets):
            if self.parameter_set_list[i][key] != val:
                raise Exception(f"There should only be one input for key: {key}")

    return input_provided

extract_optional_input

extracts an optional input if provided and return None otherwise

Parameters:

• key (str) –

  parameter key

• unique (bool, default: False ) –

  determines if the value needs to be unique. Defaults to False.

Returns:

• str –

  the value associated with key if provided, otherwise returns None

Source code in src/oqd_heisenberg_ion/common/preprocessor/base.py

def extract_optional_input(self, key, unique=False):
    """
    extracts an optional input if provided and return None otherwise

    Args:
        key (str): parameter key
        unique (bool, optional): determines if the value needs to be unique. Defaults to False.

    Returns:
        (str): the value associated with key if provided, otherwise returns None
    """

    if unique:
        if self.check_single_input(key, True):
            return self.parameter_set_list[0][key]
        else:
            return None
    else:
        if self.check_input_provided(key, True):
            return self.parameter_set_list[0][key]
        else:
            return None

write_input_file

writes a tab-delimited input file for the engine using the parameter set values

Source code in src/oqd_heisenberg_ion/common/preprocessor/base.py

def write_input_file(self):
    """
    writes a tab-delimited input file for the engine using the parameter set values
    """

    sse_input_file = os.path.join(self.simulation_folder, "inputs.txt")

    with open(sse_input_file, "w") as f:
        for param_group_name, param_group in self.processed_configs[0].items():
            for key in param_group.keys():
                text_line = key + "\t" + str(self.processed_configs[0][param_group_name][key])
                for i in range(1, self.num_parameter_sets):
                    if key not in self.keys_single_parameters:
                        text_line += "," + str(self.processed_configs[i][param_group_name][key])
                text_line += "\n"
                f.write(text_line)

factory

PreprocessorFactory

Factory for generating the required instance of the Preprocessor subclass. Carries a registry of Preprocessor subclasses

Raises:

• Exception –

  if requested subclass is not found

Source code in src/oqd_heisenberg_ion/common/preprocessor/factory.py

class PreprocessorFactory:
    """
    Factory for generating the required instance of the Preprocessor subclass. Carries a registry of Preprocessor subclasses

    Raises:
        Exception: if requested subclass is not found
    """

    registry = {}

    def register(cls, name, subclass):
        """
        registers the specified subclass with the given name in the factory

        Args:
            name (str): name to be used while registering the specified subclass
            subclass (Type[Preprocessor]): the subclass of preprocessor to be registered
        """

        cls.registry[name] = subclass

    def create(cls, name, parameter_set_list):
        """
        creates an instance of the requested subclass

        Args:
            name (str): requested subclass name
            parameter_set_list (list[dict]): list of parameter sets, each set specified as a dict, needed by the preprocessor constructor

        Raises:
            Exception: if requested Preprocessor subclass is not found

        Returns:
            (Preprocessor): instance of Preprocessor subclass requested
        """

        if name not in cls.registry:
            raise Exception(f"Preprocessor implementation not found for name: {name}")
        else:
            return cls.registry[name](parameter_set_list)

register

registers the specified subclass with the given name in the factory

Parameters:

• name (str) –

  name to be used while registering the specified subclass

• subclass (Type[Preprocessor]) –

  the subclass of preprocessor to be registered

Source code in src/oqd_heisenberg_ion/common/preprocessor/factory.py

def register(cls, name, subclass):
    """
    registers the specified subclass with the given name in the factory

    Args:
        name (str): name to be used while registering the specified subclass
        subclass (Type[Preprocessor]): the subclass of preprocessor to be registered
    """

    cls.registry[name] = subclass

create

creates an instance of the requested subclass

Parameters:

• name (str) –

  requested subclass name

• parameter_set_list (list[dict]) –

  list of parameter sets, each set specified as a dict, needed by the preprocessor constructor

Raises:

• Exception –

  if requested Preprocessor subclass is not found

Returns:

• Preprocessor –

  instance of Preprocessor subclass requested

Source code in src/oqd_heisenberg_ion/common/preprocessor/factory.py

def create(cls, name, parameter_set_list):
    """
    creates an instance of the requested subclass

    Args:
        name (str): requested subclass name
        parameter_set_list (list[dict]): list of parameter sets, each set specified as a dict, needed by the preprocessor constructor

    Raises:
        Exception: if requested Preprocessor subclass is not found

    Returns:
        (Preprocessor): instance of Preprocessor subclass requested
    """

    if name not in cls.registry:
        raise Exception(f"Preprocessor implementation not found for name: {name}")
    else:
        return cls.registry[name](parameter_set_list)