data_structs.py

"""Data structures to ensure consistency during serialization for databases.

"""
import os
import logging
import json
from easyvvuq import constants
from easyvvuq.utils.helpers import easyvvuq_serialize
import numpy

__copyright__ = """

    Copyright 2018 Robin A. Richardson, David W. Wright

    This file is part of EasyVVUQ

    EasyVVUQ is free software: you can redistribute it and/or modify
    it under the terms of the Lesser GNU General Public License as published by
    the Free Software Foundation, either version 3 of the License, or
    (at your option) any later version.

    EasyVVUQ is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    Lesser GNU General Public License for more details.

    You should have received a copy of the Lesser GNU General Public License
    along with this program.  If not, see <https://www.gnu.org/licenses/>.

"""
__license__ = "LGPL"

logger = logging.getLogger(__name__)


def check_local_dir(path, dir_type='campaign'):
    """
    Check that local path exists and if not create it.

    Parameters
    ----------
    path : str
        Directory location to check.
    dir_type : str, default='campaign'
        Type of directory we are checking (used for user and debugging
        information.)

    Returns
    -------

    """

    if not os.path.isdir(path):

        if os.path.exists(path):
            logger.critical(f'{path} specified as {dir_type} directory '
                            f'for local run but is not a directory.')
            raise IOError(f'Invalid {dir_type} directory')
        else:
            os.makedirs(path)


def check_reference(ref, run_name, ref_type='campaign'):
    """
    Validation check for a `RunInfo` reference. Checks that an integer value
    has been passed to use as a reference to another 'table' - i.e. to a
    specific campaign, app or sampler.

    Parameters
    ----------
    ref : int
        Reference to be checked.
    run_name : str
        Name of run for which the check is being performed (user info/
        debugging).
    ref_type : str, default='campaign'
        Are we checking for a campaign, sampler or app (user info/
        debugging).

    Returns
    -------

    """

    if ref is None:
        message = f'No {ref_type} id specified for run {run_name}'
        logger.critical(message)
        raise RuntimeError(message)

    if not isinstance(ref, int):
        message = (f'Invalid {ref_type} id ({ref}) specified for '
                   f'run {run_name}')
        logger.critical(message)
        raise RuntimeError(message)


class RunInfo:
    """Handles information for individual application runs.

    Parameters
    ----------
    run_name : str
        Human readable name of the run.
    app : None or int
        ID of the associated application.
    params : None or dict
        Dictionary of parameter values for this run.
    sample: None or int
        ID of the sampler that created the run.
    campaign: None or int
        ID of the associated campaign.

    Attributes
    ----------
    campaign : int
        ID of the associated campaign.
    sample : int
        ID of the sampler that created the run.
    app : int
        ID of the associated application.
    run_name : str
        Human readable name of the run.
    status : enum(Status)
    """

    def __init__(
            self,
            run_name=None,
            run_dir=None,
            app=None,
            params=None,
            sample=None,
            campaign=None,
            status=constants.Status.NEW):

        self.campaign = campaign
        self.sample = sample
        self.app = app
        self.run_name = run_name
        self.run_dir = run_dir

        if not params:
            message = f'No run configuration specified for run {run_name}'
            raise RuntimeError(message)

        self.params = params
        self.status = status

        self.iteration = 0

    def to_dict(self, flatten=False):
        """Convert to a dictionary (optionally flatten to single level)

        Parameters
        ----------
        flatten : bool
            Should the return dictionary be single level (i.e. should `params`
            or other dictionary variables be serialized).

        Returns
        -------
        dict
            Dictionary representing the run - if flattened then params are
            returned as a JSON format sting.
        """

        def convert_nonserializable(obj):
            if isinstance(obj, numpy.int64):
                return int(obj)
            raise TypeError('Unknown type:', type(obj))

        if flatten:

            out_dict = {
                'run_name': self.run_name,
                'run_dir': self.run_dir,
                'params': json.dumps(self.params, default=convert_nonserializable),
                'status': constants.Status(self.status),
                'campaign': self.campaign,
                'sampler': self.sample,
                'app': self.app,
                'iteration': self.iteration,
            }

        else:

            out_dict = {
                'run_name': self.run_name,
                'run_dir': self.run_dir,
                'params': self.params,
                'status': constants.Status(self.status),
                'campaign': self.campaign,
                'sampler': self.sample,
                'app': self.app,
                'iteration': self.iteration,
            }

        return out_dict


