class Assembly:
"""High-level interface for loading, manipulating, and building FMU Container assemblies.
`Assembly` reads a description file (CSV, JSON, or SSP), constructs the corresponding
[AssemblyNode][fmu_manipulation_toolbox.assembly.AssemblyNode] tree, and provides methods
to build the container FMU or export the assembly to a different format.
This is the main entry point for both the `fmucontainer` CLI and the Python API.
Examples:
```python
from pathlib import Path
from fmu_manipulation_toolbox.assembly import Assembly
assembly = Assembly("bouncing.csv",
fmu_directory=Path("containers/bouncing_ball"),
mt=True)
assembly.make_fmu()
```
Attributes:
filename (Path): Path to the description file.
fmu_directory (Path): Directory containing the source FMUs.
debug (bool): Whether debug mode is enabled.
root (AssemblyNode | None): Root node of the assembly tree.
"""
def __init__(self, filename: str, step_size=None, auto_link=True, auto_input=True, debug=False, sequential=False,
auto_output=True, mt=False, profiling=False, fmu_directory: Path = Path("."), auto_parameter=False,
auto_local=False, ts_multiplier=False):
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
self.default_auto_parameter = auto_parameter
self.default_auto_local = auto_local
self.default_mt = mt
self.default_sequential = sequential
self.default_profiling = profiling
self.default_ts_multiplier = ts_multiplier
self.fmu_directory = fmu_directory
self.transient_filenames: List[Path] = []
self.transient_dirnames: Set[Path] = set()
if not fmu_directory.is_dir():
raise AssemblyError(f"FMU directory is not valid: '{fmu_directory.resolve()}'")
self.input_pathname = fmu_directory / self.filename
self.description_pathname = self.input_pathname # For inclusion in FMU
self.root: Optional[AssemblyNode] = None
self.read()
def add_transient_file(self, filename: str):
self.transient_filenames.append(self.fmu_directory / filename)
self.transient_dirnames.add(Path(filename).parent)
def __del__(self):
if not self.debug:
for filename in self.transient_filenames:
try:
filename.unlink()
except FileNotFoundError:
pass
for dirname in self.transient_dirnames:
while not str(dirname) == ".":
try:
(self.fmu_directory / dirname).rmdir()
except FileNotFoundError:
pass
dirname = dirname.parent
def read(self):
"""Read and parse the description file.
The format is determined by the file extension: `.json`, `.ssp`, or `.csv`.
Raises:
AssemblyError: If the file format is not supported.
"""
logger.info(f"Reading '{self.filename}'")
if self.filename.suffix == ".json":
self.read_json()
elif self.filename.suffix == ".ssp":
self.read_ssp()
elif self.filename.suffix == ".csv":
self.read_csv()
else:
raise AssemblyError(f"Not supported file format '{self.filename}")
def write(self, filename: str):
"""Export the assembly to a file.
Args:
filename (str): Output filename. The format is determined by the
extension (`.csv` or `.json`).
Raises:
AssemblyError: If the format is not supported.
"""
if filename.endswith(".csv"):
return self.write_csv(filename)
elif filename.endswith(".json"):
return self.write_json(filename)
else:
raise AssemblyError(f"Unable to write to '{filename}': format unsupported.")
def read_csv(self):
"""Parse a CSV description file and populate the assembly tree."""
name = str(self.filename.with_suffix(".fmu"))
self.root = AssemblyNode(name, step_size=self.default_step_size, auto_link=self.default_auto_link,
mt=self.default_mt, profiling=self.default_profiling,
sequential=self.default_sequential, auto_input=self.default_auto_input,
auto_output=self.default_auto_output, auto_parameter=self.default_auto_parameter,
auto_local=self.default_auto_local, ts_multiplier=self.default_ts_multiplier)
with open(self.input_pathname) as file:
reader = csv.reader(file, delimiter=';')
self._check_csv_headers(reader)
for i, row in enumerate(reader):
if not row or row[0][0] == '#': # skip blank line of comment
continue
try:
rule, from_fmu_filename, from_port_name, to_fmu_filename, to_port_name = row
except ValueError:
logger.error(f"Line #{i+2}: expecting 5 columns. Line skipped.")
continue
try:
self._read_csv_rule(self.root, rule.upper(),
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
@staticmethod
def _check_csv_headers(reader):
headers = next(reader)
headers_lowered = [h.lower() for h in headers]
if not headers_lowered == ["rule", "from_fmu", "from_port", "to_fmu", "to_port"]:
raise AssemblyError("Header (1st line of the file) is not well formatted.")
@staticmethod
def _read_csv_rule(node: AssemblyNode, rule: str, from_fmu_filename: str, from_port_name: str,
to_fmu_filename: str, to_port_name: str):
if rule == "FMU":
if not from_fmu_filename:
raise AssemblyError("Missing FMU information.")
node.add_fmu(from_fmu_filename)
elif rule == "INPUT":
if not to_fmu_filename or not to_port_name:
raise AssemblyError("Missing INPUT ports information.")
if not from_port_name:
from_port_name = to_port_name
node.add_input(from_port_name, to_fmu_filename, to_port_name)
elif rule == "OUTPUT":
if not from_fmu_filename or not from_port_name:
raise AssemblyError("Missing OUTPUT ports information.")
if not to_port_name:
to_port_name = from_port_name
node.add_output(from_fmu_filename, from_port_name, to_port_name)
elif rule == "DROP":
if not from_fmu_filename or not from_port_name:
raise AssemblyError("Missing DROP ports information.")
node.add_drop_port(from_fmu_filename, from_port_name)
elif rule == "LINK":
node.add_link(from_fmu_filename, from_port_name, to_fmu_filename, to_port_name)
elif rule == "START":
if not from_fmu_filename or not from_port_name or not to_fmu_filename:
raise AssemblyError("Missing START ports information.")
node.add_start_value(from_fmu_filename, from_port_name, to_fmu_filename)
else:
raise AssemblyError(f"unexpected rule '{rule}'. Line skipped.")
def write_csv(self, filename: Union[str, Path]):
"""Export the assembly as a CSV file.
Args:
filename (str | Path): Output filename, relative to `fmu_directory`.
Raises:
AssemblyError: If the assembly contains nested containers
(not representable in CSV).
"""
if self.root.children:
raise AssemblyError("This assembly is not flat. Cannot export to CSV file.")
with open(self.fmu_directory / filename, "wt") as outfile:
outfile.write("rule;from_fmu;from_port;to_fmu;to_port\n")
for fmu in self.root.fmu_names_list:
outfile.write(f"FMU;{fmu};;;\n")
for port, source in self.root.input_ports.items():
outfile.write(f"INPUT;;{source};{port.fmu_name};{port.port_name}\n")
for port, target in self.root.output_ports.items():
outfile.write(f"OUTPUT;{port.fmu_name};{port.port_name};;{target}\n")
for link in self.root.links:
outfile.write(f"LINK;{link.from_port.fmu_name};{link.from_port.port_name};"
f"{link.to_port.fmu_name};{link.to_port.port_name}\n")
for port, value in self.root.start_values.items():
outfile.write(f"START;{port.fmu_name};{port.port_name};{value};\n")
for port in self.root.drop_ports:
outfile.write(f"DROP;{port.fmu_name};{port.port_name};;\n")
def read_json(self):
"""Parse a JSON description file and populate the assembly tree.
Raises:
AssemblyError: If the JSON is malformed or contains invalid keywords.
"""
with open(self.input_pathname) as file:
try:
data = json.load(file)
except json.decoder.JSONDecodeError as e:
raise AssemblyError(f"Cannot read json: {e}")
self.root = self._json_decode_node(data)
if not self.root.name:
self.root.name = str(self.filename.with_suffix(".fmu").name)
def _json_decode_node(self, data: Dict) -> AssemblyNode:
name = data.get("name", None) # 1
mt = data.get("mt", self.default_mt) # 2
profiling = data.get("profiling", self.default_profiling) # 3
sequential = data.get("sequential", self.default_sequential) # 3b
auto_link = data.get("auto_link", self.default_auto_link) # 4
auto_input = data.get("auto_input", self.default_auto_input) # 5
auto_output = data.get("auto_output", self.default_auto_output) # 6
auto_parameter = data.get("auto_parameter", self.default_auto_parameter) # 6b
auto_local = data.get("auto_local", self.default_auto_local) # 6c
step_size = data.get("step_size", self.default_step_size) # 7
ts_multiplier = data.get("ts_multiplier", self.default_ts_multiplier) # 7b
node = AssemblyNode(name, step_size=step_size, auto_link=auto_link, mt=mt, profiling=profiling,
sequential=sequential,
auto_input=auto_input, auto_output=auto_output, auto_parameter=auto_parameter,
auto_local=auto_local, ts_multiplier=ts_multiplier)
for key, value in data.items():
if key in ('name', 'step_size', 'auto_link', 'auto_input', 'auto_output', 'mt', 'profiling', 'sequential',
'auto_parameter', 'auto_local', 'ts_multiplier'):
continue # Already read
elif key == "container": # 8
if not isinstance(value, list):
raise AssemblyError("JSON: 'container' keyword should define a list.")
for sub_data in value:
node.add_sub_node(self._json_decode_node(sub_data))
elif key == "fmu": # 9
if not isinstance(value, list):
raise AssemblyError("JSON: 'fmu' keyword should define a list.")
for fmu in value:
node.add_fmu(fmu)
elif key == "input": # 10
self._json_decode_keyword('input', value, node.add_input)
elif key == "output": # 11
self._json_decode_keyword('output', value, node.add_output)
elif key == "link": # 12
self._json_decode_keyword('link', value, node.add_link)
elif key == "start": # 13
self._json_decode_keyword('start', value, node.add_start_value)
elif key == "drop": # 14
self._json_decode_keyword('drop', value, node.add_drop_port)
else:
logger.error(f"JSON: unexpected keyword {key}. Skipped.")
return node
@staticmethod
def _json_decode_keyword(keyword: str, value, function):
if not isinstance(value, list):
raise AssemblyError(f"JSON: '{keyword}' keyword should define a list.")
for line in value:
if not isinstance(line, list):
raise AssemblyError(f"JSON: unexpected '{keyword}' value: {line}.")
try:
function(*line)
except TypeError:
raise AssemblyError(f"JSON: '{keyword}' value does not contain right number of fields: {line}.")
def write_json(self, filename: Union[str, Path]):
"""Export the assembly as a JSON file.
Args:
filename (str | Path): Output filename, relative to `fmu_directory`.
"""
with open(self.fmu_directory / filename, "wt") as file:
data = self._json_encode_node(self.root)
json.dump(data, file, indent=2)
def _json_encode_node(self, node: AssemblyNode) -> Dict[str, Any]:
json_node = dict()
json_node["name"] = node.name # 1
json_node["mt"] = node.mt # 2
json_node["profiling"] = node.profiling # 3
json_node["sequential"] = node.sequential # 3b
json_node["auto_link"] = node.auto_link # 4
json_node["auto_input"] = node.auto_input # 5
json_node["auto_output"] = node.auto_output # 6
json_node["auto_parameter"] = node.auto_parameter # 6b
json_node["auto_local"] = node.auto_local # 6c
if node.step_size:
json_node["step_size"] = node.step_size # 7
if node.ts_multiplier:
json_node["ts_multiplier"] = node.ts_multiplier # 7b
if node.children:
json_node["container"] = [self._json_encode_node(child) for child in node.children.values()] # 8
if node.fmu_names_list:
json_node["fmu"] = [f"{fmu_name}" for fmu_name in sorted(node.fmu_names_list)] # 9
if node.input_ports:
json_node["input"] = [[f"{source}", f"{port.fmu_name}", f"{port.port_name}"] # 10
for port, source in node.input_ports.items()]
if node.output_ports:
json_node["output"] = [[f"{port.fmu_name}", f"{port.port_name}", f"{target}"] # 11
for port, target in node.output_ports.items()]
if node.links:
json_node["link"] = [[f"{link.from_port.fmu_name}", f"{link.from_port.port_name}", # 12
f"{link.to_port.fmu_name}", f"{link.to_port.port_name}"]
for link in node.links]
if node.start_values:
json_node["start"] = [[f"{port.fmu_name}", f"{port.port_name}", value] # 13
for port, value in node.start_values.items()]
if node.drop_ports:
json_node["drop"] = [[f"{port.fmu_name}", f"{port.port_name}"] for port in node.drop_ports] # 14
return json_node
def read_ssp(self):
"""Parse an SSP (System Structure and Parameterization) archive.
Extracts embedded FMUs and the `SystemStructure.ssd` file, then builds
the assembly tree from the SSD connections.
Warning:
This feature is in alpha stage.
"""
logger.warning("This feature is ALPHA stage.")
with zipfile.ZipFile(self.fmu_directory / self.filename) as zin:
for file in zin.filelist:
if file.filename.endswith(".fmu") or file.filename.endswith(".ssd"):
zin.extract(file, path=self.fmu_directory)
logger.debug(f"Extracted: {file.filename}")
self.add_transient_file(file.filename)
self.description_pathname = self.fmu_directory / "SystemStructure.ssd"
if self.description_pathname.is_file():
sdd = SSDParser(step_size=self.default_step_size, auto_link=False,
mt=self.default_mt, profiling=self.default_profiling,
auto_input=False, auto_output=False)
self.root = sdd.parse(self.description_pathname)
self.root.name = str(self.filename.with_suffix(".fmu"))
def make_fmu(self, dump_json=False, fmi_version=2, datalog=False, filename=None):
"""Build the FMU Container from the loaded assembly.
Args:
dump_json (bool): If `True`, also export a JSON dump of the assembly.
fmi_version (int): FMI version for the container interface (`2` or `3`).
datalog (bool): If `True`, enable data logging inside the container.
filename (str | None): Override the output filename.
"""
self.root.make_fmu(self.fmu_directory, debug=self.debug, description_pathname=self.description_pathname,
fmi_version=fmi_version, datalog=datalog, filename=filename)
if dump_json:
dump_file = Path(self.input_pathname.stem + "-dump").with_suffix(".json")
logger.info(f"Dump Json '{dump_file}'")
self.write_json(dump_file)