Removed clock, added cosapp controller

rocket_twin/drivers/mission.py

@@ -44,10 +44,11 @@ def __init__(
         init_fuel = init
         init_flight = {
             "rocket.flying": True,
-            "rocket.force_command": 1.0,
+            "controller.cos_control.f_temp": 1.0,
+            "controller.cos_control.wg_temp": 0.0,
+            "controller.cos_control.wr_temp": 1.0,
             "rocket.tank.w_out_max": 3.0,
             "g_tank.w_in": 0.0,
-            "g_tank.w_command": 0.0,
         }
 
         stop_fuel = "rocket.tank.weight_p >= rocket.tank.weight_max"

rocket_twin/systems/__init__.py

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

rocket_twin/systems/control/__init__.py

@@ -1,3 +1,5 @@
-from rocket_twin.systems.control.clock import Clock
+from rocket_twin.systems.control.cosapp_controller import CosappController
 
-__all__ = ["Clock"]
+from rocket_twin.systems.control.controller import Controller  # isort: skip
+
+__all__ = ["CosappController", "Controller"]

rocket_twin/systems/control/controller.py

@@ -0,0 +1,41 @@
+from cosapp.base import System
+from cosapp_fmu.FMUsystem import FMUSystem
+from rocket_twin.systems.control import CosappController
+
+from rocket_twin.utils import create_FMU
+
+class Controller(System):
+    """Controller of the command variables.
+
+    Inputs
+    ------
+    model_path: string,
+        the path to the .mo file, if any
+    model_name: string
+        the .fmu file name
+
+    Outputs
+    ------
+    'f': float,
+        command force
+    'wr': float,
+        command flow
+    'wg' float,
+        command flow
+    """
+
+    def setup(self, model_path=None, model_name=None):
+
+        self.add_inward('time_var', 0., desc="System time", unit='')
+        self.add_transient('x', der='1')
+
+        if model_path is None:
+            self.add_child(CosappController('cos_control'), pulling=['f', 'wr', 'wg'])
+        else:
+            fmu_path = create_FMU(model_path, model_name)
+            self.add_child(FMUSystem("fmu_controller", fmu_path=fmu_path), pulling={'f' : 'f', 'wr' : 'wr', 'wg' : 'wg', 'ti' : 'time_var'})
+
+    def compute(self):
+
+        self.time_var = self.time
+''

rocket_twin/systems/control/cosapp_controller.py

@@ -0,0 +1,35 @@
+from cosapp.base import System
+
+class CosappController(System):
+    """System which allows command through CoSApp sequences.
+    
+    Inputs
+    ------
+
+    Outputs
+    ------
+    'f': float,
+        command force
+    'wr': float,
+        command flow
+    'wg' float,
+        command flow
+    """
+
+    def setup(self):
+
+        self.add_inward('f_temp', 0., desc="Ratio of command force to maximum force", unit='')
+        self.add_inward('wr_temp', 0., desc="Ratio of command fuel flow to maximum flow", unit='')
+        self.add_inward('wg_temp', 0., desc="Ratio of command fuel flow to maximum flow", unit='')
+
+        self.add_outward('f', 0., desc="Ratio of command force to maximum force", unit='')
+        self.add_outward('wr', 0., desc="Ratio of command fuel flow to maximum flow", unit='')
+        self.add_outward('wg', 0., desc="Ratio of command fuel flow to maximum flow", unit='')
+
+    def compute(self):
+
+        self.f = self.f_temp
+        self.wr = self.wr_temp
+        self.wg = self.wg_temp
+
+

rocket_twin/systems/station/station.py

@@ -1,42 +1,36 @@
 from cosapp.base import System
-from cosapp_fmu.FMUsystem import FMUSystem
 
-from rocket_twin.systems import Clock, Pipe, Rocket, Tank
-from rocket_twin.utils import create_FMU
+from rocket_twin.systems import Controller, Pipe, Rocket, Tank
 
 
 class Station(System):
     """A space station composed by a rocket, a tank and a pipe connecting them.
 
     Inputs
     ------
-    fmu_path: string,
-        the path to the .fmu file, if there is any
+    model_path: string,
+        the path to the .mo file, if any
+    model_name: string
+        the .fmu file name
 
     Outputs
     ------
     """
 
     def setup(self, model_path=None, model_name=None):
+        self.add_child(Controller("controller", model_path=model_path, model_name=model_name))
         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"})
 
-        if model_path is not None:
-            self.add_child(Clock("clock"))
-            fmu_path = create_FMU(model_path, model_name)
-            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.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.g_tank.weight_max = 10.0

rocket_twin/systems/tank/tank.py

@@ -34,16 +34,14 @@ def setup(self):
         self.add_inward("w_out_max", 0.0, desc="Fuel output rate", unit="kg/s")
         self.add_inward("w_command", 1.0, desc="Fuel output control variable", unit="")
 
-        # Transient
-        self.add_outward("dw_dt", 0.0, desc="Fuel mass rate of change", unit="kg/s")
-        self.add_transient("weight_p", der="dw_dt", desc="Propellant weight")
-
         # Outputs
         self.add_outward("weight", 1.0, desc="Weight", unit="kg")
         self.add_outward("cg", 1.0, desc="Center of gravity", unit="m")
         self.add_outward("w_out", 0.0, desc="Fuel output rate", unit="kg/s")
 
+        # Transient
+        self.add_transient("weight_p", der="w_in - w_out", desc="Propellant weight")
+
     def compute(self):
         self.w_out = self.w_out_max * self.w_command
-        self.dw_dt = self.w_in - self.w_out
         self.weight = self.weight_s + self.weight_p

rocket_twin/tests/test_flying_rocket.py

@@ -14,12 +14,13 @@ def test_run_once(self):
 
         init = {
             "rocket.flying": True,
-            "rocket.force_command": 1.0,
+            "controller.cos_control.f_temp": 1.0,
+            "controller.cos_control.wg_temp": 0.0,
+            "controller.cos_control.wr_temp": 1.0,
             "rocket.tank.weight_p": "rocket.tank.weight_max",
             "rocket.tank.w_out_max": 3.0,
             "g_tank.w_in": 0.0,
             "g_tank.weight_p": 0.0,
-            "g_tank.w_command": 0.0,
         }
 
         stop = "rocket.tank.weight_p <= 0."

rocket_twin/tests/test_mission.py

@@ -12,11 +12,11 @@ def test_run_once(self):
         dt = 0.1
 
         init = {
-            "g_tank.weight_p": "g_tank.weight_max",
-            "rocket.force_command": 0.0,
             "rocket.tank.weight_p": 0.0,
-            "rocket.tank.w_out_max": 0.0,
-            "g_tank.w_command": 1.0,
+            "controller.cos_control.wr_temp": 0.0,
+            "controller.cos_control.f_temp": 0.0,
+            "controller.cos_control.wg_temp": 1.0,
+            "g_tank.weight_p": "g_tank.weight_max",
             "g_tank.w_in": 0.0,
             "g_tank.w_out_max": 3.0,
         }

rocket_twin/tests/test_refuel_rocket.py

@@ -13,9 +13,9 @@ def test_run_once(self):
 
         init = {
             "rocket.tank.weight_p": 0.0,
-            "rocket.tank.w_out_max": 0.0,
-            "rocket.force_command": 0.0,
-            "g_tank.w_command": 1.0,
+            "controller.cos_control.wr_temp": 0.0,
+            "controller.cos_control.f_temp": 0.0,
+            "controller.cos_control.wg_temp": 1.0,
             "g_tank.weight_p": "g_tank.weight_max",
             "g_tank.w_in": 0.0,
             "g_tank.w_out_max": 3.0,

rocket_twin/tests/test_sequences.py

@@ -32,7 +32,7 @@ def test_fuel(self):
             {
                 "name": "fuel",
                 "type": "transient",
-                "init": {"g_tank.w_out_max": 1.0, "g_tank.w_command": 1.0},
+                "init": {"g_tank.w_out_max": 1.0, "controller.cos_control.wg_temp": 1.0},
                 "dt": 0.1,
                 "stop": "rocket.tank.weight_p == rocket.tank.weight_max",
             }
@@ -59,9 +59,9 @@ def test_flight(self):
                 "init": {
                     "rocket.flying": True,
                     "rocket.tank.w_out_max": 0.5,
-                    "g_tank.w_command": 0.0,
-                    "rocket.engine.force_command": 1.0,
-                    "rocket.tank.w_command": 1.0,
+                    "controller.cos_control.wg_temp": 0.0,
+                    "controller.cos_control.f_temp": 1.0,
+                    "controller.cos_control.wr_temp": 1.0,
                 },
                 "dt": 0.1,
                 "stop": f"time > {self.sys.time} + 10.",

rocket_twin/tests/test_sequences_fmu.py

@@ -1,14 +1,13 @@
 import numpy as np
 from cosapp.drivers import RungeKutta
-from cosapp.recorders import DataFrameRecorder
 
 from rocket_twin.systems import Station
 
 
 class TestSequencesFMU:
     def test_sequence_fmu(self):
 
-        model_path = r"rocket_twin\\systems\\control\\controller.mo"
+        model_path = r"C:\\Users\\Lucs\\Documents\\Polytechnique\\PSC\\rocket-twin\\rocket_twin\\systems\\control\\controller.mo"
         model_name = "controller"
         sys = Station("sys", model_path=model_path, model_name=model_name)
         driver = sys.add_driver(RungeKutta(order=4, time_interval=[0, 15], dt=0.01))
@@ -19,12 +18,8 @@ def test_sequence_fmu(self):
             "rocket.engine.force_max": 100.0,
         }
         driver.set_scenario(init=init, values=values)
-        driver.add_recorder(DataFrameRecorder(includes=["rocket.dyn.a"]), period=1.0)
         sys.run_drivers()
 
-        data = driver.recorder.export_data()
-        print(data)
-
-        np.testing.assert_allclose(sys.rocket.dyn.a, 40.0, atol=10 ** (-1))
-        np.testing.assert_allclose(sys.g_tank.weight_p, 5.0, atol=10 ** (-2))
-        np.testing.assert_allclose(sys.rocket.tank.weight_p, 0.0, atol=10 ** (-2))
+        np.testing.assert_allclose(sys.rocket.dyn.a, 40.0, atol=10 ** (0))
+        np.testing.assert_allclose(sys.g_tank.weight_p, 5.0, atol=10 ** (0))
+        np.testing.assert_allclose(sys.rocket.tank.weight_p, 0.0, atol=10 ** (0))

rocket_twin/utils/__init__.py

@@ -1,4 +1,4 @@
-from rocket_twin.utils.run_sequences import run_sequences
 from rocket_twin.utils.create_fmu import create_FMU
+from rocket_twin.utils.run_sequences import run_sequences
 
 __all__ = ["run_sequences", "create_FMU"]

rocket_twin/utils/create_fmu.py

@@ -1,19 +1,37 @@
 import os
-import rocket_twin.systems.control
+
 from OMPython import ModelicaSystem
 
+import rocket_twin.systems.control
+
+
 def create_FMU(model_path, model_name):
+    """Create an fmu file in the control folder from an mo file.
+
+    Inputs
+    ------
+    model_path: string
+        the path of the .mo file
+    model_name: string
+        the name of the .fmu file to be created
+
+    Outputs
+    ------
+
+    fmu: string
+        the path to the .fmu file
+    """
 
     fmu_path = os.path.join(rocket_twin.systems.control.__path__[0], model_name)
     try:
         os.mkdir(fmu_path)
     except OSError:
         pass
     os.chdir(fmu_path)
-    mod=ModelicaSystem(model_path,model_name)
-    fmu =  mod.convertMo2Fmu()
+    mod = ModelicaSystem(model_path, model_name)
+    fmu = mod.convertMo2Fmu()
     for filename in os.listdir(fmu_path):
         if filename != (model_name + ".fmu"):
             os.remove(filename)
 
-    return fmu
+    return fmu