class AppInfo:
    """Handles information for particular application.

    Attributes
    ----------
    name : str or None
        Human readable application name.
    paramsspec : ParamsSpecification or None
        Description of possible parameter values.
    """

    def __init__(
            self,
            name=None,
            paramsspec=None,
            actions=None):

        self.name = name
        self.paramsspec = paramsspec
        self.actions = actions

    def to_dict(self, flatten=False):
        """Convert to a dictionary (optionally flatten to single level)

        Parameters
        ----------
        flatten : bool
            Should the return dictionary be single level (i.e. should `paramsspec`
            be serialized).

        Returns
        -------
        dict
            Dictionary representing the application- if flattened then `paramsspec`
            is returned as a JSON format sting.
        """

        if flatten:

            out_dict = self.to_dict()

            out_dict['params'] = self.paramsspec.serialize()
        else:

            out_dict = {
                'name': self.name,
                'params': self.paramsspec,
                'actions': easyvvuq_serialize(self.actions),
            }

        return out_dict


class CampaignInfo:
    """Handles information on Campaign.

    Parameters
    ----------
    name : str or None
        Human readable campaign name.
    easyvvuq_version : str or None
        Version of EasyVVUQ used to create the campaign.
    campaign_dir_prefix : str or None
        Prefix test for campaign directory.
    campaign_dir : str or None,
        Path to the campaign directory.
    runs_dir : str or None
        path to run directory (within the campaign directory)
    local : bool, default=False
        Is this campaign designed to be created and executed on the same
        machine?

    Attributes
    ----------
    name : str or None
        Human readable campaign name.
    easyvvuq_version : str or None
        Version of EasyVVUQ used to create the campaign.
    campaign_dir_prefix : str or None
        Prefix test for campaign directory.
    campaign_dir : str or None,
        Path to the campaign directory.
    runs_dir : str or None
        path to run directory (within the campaign directory)
    """

    def __init__(self, name=None, easyvvuq_version=None,
                 campaign_dir_prefix=None, campaign_dir=None,
                 runs_dir=None, local=False):

        if name is None:
            message = "CampaignInfo constructor must be passed a 'name'."
            logger.critical(message)
            raise RuntimeError(message)

        if campaign_dir is None:
            message = "CampaignInfo constructor must be passed 'campaign_dir'"
            logger.critical(message)
            raise RuntimeError(message)

        self.name = name
        self.campaign_dir_prefix = campaign_dir_prefix

        self.easyvvuq_version = easyvvuq_version

        # TODO: think about right location for path check for remote runs
        if local:
            check_local_dir(campaign_dir)

        self.campaign_dir = campaign_dir

        if runs_dir is None:
            runs_dir = os.path.join(campaign_dir, 'runs')

        if local:
            check_local_dir(runs_dir, 'runs')

        self.runs_dir = runs_dir

    @property
    def easyvvuq_version(self):
        return self._easyvvuq_version

    @easyvvuq_version.setter
    def easyvvuq_version(self, version_no):
        # TODO: check validity and compatibility
        self._easyvvuq_version = version_no

    def to_dict(self, flatten=False):
        """Convert this to a dictionary

        Parameters
        ----------
        flatten : bool
            Should the return dictionary be single level (always true here).

        Returns
        -------
        dict
            Dictionary representing the campaign.
        """

        out_dict = {
            'name': self.name,
            'campaign_dir': self.campaign_dir,
            'campaign_dir_prefix': self.campaign_dir_prefix,
            'runs_dir': self.runs_dir,
            'easyvvuq_version': self.easyvvuq_version
        }

        return out_dict