Added fmu controller

rocket_twin/drivers/mission.py

@@ -44,7 +44,7 @@ def __init__(
         init_fuel = init
         init_flight = {
             "rocket.flying": True,
-            "rocket.engine.force_command": 1.0,
+            "rocket.force_command": 1.0,
             "rocket.tank.w_out_max": 3.0,
             "g_tank.w_in": 0.0,
             "g_tank.w_command": 0.0,

rocket_twin/systems/__init__.py

@@ -1,3 +1,4 @@
+from rocket_twin.systems.control import Clock
 from rocket_twin.systems.engine import Engine
 from rocket_twin.systems.ground import Ground
 from rocket_twin.systems.physics import Dynamics
@@ -6,4 +7,4 @@
 from rocket_twin.systems.rocket import Rocket  # isort: skip
 from rocket_twin.systems.station import Station  # isort: skip
 
-__all__ = ["Engine", "Tank", "Rocket", "Pipe", "Dynamics", "Station", "Ground"]
+__all__ = ["Clock", "Engine", "Tank", "Rocket", "Pipe", "Dynamics", "Station", "Ground"]

rocket_twin/systems/control/__init__.py

@@ -0,0 +1,3 @@
+from rocket_twin.systems.control.clock import Clock
+
+__all__ = ['Clock']

rocket_twin/systems/control/clock.py

@@ -0,0 +1,13 @@
+from cosapp.base import System
+
+class Clock(System):
+
+    def setup(self):
+
+        #Transient to ensure the system is visited in every time step
+        self.add_transient('x', der='1')
+        self.add_outward('time_var', 0., desc="Command time")
+
+    def compute(self):
+
+        self.time_var = self.time

rocket_twin/systems/rocket/rocket.py

@@ -16,8 +16,8 @@ class Rocket(System):
     """
 
     def setup(self):
-        self.add_child(Engine("engine"))
-        self.add_child(Tank("tank"), pulling=["w_in"])
+        self.add_child(Engine("engine"), pulling=["force_command"])
+        self.add_child(Tank("tank"), pulling=["w_in", "w_command"])
         self.add_child(
             Dynamics(
                 "dyn",

rocket_twin/systems/station/station.py

@@ -1,7 +1,7 @@
 from cosapp.base import System
 from cosapp_fmu.FMUsystem import FMUSystem
 
-from rocket_twin.systems import Pipe, Rocket, Tank
+from rocket_twin.systems import Pipe, Rocket, Tank, Clock
 
 
 class Station(System):
@@ -14,16 +14,21 @@ class Station(System):
     ------
     """
 
-    def setup(self):
+    def setup(self, fmu_path=None):
         self.add_child(Tank("g_tank"))
         self.add_child(Pipe("pipe"))
         self.add_child(Rocket("rocket"))
 
         self.connect(self.g_tank.outwards, self.pipe.inwards, {"w_out": "w_in"})
         self.connect(self.pipe.outwards, self.rocket.inwards, {"w_out": "w_in"})
-        self.connect(self.controller.outwards, self.pipe.inwards, ["is_open"])
-        #self.connect(self.controller.outwards, self.rocket.inwards, ["switch"])
-        #self.connect(self.controller.outwards, self.g_tank.inwards, {'w_out_t' : 'w_out_temp'})
-        #self.connect(self.controller.outwards, self.rocket.tank.inwards, {'w_out_r' : 'w_out_temp'})
 
-        self.g_tank.weight_max = 10.0
+        if fmu_path is not None:
+            self.add_child(Clock('clock'))
+            self.add_child(FMUSystem('controller', fmu_path=fmu_path))
+            self.connect(self.clock.outwards, self.controller.inwards, {'time_var' : 'ti'})
+            self.connect(self.controller.outwards, self.g_tank.inwards, {'wg' : 'w_command'})
+            self.connect(self.controller.outwards, self.rocket.inwards, {'wr' : 'w_command', 'f' : 'force_command'})
+
+            self.exec_order = ['clock', 'controller', 'g_tank', 'pipe', 'rocket']
+
+        self.g_tank.weight_max = 10.0

rocket_twin/tests/test_command.py

@@ -0,0 +1,14 @@
+import numpy as np
+from rocket_twin.systems import Clock
+from cosapp.drivers import RungeKutta
+
+class TestClock:
+
+    def test_time(self):
+
+        sys = Clock('sys')
+        sys.add_driver(RungeKutta(order=4,time_interval=[0, 15], dt=1.))
+        sys.run_drivers()
+
+        np.testing.assert_allclose(sys.time_var, 15., atol=10**(-4))
+

rocket_twin/tests/test_flying_rocket.py

@@ -14,7 +14,7 @@ def test_run_once(self):
 
         init = {
             "rocket.flying": True,
-            "rocket.engine.force_command": 1.0,
+            "rocket.force_command": 1.0,
             "rocket.tank.weight_p": "rocket.tank.weight_max",
             "rocket.tank.w_out_max": 3.0,
             "g_tank.w_in": 0.0,

rocket_twin/tests/test_mission.py

@@ -13,14 +13,15 @@ def test_run_once(self):
 
         init = {
             "g_tank.weight_p": "g_tank.weight_max",
-            "rocket.engine.force_command": 0.0,
+            "rocket.force_command": 0.0,
             "rocket.tank.weight_p": 0.0,
             "rocket.tank.w_out_max": 0.0,
             "g_tank.w_command": 1.0,
             "g_tank.w_in": 0.0,
             "g_tank.w_out_max": 3.0,
         }
 
+
         stop = "rocket.tank.weight_p <= 0."
 
         includes = ["rocket.a", "g_tank.weight", "rocket.tank.weight_p"]
@@ -37,3 +38,6 @@ def test_run_once(self):
         np.testing.assert_allclose(sys.rocket.a, 40.0, atol=10 ** (-10))
         np.testing.assert_allclose(sys.rocket.tank.weight_p, 0.0, atol=10 ** (-10))
         np.testing.assert_allclose(sys.g_tank.weight_p, 5.0, atol=10 ** (-10))
+
+tm = TestMission()
+tm.test_run_once()

rocket_twin/tests/test_refuel_rocket.py

@@ -12,9 +12,9 @@ def test_run_once(self):
         dt = 0.1
 
         init = {
-            "rocket.engine.force_command": 0.0,
             "rocket.tank.weight_p": 0.0,
             "rocket.tank.w_out_max": 0.0,
+            "rocket.force_command": 0.0,
             "g_tank.w_command": 1.0,
             "g_tank.weight_p": "g_tank.weight_max",
             "g_tank.w_in": 0.0,

rocket_twin/tests/test_sequences_fmu.py

@@ -0,0 +1,26 @@
+import numpy as np
+from rocket_twin.systems import Station
+from cosapp.drivers import RungeKutta
+from cosapp.recorders import DataFrameRecorder
+
+class TestSequencesFMU:
+
+    def test_sequence_fmu(self):
+
+        sys = Station('sys', fmu_path="rocket_twin/systems/control/controller.fmu")
+        driver = sys.add_driver(RungeKutta(order=4,time_interval=[0, 15], dt=0.01))
+        init = {'g_tank.weight_p' : 10.,
+                'rocket.tank.weight_p' : 0.}
+        values = {'g_tank.w_out_max' : 1.,
+                  'rocket.tank.w_out_max' : 0.5,
+                  'rocket.engine.force_max' : 100.}
+        driver.set_scenario(init=init, values=values)
+        driver.add_recorder(DataFrameRecorder(includes=['rocket.dyn.a']), period=1.)
+        sys.run_drivers()
+
+        data=driver.recorder.export_data()
+        print(data)
+
+        np.testing.assert_allclose(sys.rocket.dyn.a, 40., atol=10**(-1))
+        np.testing.assert_allclose(sys.g_tank.weight_p, 5., atol=10**(-2))
+        np.testing.assert_allclose(sys.rocket.tank.weight_p, 0., atol=10**(-2))