Merge branch 'SSP' of https://github.com/grouperenault/fmutool into SSP

.gitignore

@@ -4,14 +4,14 @@ __pycache__
 .vs
 *.dll
 *.exe
-tests/bouncing_ball-keeponly.csv
-tests/bouncing_ball-keeponly.fmu
-tests/bouncing_ball-no-tl.csv
-tests/bouncing_ball-no-tl.fmu
-tests/bouncing_ball-removed.csv
-tests/bouncing_ball-removed.fmu
-tests/bouncing_ball-renamed.csv
-tests/bouncing_ball-renamed.fmu
+tests/operations/bouncing_ball-keeponly.csv
+tests/operations/bouncing_ball-keeponly.fmu
+tests/operations/bouncing_ball-no-tl.csv
+tests/operations/bouncing_ball-no-tl.fmu
+tests/operations/bouncing_ball-removed.csv
+tests/operations/bouncing_ball-removed.fmu
+tests/operations/bouncing_ball-renamed.csv
+tests/operations/bouncing_ball-renamed.fmu
 tests/bouncing_ball-win32.fmu
 remoting/out
 remoting/build
@@ -22,7 +22,7 @@ tests/containers/bouncing_ball/bouncing_auto.fmu
 tests/containers/bouncing_ball/bouncing_unlinked.fmu
 container/build-win64
 container/build
-tests/bouncing_ball.csv
+tests/operations/bouncing_ball.csv
 fmu_manipulation_toolbox/resources/darwin64/client_sm.dylib
 fmu_manipulation_toolbox/resources/darwin64/container.dylib
 fmu_manipulation_toolbox/resources/darwin64/server_sm

doc/README.md

