Skip to content

Studyenvironment

Classes that represent the environment of a study.

StudyEnvironment

StudyEnvironment for managing a study environment.

The StudyEnvironment provides the context where all study steps can find variables, sources, dependencies, etc.

Source code in maestrowf/datastructures/core/studyenvironment.py 
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
class StudyEnvironment:
    """
    StudyEnvironment for managing a study environment.

    The StudyEnvironment provides the context where all study
    steps can find variables, sources, dependencies, etc.
    """

    def __init__(self):
        """Initialize an empty StudyEnvironment."""
        # Types of environment objects.
        self.substitutions = {}
        self.labels = {}
        self.sources = []
        self.dependencies = {}

        # Private members
        self._tokens = set()
        self._names = set()
        # Boolean that tracks if dependencies have been acquired.
        self._is_set_up = False

        LOGGER.debug("Initialized an empty StudyEnvironment.")

    def __bool__(self):
        """
        Override for the __bool__ operator.

        :returns: True if the StudyEnvironment instance has values, False
            otherwise.
        """
        return bool(self._names)

    @property
    def is_set_up(self):
        """
        Check that the StudyEnvironment is set up.

        :returns: True is the instance is set up, False otherwise.
        """
        return self._is_set_up

    def add(self, item):
        """
        Add the item parameter to the StudyEnvironment.

        :param item: EnvObject to be added to the environment.
        """
        # TODO: Need to revist this to make this better. A label can get lost
        # because the necessary variable could have not been added yet
        # and there's too much of a need to process a dependency first.
        name = None
        LOGGER.debug("Calling add with %s", str(item))
        if isinstance(item, Dependency):
            LOGGER.debug("Adding %s of type %s.", item.name, type(item))
            LOGGER.debug("Value: %s.", item.__dict__)
            self.dependencies[item.name] = item
            name = item.name
            self._is_set_up = False
        elif isinstance(item, Substitution):
            LOGGER.debug("Value: %s", item.value)
            LOGGER.debug("Tokens: %s", self._tokens)
            name = item.name
            LOGGER.debug("Adding %s of type %s.", item.name, type(item))
            if (
                    isinstance(item.value, str) and
                    any(token in item.value for token in self._tokens)):
                LOGGER.debug("Label detected. Adding %s to labels", item.name)
                self.labels[item.name] = item
            else:
                self._tokens.add(item.token)
                self.substitutions[item.name] = item
        elif isinstance(item, Source):
            LOGGER.debug("Adding source %s", item.source)
            LOGGER.debug("Item source: %s", item.source)
            self.sources.append(item)
        else:
            error = "Received an item of type {}. Expected an item of base " \
                    "type Substitution, Source, or Dependency." \
                    .format(type(item))
            LOGGER.exception(error)
            raise TypeError(error)

        if name and name in self._names:
            error = "A duplicate name '{}' has been detected. All names " \
                    "must be unique. Aborting.".format(name)
            LOGGER.exception(error)
            raise ValueError(error)
        else:
            LOGGER.debug("{} added to set of names.".format(name))
            self._names.add(name)

    def find(self, key):
        """
        Find the environment object labeled by the specified key.

        :param key: Name of the environment object to find.
        :returns: The environment object labeled by key, None if key is not
            found.
        """
        LOGGER.debug("Looking for '%s'...", key)
        if key in self.dependencies:
            LOGGER.debug("Found '%s' in environment dependencies.", key)
            return self.dependencies[key]

        if key in self.substitutions:
            LOGGER.debug("Found '%s' in environment substitutions.", key)
            return self.substitutions[key]

        if key in self.labels:
            LOGGER.debug("Found '%s' in environment labels.", key)
            return self.labels[key]

        LOGGER.debug("'%s' not found -- \n%s", key, self)
        return None

    def remove(self, key):
        """
        Remove the environment object labeled by the specified key.

        :param key: Name of the environment object to remove.
        :returns: The environment object labeled by key.
        """
        LOGGER.debug("Looking to remove '%s'...", key)

        if key not in self._names:
            return None

        _ = self.dependencies.pop(key, None)
        if _ is not None:
            self._names.remove(key)
            return _

        _ = self.substitutions.pop(key, None)
        if _ is not None:
            self._names.remove(key)
            return _

        _ = self.labels.pop(key, None)
        if _ is not None:
            self._names.remove(key)
            return _

        LOGGER.debug("'%s' not found -- \n%s", key, self)
        return None

    def acquire_environment(self):
        """Acquire any environment items that may be stored remotely."""
        if self._is_set_up:
            LOGGER.info("Environment already set up. Returning.")
            return

        LOGGER.debug("Acquiring dependencies")
        for dependency, value in self.dependencies.items():
            LOGGER.info("Acquiring -- %s", dependency)
            value.acquire(substitutions=self.substitutions.values())

        self._is_set_up = True

    def apply_environment(self, item):
        """
        Apply the environment to the specified item.

        :param item: String to apply environment to.
        :returns: String with the environment applied.
        """
        if not item:
            return item

        LOGGER.debug("Applying environment to %s", item)
        LOGGER.debug("Processing labels...")
        for label, value in self.labels.items():
            LOGGER.debug("Looking for %s in %s", label, item)
            item = value.substitute(item)
            LOGGER.debug("After substitution: %s", item)

        LOGGER.debug("Processing dependencies...")
        for label, dependency in self.dependencies.items():
            LOGGER.debug("Looking for %s in %s", label, item)
            item = dependency.substitute(item)
            LOGGER.debug("After substitution: %s", item)
            LOGGER.debug("Acquiring %s.", label)

        LOGGER.debug("Processing substitutions...")
        for substitution, value in self.substitutions.items():
            LOGGER.debug("Looking for %s in %s", substitution, item)
            item = value.substitute(item)
            LOGGER.debug("After substitution: %s", item)

        return item

