|
5 | 5 |
|
6 | 6 | import abc |
7 | 7 | import ast |
| 8 | +import csv |
8 | 9 | from dataclasses import dataclass |
9 | 10 | import logging |
10 | 11 | import numbers |
@@ -936,6 +937,29 @@ def setOptimizationOptions( |
936 | 937 | datatype="optimization-option", |
937 | 938 | overridedata=None) |
938 | 939 |
|
| 940 | + @staticmethod |
| 941 | + def toInputs(data: dict[str, list[float]]) -> dict[str, list[tuple[float, float]]]: |
| 942 | + """ |
| 943 | + Converts a dictionary of lists (from pandas DataFrame.to_dict(orient='list')) |
| 944 | + into the OMPython setInputs input format. |
| 945 | +
|
| 946 | + Example: mod.setInputs(**toInputs(pdf.to_dict(orient='list'))) |
| 947 | +
|
| 948 | + Assumes the dictionary contains a key named 'time'. |
| 949 | + """ |
| 950 | + if "time" not in data: |
| 951 | + raise ValueError("The provided data must contain a 'time' key.") |
| 952 | + |
| 953 | + time_series = data["time"] |
| 954 | + |
| 955 | + inputs = { |
| 956 | + var_name: list(zip(time_series, values)) |
| 957 | + for var_name, values in data.items() |
| 958 | + if var_name != "time" |
| 959 | + } |
| 960 | + |
| 961 | + return inputs |
| 962 | + |
939 | 963 | def setInputs( |
940 | 964 | self, |
941 | 965 | *args: Any, |
@@ -998,6 +1022,44 @@ def setInputs( |
998 | 1022 |
|
999 | 1023 | return True |
1000 | 1024 |
|
| 1025 | + def setInputsCSV( |
| 1026 | + self, |
| 1027 | + csvfile: os.PathLike, |
| 1028 | + ) -> None: |
| 1029 | + """ |
| 1030 | + Read content from a CSV file and use it to define the time based input data. |
| 1031 | + """ |
| 1032 | + |
| 1033 | + # real type is 'dict[str, list[tuple[float, float]]]' - 'dict[str, Any]' is used to make setInputs() happy |
| 1034 | + inputs: dict[str, Any] = {} |
| 1035 | + try: |
| 1036 | + with open(csvfile, newline='') as csvfh: |
| 1037 | + dialect = csv.Sniffer().sniff(csvfh.read(1024)) |
| 1038 | + csvfh.seek(0) |
| 1039 | + reader = csv.DictReader(csvfh, dialect=dialect) |
| 1040 | + |
| 1041 | + keys: list[str] = [] |
| 1042 | + for idx, line in enumerate(reader): |
| 1043 | + if not keys: |
| 1044 | + keys = list(line.keys()) |
| 1045 | + for var in keys[1:]: |
| 1046 | + if var in inputs: |
| 1047 | + raise ModelicaSystemError(f"Error reading {csvfile}: duplicated column {var}!") |
| 1048 | + inputs[var] = [] |
| 1049 | + try: |
| 1050 | + # use key[0] as time; all other columns use the header as name |
| 1051 | + for var in keys[1:]: |
| 1052 | + inputs[var].append((float(line[keys[0]]), float(line[var]))) |
| 1053 | + except (ValueError, TypeError) as exc2: |
| 1054 | + raise ModelicaSystemError(f"Invalid value reading {csvfile} line {idx}/{var}: " |
| 1055 | + f"{line}!") from exc2 |
| 1056 | + |
| 1057 | + except IOError as exc1: |
| 1058 | + raise ModelicaSystemError(f"Error reading {csvfile}: {exc1}") from exc1 |
| 1059 | + |
| 1060 | + if inputs: |
| 1061 | + self.setInputs(**inputs) |
| 1062 | + |
1001 | 1063 | def _createCSVData(self, csvfile: Optional[OMPathABC] = None) -> OMPathABC: |
1002 | 1064 | """ |
1003 | 1065 | Create a csv file with inputs for the simulation/optimization of the model. If csvfile is provided as argument, |
|
0 commit comments