Source code for pacman.executor.token_states
# Copyright (c) 2017-2019 The University of Manchester
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program 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
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from six import iteritems
class _TokenState(object):
""" Determines whether a token has been fulfilled or not
"""
__slots__ = [
# The parts of the token that have not yet completed
"_incomplete_parts",
# The parts of the token that have completed
"_complete_parts"
]
def __init__(self):
self._incomplete_parts = set()
self._complete_parts = set()
def is_tracking_token_part(self, part):
""" Determine if the given part is being tracked
"""
return part in self._incomplete_parts or part in self._complete_parts
def track_token_part(self, part):
""" Indicates that this state should start tracking the completion\
of a part of a token. Part can be None to indicate the tracking\
of a whole token.
"""
self._incomplete_parts.add(part)
def complete_token_part(self, part):
""" Indicates that a part of this token has completed. If part is\
None, indicates that the whole token has completed.
"""
if part is None:
self._complete_parts.update(self._incomplete_parts)
self._incomplete_parts = set()
self._complete_parts.add(part)
self._incomplete_parts.discard(part)
def is_complete(self, part=None):
""" If part is None, true if all parts have completed, otherwise\
checks for completion of a specific part of the token
"""
if part is None:
return not self._incomplete_parts
return part in self._complete_parts
[docs]class TokenStates(object):
""" Keeps track of multiple token state objects to determine if they\
are complete
"""
__slots__ = [
# The tokens being tracked
"_tokens"
]
def __init__(self):
self._tokens = dict()
[docs] def is_tracking_token(self, token):
""" Determine if the token is being tracked
"""
if token.name not in self._tokens:
return False
if token.part is None:
return True
return self._tokens[token.name].is_tracking_token_part(token.part)
[docs] def track_token(self, token):
""" Start tracking a token
"""
if token.name not in self._tokens:
self._tokens[token.name] = _TokenState()
self._tokens[token.name].track_token_part(token.part)
[docs] def process_output_token(self, output_token):
""" Process an output token marking a process or part as complete
"""
if output_token.name in self._tokens:
self._tokens[output_token.name].complete_token_part(
output_token.part)
[docs] def is_token_complete(self, token):
""" True if the given token has completed
"""
# If the token is not tracked, is assumed to be complete
if token.name not in self._tokens:
return False
# The token is complete if the token part is complete or the token
# as a whole is complete
token_state = self._tokens[token.name]
return token_state.is_complete(token.part) or token_state.is_complete()
[docs] def get_completed_tokens(self):
""" Get a list of tokens that have been completed
"""
return [
name for name, token in iteritems(self._tokens)
if token.is_complete()
]