is_set_up property

Check that the StudyEnvironment is set up.

Returns:

Type Description

True is the instance is set up, False otherwise.

__bool__()

Override for the bool operator.

Returns:

Type Description

True if the StudyEnvironment instance has values, False otherwise.
Source code in maestrowf/datastructures/core/studyenvironment.py 
63
64
65
66
67
68
69
70
def __bool__(self):
    """
    Override for the __bool__ operator.

    :returns: True if the StudyEnvironment instance has values, False
        otherwise.
    """
    return bool(self._names)

__init__()

Initialize an empty StudyEnvironment.

Source code in maestrowf/datastructures/core/studyenvironment.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def __init__(self):
    """Initialize an empty StudyEnvironment."""
    # Types of environment objects.
    self.substitutions = {}
    self.labels = {}
    self.sources = []
    self.dependencies = {}

    # Private members
    self._tokens = set()
    self._names = set()
    # Boolean that tracks if dependencies have been acquired.
    self._is_set_up = False

    LOGGER.debug("Initialized an empty StudyEnvironment.")

acquire_environment()

Acquire any environment items that may be stored remotely.

Source code in maestrowf/datastructures/core/studyenvironment.py 
185
186
187
188
189
190
191
192
193
194
195
196
def acquire_environment(self):
    """Acquire any environment items that may be stored remotely."""
    if self._is_set_up:
        LOGGER.info("Environment already set up. Returning.")
        return

    LOGGER.debug("Acquiring dependencies")
    for dependency, value in self.dependencies.items():
        LOGGER.info("Acquiring -- %s", dependency)
        value.acquire(substitutions=self.substitutions.values())

    self._is_set_up = True

add(item)

Add the item parameter to the StudyEnvironment.

Parameters:

Name Type Description Default
item

EnvObject to be added to the environment.

 required
