Source code for pipeline.hsd.tasks.common.common

"""
This module defines common single dish Results and task container Classes.

Classes and Methods:
    absolute_path: A method to convert a path to an absolute path.
    SingleDishResults: Common single dish Results class.
    ParameterContainerJob: A container class to store task with parameters. 
"""
import os

from typing import Any, Optional

import pipeline.infrastructure as infrastructure
import pipeline.infrastructure.basetask as basetask

LOG = infrastructure.get_logger(__name__)


[docs]def absolute_path(name: str) -> str: """Return an absolute path of a given file.""" return os.path.abspath(os.path.expanduser(os.path.expandvars(name)))
[docs]class SingleDishResults(basetask.Results): """ Single Dish Results class. This class inherits a common Results class, pipeline.infrastructre.basetask.Results. See the documentation of the super class for details. The class is usually used as a superclass of task specific Results class. Attributes: task: A task class associated with this class. success: A boolean to indicate if the task execution was successful (True) or not (False). outcome: Outcome of the task. error: A set of strings to store error messages. """ def __init__(self, task: Optional[basetask.StandardTaskTemplate]=None, success: Optional[bool]=None, outcome: Any=None): """ Initialize class attributes and super class. Args: task: A taks class object. success: True if the task completes successfully. Otherwise, False. outcome: Outcome of the task. """ super(SingleDishResults, self).__init__() self.task = task self.success = success self.outcome = outcome self.error = set()
[docs] def merge_with_context(self, context: infrastructure.launcher.Context): """ Merge these results with the given context. See the documenetation of super class for more details. """ self.error.clear()
def _outcome_name(self) -> str: """Return an absolute path of outcome stored as a file.""" # usually, outcome is a name of the file return absolute_path(self.outcome) def _get_outcome(self, key: str) -> Any: """ Return a value in outcome dictionary. Args: key: A key of outcome dictionary to get a value. Retruns: The value in outcome dictionary obtained by a given key. This method returns None, if the outcome is not a dictionary. """ if isinstance(self.outcome, dict): return self.outcome.get(key, None) else: return None def __repr__(self) -> str: """Return a printable representation of the class.""" # taskname = self.task if hasattr(self,'task') else 'none' s = '%s:\n\toutcome is %s' % (self.__class__.__name__, self._outcome_name()) return s
# TODO(ksugimoto) THIS CLASS WILL BE REMOVED SHORTLY. # TODO(ksugimoto) IGNORE IN CODE REVIEW.
[docs]class ParameterContainerJob(object): """ A container class to store task class object with parameter values. Attributes: task: A task class object to execute. parameters: A disctionary of task parameters (key: parameter name, value: parameter value) to invoke task. """ def __init__(self, task: basetask.StandardTaskTemplate, **parameters): """ Initialize class attributes. Args: task: Task to be executed. **parameters: Task parameters and their values to invoke task. """ self.task = task self.parameters = parameters
[docs] def execute(self, dry_run: bool=True) -> basetask.Results: """ Invoke task using given parameters. Args: dry_run: Run task if False, or only display commands if True. Retruns: A Result object of the task. """ result = self.task.execute(dry_run, **self.parameters) return result