Source code for yggdrasil.drivers.DummyModelDriver

from collections import OrderedDict
from yggdrasil.drivers.InterpretedModelDriver import InterpretedModelDriver

[docs]class DummyModelDriver(InterpretedModelDriver): r"""Class that stands-in to act as a model utilizing unmatched input/output channels.""" executable_type = 'other' language = 'dummy' full_language = False base_languages = ['python'] language_ext = [] no_executable = True comms_implicit = True def __init__(self, *args, **kwargs): self.runner = kwargs['runner'] super(DummyModelDriver, self).__init__(*args, **kwargs)
[docs] @classmethod def is_language_installed(self): r"""Determine if this model driver is installed on the current machine. Returns: bool: Truth of if this model driver can be run on the current machine. """ # This is being run so python exists return True
[docs] @classmethod def configuration_steps(cls): r"""Get a list of configuration steps with tuples of flags and boolean values. Returns: OrderedDict: Pairs of descriptions and states for different steps in the configuration all steps must be True for the language to be configured. """ return OrderedDict()
[docs] @classmethod def language_version(cls, **kwargs): r"""Determine the version of this language. Args: **kwargs: Keyword arguments are passed to cls.run_executable. Returns: str: Version of compiler/interpreter for this language. """ return '0'
[docs] def before_start(self): r"""Actions to perform before the run starts.""" pass
[docs] def before_loop(self): r"""Actions before loop.""" pass
[docs] def run_loop(self): r"""Loop to check if model is still running and forward output.""" for drv in self.runner.modeldrivers.values(): if (drv['name'] != and drv['instance'].is_alive(): self.wait_flag_attr('break_flag', timeout=1.0) return self.set_break_flag()
@property def service_partner(self): r"""dict: YAML representation of the dummy model that should stand-in for the model client-side.""" out = {'inputs': [], 'outputs': []} dir2opp = {'input': 'output', 'output': 'input'} for io1, io2 in dir2opp.items(): for drv in self.yml['%s_drivers' % io1]: name = drv[io1 + 's'][0]['name'] comm = getattr(drv['instance'], '%scomm' % io2[0]) x = comm.opp_comm_kwargs(for_yaml=True) x['name'] = name.split( drv[io1 + 's'][0]['partner_model'] + ':')[-1] out[io2 + 's'].append(x) if drv['instance']._connection_type.startswith('rpc_'): assert io2 == 'input' out[io1 + 's'].append({'name': x['name'] + '_response'}) out['is_server'] = {io2: x['name'], io1: x['name'] + '_response'} for io1, io2 in dir2opp.items(): assert out[io2 + 's'] # TODO: Is there a case where a DummyModelDriver will be created # for a model that does not have inputs or outputs? # if not out[io2 + 's']: # out.pop(io2 + 's') return out