Source code in maestrowf/datastructures/core/studyenvironment.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def add(self, item):
    """
    Add the item parameter to the StudyEnvironment.

    :param item: EnvObject to be added to the environment.
    """
    # TODO: Need to revist this to make this better. A label can get lost
    # because the necessary variable could have not been added yet
    # and there's too much of a need to process a dependency first.
    name = None
    LOGGER.debug("Calling add with %s", str(item))
    if isinstance(item, Dependency):
        LOGGER.debug("Adding %s of type %s.", item.name, type(item))
        LOGGER.debug("Value: %s.", item.__dict__)
        self.dependencies[item.name] = item
        name = item.name
        self._is_set_up = False
    elif isinstance(item, Substitution):
        LOGGER.debug("Value: %s", item.value)
        LOGGER.debug("Tokens: %s", self._tokens)
        name = item.name
        LOGGER.debug("Adding %s of type %s.", item.name, type(item))
        if (
                isinstance(item.value, str) and
                any(token in item.value for token in self._tokens)):
            LOGGER.debug("Label detected. Adding %s to labels", item.name)
            self.labels[item.name] = item
        else:
            self._tokens.add(item.token)
            self.substitutions[item.name] = item
    elif isinstance(item, Source):
        LOGGER.debug("Adding source %s", item.source)
        LOGGER.debug("Item source: %s", item.source)
        self.sources.append(item)
    else:
        error = "Received an item of type {}. Expected an item of base " \
                "type Substitution, Source, or Dependency." \
                .format(type(item))
        LOGGER.exception(error)
        raise TypeError(error)

    if name and name in self._names:
        error = "A duplicate name '{}' has been detected. All names " \
                "must be unique. Aborting.".format(name)
        LOGGER.exception(error)
        raise ValueError(error)
    else:
        LOGGER.debug("{} added to set of names.".format(name))
        self._names.add(name)

apply_environment(item)

Apply the environment to the specified item.

Parameters:

Name Type Description Default
item

String to apply environment to.

 required

Returns:

Type Description

String with the environment applied.
Source code in maestrowf/datastructures/core/studyenvironment.py 
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def apply_environment(self, item):
    """
    Apply the environment to the specified item.

    :param item: String to apply environment to.
    :returns: String with the environment applied.
    """
    if not item:
        return item

    LOGGER.debug("Applying environment to %s", item)
    LOGGER.debug("Processing labels...")
    for label, value in self.labels.items():
        LOGGER.debug("Looking for %s in %s", label, item)
        item = value.substitute(item)
        LOGGER.debug("After substitution: %s", item)

    LOGGER.debug("Processing dependencies...")
    for label, dependency in self.dependencies.items():
        LOGGER.debug("Looking for %s in %s", label, item)
        item = dependency.substitute(item)
        LOGGER.debug("After substitution: %s", item)
        LOGGER.debug("Acquiring %s.", label)

    LOGGER.debug("Processing substitutions...")
    for substitution, value in self.substitutions.items():
        LOGGER.debug("Looking for %s in %s", substitution, item)
        item = value.substitute(item)
        LOGGER.debug("After substitution: %s", item)

    return item

find(key)

Find the environment object labeled by the specified key.

Parameters:

Name Type Description Default
key

Name of the environment object to find.

 required

Returns:

Type Description

The environment object labeled by key, None if key is not found.
Source code in maestrowf/datastructures/core/studyenvironment.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def find(self, key):
    """
    Find the environment object labeled by the specified key.

    :param key: Name of the environment object to find.
    :returns: The environment object labeled by key, None if key is not
        found.
    """
    LOGGER.debug("Looking for '%s'...", key)
    if key in self.dependencies:
        LOGGER.debug("Found '%s' in environment dependencies.", key)
        return self.dependencies[key]

    if key in self.substitutions:
        LOGGER.debug("Found '%s' in environment substitutions.", key)
        return self.substitutions[key]

    if key in self.labels:
        LOGGER.debug("Found '%s' in environment labels.", key)
        return self.labels[key]

    LOGGER.debug("'%s' not found -- \n%s", key, self)
    return None

remove(key)

Remove the environment object labeled by the specified key.

Parameters:

Name Type Description Default
key

Name of the environment object to remove.

 required

Returns:

Type Description

The environment object labeled by key.
Source code in maestrowf/datastructures/core/studyenvironment.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
def remove(self, key):
    """
    Remove the environment object labeled by the specified key.

    :param key: Name of the environment object to remove.
    :returns: The environment object labeled by key.
    """
    LOGGER.debug("Looking to remove '%s'...", key)

    if key not in self._names:
        return None

    _ = self.dependencies.pop(key, None)
    if _ is not None:
        self._names.remove(key)
        return _

    _ = self.substitutions.pop(key, None)
    if _ is not None:
        self._names.remove(key)
        return _

    _ = self.labels.pop(key, None)
    if _ is not None:
        self._names.remove(key)
        return _

    LOGGER.debug("'%s' not found -- \n%s", key, self)
    return None