@@ -20,7 +20,7 @@ FMU Manipulation Toolbox import supports FMI-2.0 and Co-Simulation interface.
 - Simulink
 - [Reference FMUs](https://github.com/modelica/Reference-FMUs)
 
-Automated testsuite use [bouncing_ball.fmu](../tests/bouncing_ball.fmu).
+Automated testsuite use [bouncing_ball.fmu](../tests/operations/bouncing_ball.fmu).
 
 
 ### FMU Export Compatibility information

fmu_manipulation_toolbox/assembly.py

@@ -31,7 +31,7 @@ def __init__(self, from_port: Port, to_port: Port):
 
 
 class AssemblyNode:
-    def __init__(self, name: str, step_size: float = None, mt = False, profiling = False,
+    def __init__(self, name: str, step_size: float = None, mt=False, profiling=False,
                  auto_link=True, auto_input=True, auto_output=True):
         self.name = name
         self.step_size = step_size
@@ -120,10 +120,11 @@ def __repr__(self):
 
 
 class Assembly:
-    def __init__(self, filename: str, step_size = None, auto_link: bool = True,  auto_input: bool = True,
-                 auto_output: bool = True, mt: bool = False, profiling: bool = False, fmu_directory: Path = "."):
+    def __init__(self, filename: str, step_size=None, auto_link=True,  auto_input=True, debug=False,
+                 auto_output=True, mt=False, profiling=False, fmu_directory: Path = "."):
         self.filename = Path(filename)
         self.default_auto_input = auto_input
+        self.debug=debug
         self.default_auto_output = auto_output
         self.default_step_size = step_size
         self.default_auto_link = auto_link
@@ -136,20 +137,22 @@ def __init__(self, filename: str, step_size = None, auto_link: bool = True,  aut
             raise FMUContainerError(f"FMU directory is not valid: '{fmu_directory}'")
 
         self.description_pathname = fmu_directory / self.filename  # For inclusion in FMU
-        self.root = self.read()
+        self.root = None
+        self.read()
 
     def __del__(self):
-        for filename in self.transient_filenames:
-            filename.unlink()
+        if not self.debug:
+            for filename in self.transient_filenames:
+                filename.unlink()
 
-    def read(self) -> AssemblyNode:
+    def read(self):
         logger.info(f"Reading '{self.filename}'")
         if self.filename.suffix == ".json":
-            return self.read_json()
+            self.read_json()
         elif self.filename.suffix == ".ssp":
-            return self.read_ssp()
+            self.read_ssp()
         elif self.filename.suffix == ".csv":
-            return self.read_csv()
+            self.read_csv()
         else:
             raise FMUContainerError(f"Not supported file format '{self.filename}")
 
@@ -161,11 +164,11 @@ def write(self, filename: str):
         else:
             logger.critical(f"Unable to write to '{filename}': format unsupported.")
 
-    def read_csv(self) -> AssemblyNode:
+    def read_csv(self):
         name = str(self.filename.with_suffix(".fmu"))
-        root = AssemblyNode(name, step_size=self.default_step_size, auto_link=self.default_auto_link,
-                            mt=self.default_mt, profiling=self.default_profiling, auto_input=self.default_auto_input,
-                            auto_output=self.default_auto_output)
+        self.root = AssemblyNode(name, step_size=self.default_step_size, auto_link=self.default_auto_link,
+                                 mt=self.default_mt, profiling=self.default_profiling, auto_input=self.default_auto_input,
+                                 auto_output=self.default_auto_output)
 
         with open(self.description_pathname) as file:
             reader = csv.reader(file, delimiter=';')
@@ -183,17 +186,14 @@ def read_csv(self) -> AssemblyNode:
                 rule = rule.upper()
                 if rule in ("LINK", "INPUT", "OUTPUT", "DROP", "FMU", "START"):
                     try:
-                        self._read_csv_rule(root, rule,
-                                            from_fmu_filename, from_port_name,
-                                            to_fmu_filename, to_port_name)
+                        self._read_csv_rule(self.root, rule,
+                                            from_fmu_filename, from_port_name, to_fmu_filename, to_port_name)
                     except AssemblyError as e:
                         logger.error(f"Line #{i+2}: {e}. Line skipped.")
                         continue
                 else:
                     logger.error(f"Line #{i+2}: unexpected rule '{rule}'. Line skipped.")
 
-        return root
-
     def write_csv(self, filename: Union[str, Path]):
         if self.root.children:
             raise AssemblyError("This assembly is not flat. Cannot export to CSV file.")
@@ -257,16 +257,14 @@ def check_csv_headers(reader):
         if not headers == ["rule", "from_fmu", "from_port", "to_fmu", "to_port"]:
             raise AssemblyError("Header (1st line of the file) is not well formatted.")
 
-    def read_json(self) -> AssemblyNode:
+    def read_json(self):
         with open(self.description_pathname) as file:
             try:
                 data = json.load(file)
             except json.decoder.JSONDecodeError as e:
                 raise FMUContainerError(f"Cannot read json: {e}")
-        root = self.json_decode_node(data)
-        root.name = str(self.filename.with_suffix(".fmu"))
-
-        return root
+        self.root = self.json_decode_node(data)
+        self.root.name = str(self.filename.with_suffix(".fmu"))
 
     def write_json(self, filename: Union[str, Path]):
         with open(self.fmu_directory / filename, "wt") as file:
@@ -348,22 +346,48 @@ def json_decode_node(self, data) -> AssemblyNode:
 
         return node
 
-    def read_ssp(self) -> AssemblyNode:
+    def read_ssp(self):
         logger.warning("This feature is ALPHA stage.")
-        name = str(self.filename.with_suffix(".fmu"))
-        root = AssemblyNode(name, step_size=self.default_step_size, auto_link=self.default_auto_link,
-                            mt=self.default_mt, profiling=self.default_profiling, auto_input=self.default_auto_input,
-                            auto_output=self.default_auto_output)
+        node_stack: List[AssemblyNode] = []
+
         def start_element(tag_name, attrs):
             if tag_name == 'ssd:Connection':
-                root.add_link(attrs['startElement'] + '.fmu', attrs['startConnector'],
-                              attrs['endElement'] + '.fmu', attrs['endConnector'])
+                if 'startElement' in attrs:
+                    if 'endElement' in attrs:
+                        node_stack[-1].add_link(attrs['startElement'] + '.fmu', attrs['startConnector'],
+                                                attrs['endElement'] + '.fmu', attrs['endConnector'])
+                    else:
+                        node_stack[-1].add_output(attrs['startElement'] + '.fmu', attrs['startConnector'],
+                                                  attrs['endConnector'])
+                else:
+                    node_stack[-1].add_input(attrs['startConnector'],
+                                             attrs['endElement'] + '.fmu', attrs['endConnector'])
+
+            elif tag_name == 'ssd:System':
+                name = attrs['name']+".fmu"
+                logger.info(f"System: {name}")
+                node = AssemblyNode(name, step_size=self.default_step_size, auto_link=self.default_auto_link,
+                                    mt=self.default_mt, profiling=self.default_profiling,
+                                    auto_input=self.default_auto_input, auto_output=self.default_auto_output)
+                if node_stack:
+                    node_stack[-1].add_sub_node(node)
+                else:
+                    self.root = node
+
+                node_stack.append(node)
+
+            elif tag_name == 'ssd:Component':
+                logger.info(f"Component: {attrs}")
+
+        def end_element(tag_name):
+            if tag_name == 'ssd:System':
+                node_stack.pop()
 
         # TODO: handle nested SSD
         with zipfile.ZipFile(self.fmu_directory / self.filename) as zin:
             for file in zin.filelist:
                 target_filename = Path(file.filename).name
-                if file.filename.endswith(".fmu") or file.filename == "SystemStructure.ssd":
+                if file.filename.endswith(".fmu") or file.filename.endswith(".ssd"):
                     zin.getinfo(file.filename).filename = target_filename
                     zin.extract(file, path=self.fmu_directory)
                     logger.debug(f"Extraction {file.filename}")
@@ -375,8 +399,8 @@ def start_element(tag_name, attrs):
             with open(self.description_pathname, "rb") as file:
                 parser = xml.parsers.expat.ParserCreate()
                 parser.StartElementHandler = start_element
+                parser.EndElementHandler = end_element
                 parser.ParseFile(file)
-        return root
 
-    def make_fmu(self, debug=False):
-        self.root.make_fmu(self.fmu_directory, debug=debug, description_pathname=self.description_pathname)
+    def make_fmu(self):
+        self.root.make_fmu(self.fmu_directory, debug=self.debug, description_pathname=self.description_pathname)

fmu_manipulation_toolbox/cli.py

@@ -203,7 +203,7 @@ def fmucontainer():
         try:
             assembly = Assembly(filename, step_size=step_size, auto_link=config.auto_link,
                                 auto_input=config.auto_input, auto_output=config.auto_output, mt=config.mt,
-                                profiling=config.profiling, fmu_directory=fmu_directory)
+                                profiling=config.profiling, fmu_directory=fmu_directory, debug=config.debug)
 
         except FileNotFoundError as e:
             logger.fatal(f"Cannot read file: {e}")
@@ -213,7 +213,7 @@ def fmucontainer():
             continue
 
         try:
-            assembly.make_fmu(debug=config.debug)
+            assembly.make_fmu()
         except FMUContainerError as e:
             logger.fatal(f"{filename}: {e}")
             continue

fmu_manipulation_toolbox/fmu_container.py

@@ -30,10 +30,9 @@ def __init__(self, attrs: Dict[str, str]):
     def set_port_type(self, type_name: str, attrs: Dict[str, str]):
         self.type_name = type_name
         self.child = attrs.copy()
-        try:
-            self.child.pop("unit")  # Unit are not supported
-        except KeyError:
-            pass
+        for unsupported in ("unit", "declaredType"):
+            if unsupported in self.child:
+                self.child.pop(unsupported)
 
     def xml(self, vr: int, name=None, causality=None, start=None):
 
@@ -104,10 +103,16 @@ def cosimulation_attrs(self, attrs: Dict[str, str]):
             self.capabilities[capability] = attrs.get(capability, "false")
 
     def experiment_attrs(self, attrs):
-        self.step_size = float(attrs['stepSize'])
+        try:
+            self.step_size = float(attrs['stepSize'])
+        except KeyError:
+            logger.warning(f"FMU '{self.name}' does not specify preferred step size")
+            pass
 
     def scalar_type(self, type_name, attrs):
-        self.current_port.set_port_type(type_name, attrs)
+        if self.current_port:
+            self.current_port.set_port_type(type_name, attrs)
+        self.current_port = None
 
     def __repr__(self):
         return f"FMU '{self.name}' ({len(self.ports)} variables)"
@@ -341,6 +346,8 @@ def minimum_step_size(self) -> float:
     def sanity_check(self, step_size: Union[float, None]):
         nb_error = 0
         for fmu in self.execution_order:
+            if not fmu.step_size:
+                continue
             ts_ratio = step_size / fmu.step_size
             if ts_ratio < 1.0:
                 logger.error(f"Container step_size={step_size}s is lower than FMU '{fmu.name}' "

tests/test_suite.py

@@ -1,8 +1,8 @@
 import unittest
 import sys
-import os
+from pathlib import Path
 
-sys.path.insert(0, os.path.relpath(os.path.join(os.path.dirname(__file__), "..")))
+sys.path.insert(0, str(Path(__file__).parent.parent))
 from fmu_manipulation_toolbox.fmu_operations import *
 from fmu_manipulation_toolbox.fmu_container import *
 from fmu_manipulation_toolbox.assembly import *
@@ -11,17 +11,16 @@
 class FMUManipulationToolboxTestSuite(unittest.TestCase):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
-        self.fmu_filename = "bouncing_ball.fmu"
+        self.fmu_filename = "operations/bouncing_ball.fmu"
 
     def assert_identical_files(self, filename1, filename2):
-        with open(filename1, mode="rt", newline=None) as a, \
-                open(filename2, mode="rt", newline=None) as b:
+        with open(filename1, mode="rt", newline=None) as a, open(filename2, mode="rt", newline=None) as b:
             self.assertTrue(all(lineA == lineB for lineA, lineB in zip(a, b)))
 
     def assert_names_match_ref(self, fmu_filename):
         fmu = FMU(fmu_filename)
-        csv_filename = os.path.splitext(fmu_filename)[0] + ".csv"
-        ref_filename = "REF-" + csv_filename
+        csv_filename = Path(fmu_filename).with_suffix(".csv")
+        ref_filename = csv_filename.with_stem("REF-"+csv_filename.stem)
         operation = OperationSaveNamesToCSV(csv_filename)
         fmu.apply_operation(operation)
         self.assert_identical_files(ref_filename, csv_filename)
@@ -33,14 +32,14 @@ def assert_operation_match_ref(self, fmu_filename, operation):
         self.assert_names_match_ref(fmu_filename)
 
     def test_strip_top_level(self):
-        self.assert_operation_match_ref("bouncing_ball-no-tl.fmu", OperationStripTopLevel())
+        self.assert_operation_match_ref("operations/bouncing_ball-no-tl.fmu", OperationStripTopLevel())
 
     def test_save_names_to_CSV(self):
-        self.assert_names_match_ref("bouncing_ball.fmu")
+        self.assert_names_match_ref("operations/bouncing_ball.fmu")
 
     def test_rename_from_CSV(self):
-        self.assert_operation_match_ref("bouncing_ball-renamed.fmu",
-                                        OperationRenameFromCSV("bouncing_ball-modified.csv"))
+        self.assert_operation_match_ref("operations/bouncing_ball-renamed.fmu",
+                                        OperationRenameFromCSV("operations/bouncing_ball-modified.csv"))
 
     @unittest.skipUnless(sys.platform.startswith("win"), "Supported only on Windows")
     def test_add_remoting_win32(self):
@@ -50,17 +49,17 @@ def test_add_remoting_win32(self):
         fmu.repack("bouncing_ball-win32.fmu")
 
     def test_remove_regexp(self):
-        self.assert_operation_match_ref("bouncing_ball-removed.fmu",
+        self.assert_operation_match_ref("operations/bouncing_ball-removed.fmu",
                                         OperationRemoveRegexp("e"))
 
     def test_keep_only_regexp(self):
-        self.assert_operation_match_ref("bouncing_ball-keeponly.fmu",
+        self.assert_operation_match_ref("operations/bouncing_ball-keeponly.fmu",
                                         OperationKeepOnlyRegexp("e"))
 
     def test_container(self):
-        assembly = Assembly("bouncing.csv", fmu_directory=Path("containers/bouncing_ball"), mt=True)
+        assembly = Assembly("bouncing.csv", fmu_directory=Path("containers/bouncing_ball"), mt=True, debug=True)
         assembly.write_json("bouncing.json")
-        assembly.make_fmu(debug=True)
+        assembly.make_fmu()
         self.assert_identical_files("containers/bouncing_ball/REF_container.txt",
                                     "containers/bouncing_ball/bouncing/resources/container.txt")
         self.assert_identical_files("containers/bouncing_ball/REF_bouncing.json",