import pytest
from ymmsl import load
from ymmsl.v0_2 import Conduit, Model, Reference, resolve
from libmuscle.manager.hammer import Plate, flatten
[docs]
@pytest.fixture
def c1() -> Conduit:
return Conduit('macro.out', 'micro.in')
[docs]
@pytest.fixture
def c2() -> Conduit:
return Conduit('micro.in', 'micro.program.in')
[docs]
@pytest.fixture
def c3() -> Conduit:
return Conduit('micro.program.out', 'micro.out')
[docs]
@pytest.fixture
def c4() -> Conduit:
return Conduit('micro.out', 'macro.in')
[docs]
@pytest.fixture
def c5() -> Conduit:
return Conduit('micro.in', 'micro.program2.in')
[docs]
@pytest.fixture
def plate(c1: Conduit, c2: Conduit, c3: Conduit, c4: Conduit, c5: Conduit) -> Plate:
plate = Plate()
plate.add(c1, True, False)
plate.add(c2, False, True)
plate.add(c3, True, False)
plate.add(c4, False, True)
plate.add(c5, False, True)
return plate
[docs]
def has_conduit(model: Model, sender: str, receiver: str, filters: str = '') -> bool:
"""Helper function for checking whether a model has a given conduit.
They're in a list, this makes it easier to search, although still inefficient.
Filters are given in a single string, separated by single spaces, e.g. 'last pad'.
"""
for conduit in model.conduits:
if conduit.sender == sender and conduit.receiver == receiver:
cfs = ' '.join([f.value for f in conduit.filters])
if cfs == filters:
return True
return False
[docs]
def test_plate(plate, c2, c5) -> None:
macro_out_conduits = plate.pop_by_sender(Reference('micro.in'))
assert macro_out_conduits == {
Reference('micro.program.in'): (c2, True),
Reference('micro.program2.in'): (c5, True)}
# check that they were removed
assert len(plate.pop_by_sender(Reference('micro.in'))) == 0
[docs]
def test_plate2(plate, c1) -> None:
micro2_in_conduits = plate.pop_by_receiver(Reference('micro.in'))
assert micro2_in_conduits == {Reference('macro.out'): (c1, True)}
# check that it was removed
assert len(plate.pop_by_receiver(Reference('micro.in'))) == 0
[docs]
def test_flatten_simple() -> None:
nested_config_yaml = (
'ymmsl_version: v0.2\n'
'description: Simple nested test configuration\n'
'models:\n'
' outer:\n'
' description: Model containing a submodel\n'
' components:\n'
' init:\n'
' ports:\n'
' o_f: init_out\n'
' description: Creates initial conditions\n'
' implementation: p1\n'
' sim:\n'
' ports:\n'
' f_init: init_in\n'
' description: Simulates stuff\n'
' implementation: inner\n'
' conduits:\n'
' init.init_out: sim.init_in\n'
' inner:\n'
' ports:\n'
' f_init: init_in\n'
' description: Nested simulation model\n'
' components:\n'
' simulate:\n'
' ports:\n'
' f_init: in\n'
' description: Simulation program\n'
' implementation: p2\n'
' conduits:\n'
' init_in: simulate.in\n'
'programs:\n'
' p1:\n'
' description: Program that calculates initial conditions\n'
' executable: /home/user/codes/p1\n'
' p2:\n'
' description: Program that simulates stuff\n'
' executable: /home/user/codes/p2\n'
)
nested_config = load(nested_config_yaml)
flat_config = flatten(nested_config)
assert len(flat_config.models) == 1
assert 'outer' in flat_config.models
model = flat_config.root_model()
assert model.name == 'outer'
assert len(model.components) == 2
assert 'init' in model.components
assert 'sim.simulate' in model.components
assert len(model.conduits) == 1
assert has_conduit(model, 'init.init_out', 'sim.simulate.in')
[docs]
def test_flatten_deep() -> None:
nested_config_yaml = (
'ymmsl_version: v0.2\n'
'description: Three-deep nested test configuration\n'
'models:\n'
' top:\n'
' description: Model at the top\n'
' components:\n'
' c1:\n'
' ports:\n'
' o_i: out\n'
' s: in\n'
' description: Calls c2\n'
' implementation: p1\n'
' c2:\n'
' ports:\n'
' f_init: init_in\n'
' o_f: final_out\n'
' description: Submodel\n'
' implementation: middle\n'
' conduits:\n'
' c1.out: c2.init_in\n'
' c2.final_out: c1.in\n'
' middle:\n'
' ports:\n'
' f_init: init_in\n'
' o_f: final_out\n'
' description: Middle simulation model, empty shell really\n'
' components:\n'
' c3:\n'
' ports:\n'
' f_init: init_in\n'
' o_f: final_out\n'
' description: Forward again...\n'
' implementation: bottom\n'
' conduits:\n'
' init_in: c3.init_in\n'
' c3.final_out: final_out\n'
' bottom:\n'
' ports:\n'
' f_init: init_in\n'
' o_f: final_out\n'
' description: Bottom simulation model\n'
' components:\n'
' c4:\n'
' ports:\n'
' f_init: init_in\n'
' o_f: final_out\n'
' description: Actual simulation model\n'
' implementation: p2\n'
' conduits:\n'
' init_in: c4.init_in\n'
' c4.final_out: final_out\n'
'programs:\n'
' p1:\n'
' description: Program that calculates initial conditions\n'
' executable: /home/user/codes/p1\n'
' p2:\n'
' description: Program that simulates stuff\n'
' executable: /home/user/codes/p2\n'
)
nested_config = load(nested_config_yaml)
flat_config = flatten(nested_config)
assert len(flat_config.models) == 1
assert 'top' in flat_config.models
model = flat_config.root_model()
assert model.name == 'top'
assert len(model.components) == 2
assert 'c1' in model.components
assert 'c2.c3.c4' in model.components
assert len(model.conduits) == 2
assert has_conduit(model, 'c1.out', 'c2.c3.c4.init_in')
assert has_conduit(model, 'c2.c3.c4.final_out', 'c1.in')
[docs]
def test_flatten_nested_ensemble() -> None:
nested_config_yaml = (
'ymmsl_version: v0.2\n'
'description: Three-deep nested test configuration\n'
'models:\n'
' top:\n'
' description: Model at the top\n'
' components:\n'
' c1:\n'
' ports:\n'
' o_i: out\n'
' s: in\n'
' description: Calls c2\n'
' implementation: p1\n'
' c2:\n'
' ports:\n'
' f_init: init_in\n'
' o_f: final_out\n'
' description: Submodel\n'
' implementation: middle\n'
' multiplicity: 10\n'
' conduits:\n'
' c1.out: c2.init_in\n'
' c2.final_out: c1.in\n'
' middle:\n'
' ports:\n'
' f_init: init_in\n'
' o_f: final_out\n'
' description: This should get replicated\n'
' components:\n'
' c3:\n'
' ports:\n'
' f_init: init_in\n'
' o_f: final_out\n'
' description: Another subcomponent\n'
' implementation: nested\n'
' multiplicity: 5\n'
' conduits:\n'
' init_in: c3.init_in\n'
' c3.final_out: final_out\n'
' nested:\n'
' ports:\n'
' f_init: init_in\n'
' o_f: final_out\n'
' description: This should get replicated, nested\n'
' components:\n'
' c4:\n'
' ports:\n'
' f_init: init_in\n'
' o_f: final_out\n'
' description: First component\n'
' implementation: p2\n'
' c5:\n'
' ports:\n'
' f_init: init_in\n'
' o_f: final_out\n'
' description: Second component\n'
' implementation: p2\n'
' conduits:\n'
' init_in: c4.init_in\n'
' c4.final_out: c5.init_in\n'
' c5.final_out: final_out\n'
'programs:\n'
' p1:\n'
' description: Program that calculates initial conditions\n'
' executable: /home/user/codes/p1\n'
' p2:\n'
' description: Program that simulates stuff\n'
' executable: /home/user/codes/p2\n'
)
nested_config = load(nested_config_yaml)
flat_config = flatten(nested_config)
assert len(flat_config.models) == 1
assert 'top' in flat_config.models
model = flat_config.root_model()
assert model.name == 'top'
assert len(model.components) == 3
assert 'c1' in model.components
assert 'c2.c3.c4' in model.components
assert model.components['c2.c3.c4'].multiplicity == [10, 5]
assert 'c2.c3.c5' in model.components
assert model.components['c2.c3.c5'].multiplicity == [10, 5]
assert len(model.conduits) == 3
assert has_conduit(model, 'c2.c3.c4.final_out', 'c2.c3.c5.init_in')
assert has_conduit(model, 'c1.out', 'c2.c3.c4.init_in')
assert has_conduit(model, 'c2.c3.c5.final_out', 'c1.in')
[docs]
def test_flatten_conduit_filters() -> None:
nested_config_yaml = (
'ymmsl_version: v0.2\n'
'description: Macro-micro to macro-micro dispatch, nested\n'
'models:\n'
' macro_micro:\n'
' ports:\n'
' f_init: macro_state_in micro_state_in\n'
' o_f: macro_state_out micro_state_out\n'
' description: Macro-micro model\n'
' components:\n'
' macro:\n'
' ports:\n'
' f_init: state_in\n'
' o_i: boundary_out\n'
' s: boundary_in\n'
' o_f: state_out\n'
' description: |\n'
' Macro model that communicates its boundary conditions\n'
' while running, and outputs its state at the end.\n'
' implementation: p1\n'
' micro:\n'
' ports:\n'
' f_init: state_in boundary_in\n'
' o_f: state_out boundary_out\n'
' description: |\n'
' Stateful micro model that takes new boundary conditions and\n'
' optionally a new state on each run. If a new state is received\n'
' then it re-initialises, otherwise the current state is kept.\n'
' Re-equilibrates to the new boundary conditions, then outputs\n'
' boundary and state at the end of each run.\n'
' implementation: p2\n'
' conduits:\n'
' macro_state_in: macro.state_in\n'
' micro_state_in: pad micro.state_in\n'
' macro.boundary_out: micro.boundary_in\n'
' micro.boundary_out: macro.boundary_in\n'
' macro.state_out: macro_state_out\n'
' micro.state_out: last micro_state_out\n'
' init:\n'
' ports:\n'
' o_f: macro_state_out micro_state_out\n'
' description: Creates initial states\n'
' components:\n'
' init_macro:\n'
' ports:\n'
' o_f: state_out\n'
' description: Creates initial conditions for macro\n'
' implementation: p3\n'
' init_micro:\n'
' ports:\n'
' o_f: state_out\n'
' description: Creates initial conditions for micro\n'
' implementation: p4\n'
' conduits:\n'
' init_macro.state_out: macro_state_out\n'
' init_micro.state_out: micro_state_out\n'
' macro_micro_dispatch:\n'
' description: |\n'
' Creates initial conditions, the runs a model until conditions\n'
' change and that model doesn\'t apply anymore, after which we\n'
' pass the state to a second model to finish the calculation.\n'
' Actually uses the same model twice, but maybe with different\n'
' settings. Note that the model here is a closed box, we cannot\n'
' tell that there is a macro-micro inside, other than from the name\n'
' of the model and its ports.\n'
' components:\n'
' init:\n'
' ports:\n'
' o_f: macro_state_out micro_state_out\n'
' description: Calculates initial conditions\n'
' implementation: init\n'
' first:\n'
' ports:\n'
' f_init: macro_state_in micro_state_in\n'
' o_f: macro_state_out micro_state_out\n'
' description: First macro-micro model\n'
' implementation: macro_micro\n'
' second:\n'
' ports:\n'
' f_init: macro_state_in micro_state_in\n'
' description: First macro-micro model\n'
' implementation: macro_micro\n'
' conduits:\n'
' init.macro_state_out: first.macro_state_in\n'
' init.micro_state_out: first.micro_state_in\n'
' first.macro_state_out: second.macro_state_in\n'
' first.micro_state_out: second.micro_state_in\n'
'programs:\n'
' p1:\n'
' description: Implements macro model\n'
' executable: /home/user/codes/macro\n'
' p2:\n'
' description: Implements micro model\n'
' executable: /home/user/codes/micro\n'
' p3:\n'
' description: Program that calculates initial conditions for macro\n'
' executable: /home/user/codes/macro_ic\n'
' p4:\n'
' description: Program that calculates initial conditions for micro\n'
' executable: /home/user/codes/micro_ic\n'
)
nested_config = load(nested_config_yaml)
flat_config = flatten(nested_config)
assert len(flat_config.models) == 1
assert 'macro_micro_dispatch' in flat_config.models
model = flat_config.root_model()
assert model.name == 'macro_micro_dispatch'
assert len(model.components) == 6
assert 'init.init_macro' in model.components
assert 'init.init_micro' in model.components
assert 'first.macro' in model.components
assert 'first.micro' in model.components
assert 'second.macro' in model.components
assert 'second.micro' in model.components
assert len(model.conduits) == 8
assert has_conduit(model, 'init.init_macro.state_out', 'first.macro.state_in')
assert has_conduit(
model, 'init.init_micro.state_out', 'first.micro.state_in', 'pad')
assert has_conduit(model, 'first.macro.boundary_out', 'first.micro.boundary_in')
assert has_conduit(model, 'first.micro.boundary_out', 'first.macro.boundary_in')
assert has_conduit(model, 'first.macro.state_out', 'second.macro.state_in')
assert has_conduit(
model, 'first.micro.state_out', 'second.micro.state_in', 'last pad')
assert has_conduit(model, 'second.macro.boundary_out', 'second.micro.boundary_in')
assert has_conduit(model, 'second.micro.boundary_out', 'second.macro.boundary_in')
[docs]
def test_flatten_multicast() -> None:
nested_config_yaml = (
'ymmsl_version: v0.2\n'
'description: Testing wiring of multicast conduits\n'
'models:\n'
' outer:\n'
' description: Model containing a submodel\n'
' components:\n'
' c1:\n'
' ports:\n'
' o_f: out\n'
' description: Sends some data\n'
' implementation: p1\n'
' c2:\n'
' ports:\n'
' f_init: in\n'
' o_f: out1 out2\n'
' description: Submodel\n'
' implementation: inner\n'
' c3:\n'
' ports:\n'
' f_init: in1 in2 in3\n'
' description: Receives some data\n'
' implementation: p3\n'
' c4:\n'
' ports:\n'
' f_init: in1 in2\n'
' description: Receives some data\n'
' implementation: p3\n'
' conduits:\n'
' c1.out:\n'
' - c2.in\n'
' - c3.in1\n'
' c2.out1:\n'
' - c3.in2\n'
' - c4.in1\n'
' c2.out2:\n'
' - c3.in3\n'
' - c4.in2\n'
' inner:\n'
' ports:\n'
' f_init: in\n'
' o_f: out1 out2\n'
' description: Nested simulation model\n'
' components:\n'
' c1:\n'
' ports:\n'
' f_init: in\n'
' o_f: out\n'
' description: Simulation program\n'
' implementation: p2\n'
' c2:\n'
' ports:\n'
' f_init: in\n'
' o_f: out\n'
' description: Simulation program\n'
' implementation: p2\n'
' conduits:\n'
' in:\n'
' - c1.in\n'
' - c2.in\n'
' c1.out:\n'
' - out1\n'
' - out2\n'
'programs:\n'
' p1:\n'
' description: Program that sends something\n'
' executable: /home/user/codes/p1\n'
' p2:\n'
' description: Program that simulates stuff\n'
' executable: /home/user/codes/p2\n'
' p3:\n'
' description: Program that receives some things\n'
' executable: /home/user/codes/p2\n'
)
nested_config = load(nested_config_yaml)
flat_config = flatten(nested_config)
assert len(flat_config.models) == 1
assert 'outer' in flat_config.models
model = flat_config.root_model()
assert model.name == 'outer'
assert len(model.components) == 5
assert 'c1' in model.components
assert 'c2.c1' in model.components
assert 'c2.c2' in model.components
assert 'c3' in model.components
assert 'c4' in model.components
assert len(model.conduits) == 7
assert has_conduit(model, 'c1.out', 'c3.in1')
assert has_conduit(model, 'c1.out', 'c2.c1.in')
assert has_conduit(model, 'c1.out', 'c2.c2.in')
assert has_conduit(model, 'c2.c1.out', 'c3.in2')
assert has_conduit(model, 'c2.c1.out', 'c4.in1')
assert has_conduit(model, 'c2.c1.out', 'c3.in3')
assert has_conduit(model, 'c2.c1.out', 'c4.in2')
[docs]
def test_flatten_passthrough_overload() -> None:
nested_config_yaml = (
'ymmsl_version: v0.2\n'
'description: Testing model passthroughs and custom implementations\n'
'models:\n'
' framework:\n'
' description: Model with a passthrough\n'
' components:\n'
' c1:\n'
' ports:\n'
' o_f: out\n'
' description: Sends some data\n'
' implementation: p1\n'
' c2:\n'
' ports:\n'
' f_init: in\n'
' o_f: out\n'
' description: Extension point\n'
' implementation: passthrough\n'
' c3:\n'
' ports:\n'
' f_init: in\n'
' description: Receives some data\n'
' implementation: p3\n'
' conduits:\n'
' c1.out: c2.in\n'
' c2.out: c3.in\n'
' passthrough:\n'
' ports:\n'
' f_init: in\n'
' o_f: out\n'
' description: Passes through the message\n'
' conduits:\n'
' in: out\n'
'programs:\n'
' p1:\n'
' description: Program that sends something\n'
' executable: /home/user/codes/p1\n'
' p2:\n'
' description: Program that simulates stuff\n'
' executable: /home/user/codes/p2\n'
' p3:\n'
' description: Program that receives some things\n'
' executable: /home/user/codes/p2\n'
)
nested_config = load(nested_config_yaml)
flat_config = flatten(nested_config)
assert len(flat_config.models) == 1
assert 'framework' in flat_config.models
model = flat_config.root_model()
assert model.name == 'framework'
assert len(model.components) == 2
assert 'c1' in model.components
assert 'c3' in model.components
assert len(model.conduits) == 1
assert has_conduit(model, 'c1.out', 'c3.in')
nested_config = load(nested_config_yaml)
nested_config.custom_implementations[Reference('framework.c2')] = Reference('p2')
resolve(Reference([]), nested_config) # apply custom_implementations
flat_config = flatten(nested_config)
assert len(flat_config.models) == 1
assert 'framework' in flat_config.models
model = flat_config.root_model()
assert model.name == 'framework'
assert len(model.components) == 3
assert 'c1' in model.components
assert 'c2' in model.components
assert 'c3' in model.components
assert len(model.conduits) == 2
assert has_conduit(model, 'c1.out', 'c2.in')
assert has_conduit(model, 'c2.out', 'c3.in')
assert len(flat_config.custom_implementations) == 0
assert model.components['c2'].implementation == 'p2'
# optional component without implementation
[docs]
def test_remove_no_implementation() -> None:
nested_config_yaml = (
'ymmsl_version: v0.2\n'
'description: Testing a component without implementation\n'
'models:\n'
' optional_micro:\n'
' description: Macro-micro with optional micro\n'
' components:\n'
' macro:\n'
' ports:\n'
' o_i: out\n'
' s: in\n'
' description: |\n'
' Macro model that can run with and without a micro.\n'
' implementation: p1\n'
' micro:\n'
' ports:\n'
' f_init: init\n'
' o_f: final\n'
' description: Optional micro model, not implemented\n'
' conduits:\n'
' macro.out: micro.init\n'
' micro.final: macro.in\n'
'programs:\n'
' p1:\n'
' description: Macro model implementation\n'
' executable: /home/user/codes/p1\n'
' p2:\n'
' description: Micro model implementation\n'
' executable: /home/user/codes/p2\n'
)
nested_config = load(nested_config_yaml)
flat_config = flatten(nested_config)
assert len(flat_config.models) == 1
assert 'optional_micro' in flat_config.models
model = flat_config.root_model()
assert model.name == 'optional_micro'
assert len(model.components) == 1
assert 'macro' in model.components
assert len(model.conduits) == 0
nested_config.custom_implementations[Reference('optional_micro.micro')] = \
Reference('p2')
resolve(Reference([]), nested_config)
flat_config = flatten(nested_config)
assert len(flat_config.models) == 1
assert 'optional_micro' in flat_config.models
model = flat_config.root_model()
assert model.name == 'optional_micro'
assert len(model.components) == 2
assert 'macro' in model.components
assert 'micro' in model.components
assert len(model.conduits) == 2
assert has_conduit(model, 'macro.out', 'micro.init')
assert has_conduit(model, 'micro.final', 'macro.in')
assert len(flat_config.custom_implementations) == 0
assert model.components['micro'].implementation == 'p2'
nested_config = load(nested_config_yaml)
nested_config.models['optional_micro'].components['micro'].implementation = \
Reference('p2')
nested_config.custom_implementations[Reference('optional_micro.micro')] = None
resolve(Reference([]), nested_config)
flat_config = flatten(nested_config)
assert len(flat_config.models) == 1
assert 'optional_micro' in flat_config.models
model = flat_config.root_model()
assert model.name == 'optional_micro'
assert len(model.components) == 1
assert 'macro' in model.components
assert len(model.conduits) == 0
[docs]
def test_nested_custom_implementations() -> None:
nested_config_yaml = (
'ymmsl_version: v0.2\n'
'description: Testing nested custom implementations\n'
'models:\n'
' outer:\n'
' description: Outer model\n'
' components:\n'
' c1:\n'
' ports:\n'
' o_f: out\n'
' description: Component c1\n'
' implementation: p1\n'
' c2:\n'
' ports:\n'
' f_init: in\n'
' description: Component c1\n'
' implementation: p1\n'
' conduits:\n'
' c1.out: c2.in\n'
' middle:\n'
' ports:\n'
' o_f: out\n'
' description: Middle model\n'
' components:\n'
' c1:\n'
' ports:\n'
' o_f: out\n'
' description: Another component c1\n'
' c2:\n'
' ports:\n'
' o_f: out\n'
' description: Component c2\n'
' conduits:\n'
' c2.out: out\n'
' inner:\n'
' ports:\n'
' o_f: out\n'
' description: Inner model\n'
' components:\n'
' c1:\n'
' ports:\n'
' o_f: out\n'
' description: Yet another c1, it could happen...\n'
' implementation: p1\n'
' conduits:\n'
' c1.out: out\n'
'programs:\n'
' p1:\n'
' description: Program 1\n'
' executable: /home/user/codes/p1\n'
' p2:\n'
' description: Program 2\n'
' executable: /home/user/codes/p2\n'
'custom_implementations:\n'
' outer.c1: middle\n'
' outer.c1.c1: inner\n'
' outer.c1.c2: inner\n'
' outer.c1.c2.c1: p2\n'
)
nested_config = load(nested_config_yaml)
resolve(Reference([]), nested_config)
flat_config = flatten(nested_config)
flat_model = flat_config.models['outer']
assert len(flat_model.components) == 3
assert flat_model.components['c1.c1.c1'].implementation == 'p1'
assert flat_model.components['c1.c2.c1'].implementation == 'p2'
assert len(flat_model.conduits) == 1
assert flat_model.conduits[0].sender == 'c1.c2.c1.out'
assert flat_model.conduits[0].receiver == 'c2.in'