Skip to content

Index

mast

logger_model

nrg

reader
symphoniepro_txt_reader
SymphonieProSiteInfo

Symphonie Text File format header object. This class contains site basic information on the data log file.

Source code in src/nrg_parser/mast/logger_model/nrg/reader/symphoniepro_txt_reader.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
class SymphonieProSiteInfo:
    """
    Symphonie Text File format header object.
    This class contains site basic information on the data log file.
    """

    site_number = None
    site_description = None
    project = None
    tower = None
    location = None
    coordinate_system = None
    latitude = None
    longitude = None
    elevation = None
    timezone = None
    site_units = None
    logger_date = None
    logger_model = None
    logger_sn = None
    logger_firmware = None
    ipack_model = None
    ipack_sn = None
    ipack_firmware = None
SymphonieProTxtReader

Symphonie Text File Reader This class takes Symphonie Desktop Pro rld -> txt converted file.

Source code in src/nrg_parser/mast/logger_model/nrg/reader/symphoniepro_txt_reader.py 
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
class SymphonieProTxtReader:
    """
    Symphonie Text File Reader
    This class takes Symphonie Desktop Pro rld -> txt converted file.
    """

    def __init__(self, txt_filepath: str):
        """
        Initializes SymphonieProTxt by reading/parsing its file content,
        both headers and timeseries data.

        Parameters:
        -----------
            txt_filepath: symphonie pro txt data file path

        Attributes:
        -----------
        first_timestamp: datetime
        last_timestamp: datetime
        header_sections_dict: dict
        sensor_history_list: list
        header_line_ctr: int 
        """
        # initialize
        self.header_sections_dict = {}
        self.sensor_history_list = []
        self.header_line_ctr = 0
        self.timeseries_data = pd.DataFrame()
        self.first_timestamp = None
        self.last_timestamp = None

        # do parameter check here, if file exist, file not empty

        try:
            # read headers start
            line_ctr = 0

            logger.debug(f"opening file: {txt_filepath}")
            with open(txt_filepath, "rt") as file_data:
                for line in file_data:
                    line_ctr += 1
                    line = line.strip()

                    if line == "Data":
                        break
                    if line == "Sensor History":
                        continue

                    # initialize section settings:
                    section_name = line
                    section = {}

                    # sensor history -- channel definition
                    if line.startswith("Channel:"):
                        split_arr = line.split(":", 1)
                        section[split_arr[0].strip()] = split_arr[1].strip()
                        section_name = "Ch_" + split_arr[1].strip()

                    # header section loop
                    for line in file_data:
                        line_ctr += 1
                        str_line = line.strip()
                        if (
                            str_line == "Data"
                        ):  # means timeseries section, break from section loop
                            break
                        elif (
                            str_line == ""
                        ):  # if empty, it means new section break from section loop
                            break
                        elif str_line.find(":") != -1:  # normal header data
                            split_arr = str_line.split(":", 1)
                            section[split_arr[0].strip()] = util_func.try_string(
                                split_arr[1].strip()
                            )

                    if section_name.startswith("Ch_"):
                        self.sensor_history_list.append(section)
                    else:
                        self.header_sections_dict[section_name] = section

            self.header_line_ctr = line_ctr

            # read headers end

            # read timeseries data
            try:
                self.timeseries_data = pd.read_csv(
                    txt_filepath,
                    skiprows=self.header_line_ctr,
                    sep="\t",
                    encoding="utf-8",
                )
            except Exception:
                logger.debug(traceback.format_exc())
                pass  # file has no data

            # get first and last timestamp if timeseries data is available
            if len(self.timeseries_data) > 0:
                self.timeseries_data["Timestamp"] = pd.to_datetime(
                    self.timeseries_data["Timestamp"], errors="coerce"
                )
                self.first_timestamp = self.timeseries_data["Timestamp"].min()
                self.last_timestamp = self.timeseries_data["Timestamp"].max()

        except Exception as e:
            # handle any possible exception
            logging.error(f"Failed to load file {txt_filepath} error: str{e}")

    def get_all_metadata(self):
        self.get_header()
        self.get_site_info()
        self.get_iea_logger_main_config()
        self.get_iea_sensor_config()

    def get_header(self, header_section_name=None, header=None):
        """
        Helper method that returns symphonie pro txt file in dictionary format if
        header is not given, else return header value.

        Args:
            header_section_name: section headers name in symphonie pro txt file
            header: headers name in symphonie pro txt file
        """

        if header_section_name is None:
            return self.header_sections_dict  # return all headers in dict

        if header_section_name in self.header_sections_dict:
            section = self.header_sections_dict[header_section_name]
            if header is not None:
                if header in section:
                    self.header = section[header]
                    return section[header]
                else:
                    logging.debug(f"{header} header does not exists.")
                    self.header = None
                    return None
            else:
                self.header = section
                return section

        else:
            logging.debug(f"{header_section_name} section does not exists")
            self.header = None
            return None

    def get_site_info(self):
        """
        Helper method that returns SymphonieProSiteInfo
        SymphoniePRO basic site information from the headers

        Returns:
            site_info - Symphonie Pro site information
        """
        site_info = SymphonieProSiteInfo()

        # Export Parameters
        exp_params = "Export Parameters"
        site_info.site_number = self.get_header(
            header_section_name=exp_params, header="Site Number"
        )

        # site properties
        site_properties = "Site Properties"
        site_info.site_description = self.get_header(
            header_section_name=site_properties, header="Site Description"
        )
        site_info.project = self.get_header(
            header_section_name=site_properties, header="Project"
        )
        site_info.tower = self.get_header(
            header_section_name=site_properties, header="Tower"
        )
        site_info.location = self.get_header(
            header_section_name=site_properties, header="Location"
        )
        site_info.coordinate_system = self.get_header(
            header_section_name=site_properties, header="Coordinate System"
        )
        site_info.latitude = util_func.try_float(
            self.get_header(header_section_name=site_properties, header="Latitude")
        )
        site_info.longitude = util_func.try_float(
            self.get_header(header_section_name=site_properties, header="Longitude")
        )
        site_info.elevation = util_func.try_int(
            self.get_header(header_section_name=site_properties, header="Elevation")
        )
        site_info.timezone = self.get_header(
            header_section_name=site_properties, header="Time Zone"
        )
        site_info.site_units = self.get_header(
            header_section_name=site_properties, header="Site Units"
        )

        # logger history
        logger_history = "Logger History"
        site_info.logger_date = self.get_header(
            header_section_name=logger_history, header="Date"
        )
        site_info.logger_model = self.get_header(
            header_section_name=logger_history, header="Model"
        )
        site_info.logger_sn = self.get_header(
            header_section_name=logger_history, header="Serial Number"
        )
        site_info.logger_firmware = self.get_header(
            header_section_name=logger_history, header="Firmware"
        )

        # iPack History
        ipack_history = "iPack History"
        site_info.ipack_model = self.get_header(
            header_section_name=ipack_history, header="Model"
        )
        site_info.ipack_sn = self.get_header(
            header_section_name=ipack_history, header="Serial Number"
        )
        site_info.ipack_firmware = self.get_header(
            header_section_name=ipack_history, header="Firmware"
        )

        self.site_info = site_info

    def get_iea_logger_main_config(self):
        """
        Helper method that returns the logger main configuration in IEATask43 format
        Returns:
            logger_main_config_dict: logger main configuration in dictionary format
        """

        logger_main_config_dict = {}

        # header sections
        logger_history = "Logger History"
        exp_params = "Export Parameters"
        site_properties = "Site Properties"

        logger_main_config_dict["logger_serial_number"] = self.get_header(
            header_section_name=logger_history, header="Serial Number"
        )
        logger_main_config_dict["logger_model_name"] = self.get_header(
            header_section_name=logger_history, header="Model"
        )
        logger_main_config_dict["logger_id"] = self.get_header(
            header_section_name=exp_params, header="Site Number"
        )
        logger_main_config_dict["logger_name"] = self.get_header(
            header_section_name=site_properties, header="Project"
        )

        logger_main_config_dict["date_from"] = self.first_timestamp
        logger_main_config_dict["date_to"] = self.last_timestamp

        logger_main_config_dict["latitude_ddeg"] = util_func.try_float(
            self.get_header(header_section_name=site_properties, header="Latitude")
        )
        logger_main_config_dict["longitude_ddeg"] = util_func.try_float(
            self.get_header(header_section_name=site_properties, header="Longitude")
        )
        logger_main_config_dict["measurement_station_type_id"] = "mast"
        logger_main_config_dict["offset_from_utc_hrs"] = util_func.try_float(
            self.get_header(header_section_name=site_properties, header="Time Zone")
            .replace("UTC", "")
            .replace(":", ".")
        )
        logger_main_config_dict["firmware_version"] = self.get_header(
            header_section_name=logger_history, header="Firmware"
        )
        self.logger_main_config_dict = logger_main_config_dict

    def get_iea_sensor_config(self):
        """
        Helper method that returns the measurement point and sensor configuration in IEATask43 format
        Returns:
            sensor_config_dict_list: list of measurement points and sensor configurations (dict)
        """
        sensor_config_dict_list = []
        column_headers = list(self.timeseries_data.columns.values)

        # mast sensor configuration
        for ch_info in self.sensor_history_list:
            sensor_config_dict = {}
            match ch_info["Type"]:
                case "Calculated":
                    pass
                    # sensor_config_dict["sensor_serial_number"] = None
                    # sensor_config_dict["logger_height"] = None
                    # sensor_config_dict["logger_slope"] = None
                    # sensor_config_dict["boom_orientation_deg"] = None
                case _:
                    sensor_config_dict["connection_channel"] = util_func.try_int(ch_info["Channel"])
                    sensor_config_dict["sensor_serial_number"] = ch_info["Serial Number"]
                    sensor_config_dict["logger_height"] = util_func.try_float(ch_info["Height"])
                    sensor_config_dict["logger_slope"] = util_func.try_float(
                        ch_info["Scale Factor"]
                    )
                    sensor_config_dict["logger_offset"] = util_func.try_float(ch_info["Offset"])
                    sensor_config_dict["boom_orientation_deg"] = util_func.try_float(
                        ch_info["Bearing"]
                    )
                    sensor_config_dict["boom_orientation_deg"] = util_func.try_float(
                        ch_info["Bearing"]
                    )
            if ch_info["Units"] is not None:
                if ch_info["Units"].lower() in iea_mapping_config.IEA_UNIT_DICT:
                    # map unit to iea task43
                    sensor_config_dict[
                        "measurement_unit_id"
                    ] = iea_mapping_config.IEA_UNIT_DICT[ch_info["Units"].lower()]
                else:
                    sensor_config_dict["measurement_unit_id"] = "-"
            else:
                sensor_config_dict["measurement_unit_id"] = None

            if ch_info["Type"] == "Vane":
                sensor_config_dict[
                    "vane_dead_band_orientation_deg"
                ] = util_func.try_float(ch_info["Vane Mounting Angle"])

            match sensor_config_dict["measurement_unit_id"]:
                case "m/s":
                    sensor_config_dict["measurement_type"] = "wind_speed"
                    sensor_config_dict["sensor_type"] = "anemometer"
                case "deg":
                    sensor_config_dict["measurement_type"] = "wind_direction"
                    sensor_config_dict["sensor_type"] = "wind_vane"
                case "deg_C":
                    sensor_config_dict["measurement_type"] = "air_temperature"
                    sensor_config_dict["sensor_type"] = "thermometer"
                case "deg_F":
                    sensor_config_dict["measurement_type"] = "air_temperature"
                    sensor_config_dict["sensor_type"] = "thermometer"
                case "K":
                    sensor_config_dict["measurement_type"] = "air_temperature"
                    sensor_config_dict["sensor_type"] = "thermometer"
                case "%":
                    sensor_config_dict["measurement_type"] = "relative_humidity"
                    sensor_config_dict["sensor_type"] = "hygrometer"
                case "mbar":
                    sensor_config_dict["measurement_type"] = "air_pressure"
                    sensor_config_dict["sensor_type"] = "barometer"
                case "hPa":
                    sensor_config_dict["measurement_type"] = "air_pressure"
                    sensor_config_dict["sensor_type"] = "barometer"
                case "atm":
                    sensor_config_dict["measurement_type"] = "air_pressure"
                    sensor_config_dict["sensor_type"] = "barometer"
                case "mmHg":
                    sensor_config_dict["measurement_type"] = "air_pressure"
                    sensor_config_dict["sensor_type"] = "barometer"
                case "inHg":
                    sensor_config_dict["measurement_type"] = "air_pressure"
                    sensor_config_dict["sensor_type"] = "barometer"
                case "kg/m^2":
                    sensor_config_dict["measurement_type"] = "air_pressure"
                    sensor_config_dict["sensor_type"] = "barometer"
                case "kg/m^3":
                    sensor_config_dict["measurement_type"] = "air_density"
                    sensor_config_dict["sensor_type"] = "thermometer"
                case "V":
                    sensor_config_dict["measurement_type"] = "voltage"
                    sensor_config_dict["sensor_type"] = "voltmeter"
                case "mA":
                    sensor_config_dict["measurement_type"] = "current"
                    sensor_config_dict["sensor_type"] = "ammeter"
                case "A":
                    sensor_config_dict["measurement_type"] = "current"
                    sensor_config_dict["sensor_type"] = "ammeter"
                case "ohm":
                    sensor_config_dict["measurement_type"] = "resistance"
                    sensor_config_dict["sensor_type"] = "ice_detection_sensor"
                case "mm":
                    sensor_config_dict["measurement_type"] = "precipitation"
                    sensor_config_dict["sensor_type"] = "rain_gauge"
                case "W/m^2":
                    sensor_config_dict[
                        "measurement_type"
                    ] = "global_horizontal_irradiance"
                    sensor_config_dict["sensor_type"] = "pyranometer"
                case _:
                    sensor_config_dict["sensor_type"] = "other"
                    sensor_config_dict["measurement_type"] = "other"

            sensor_config_dict["date_from"] = self.first_timestamp
            sensor_config_dict["date_to"] = self.last_timestamp

            column_name_list = []
            ch_header = [
                columns
                for columns in column_headers
                if columns.startswith("Ch" + ch_info["Channel"] + "_")
            ]

            for column in ch_header:
                # parse statistic type
                chunks = column.split("_")
                statistic_type = chunks[len(chunks) - 2].lower()

                if statistic_type == "gustdir":
                    statistic_type = "gust"

                if statistic_type in iea_mapping_config.IEA_STATISTIC_TYPE:
                    statistic_type = statistic_type
                else:
                    statistic_type = "avg"

                column_name = column + "#" + statistic_type
                column_name_list.append(column_name)

            sensor_config_dict["column_name_list"] = ";".join(column_name_list)
            sensor_config_dict_list.append(sensor_config_dict)

        self.sensor_config_dict_list = sensor_config_dict_list
__init__(txt_filepath) 
txt_filepath: symphonie pro txt data file path

first_timestamp: datetime last_timestamp: datetime header_sections_dict: dict sensor_history_list: list header_line_ctr: int

Source code in src/nrg_parser/mast/logger_model/nrg/reader/symphoniepro_txt_reader.py 
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def __init__(self, txt_filepath: str):
    """
    Initializes SymphonieProTxt by reading/parsing its file content,
    both headers and timeseries data.

    Parameters:
    -----------
        txt_filepath: symphonie pro txt data file path

    Attributes:
    -----------
    first_timestamp: datetime
    last_timestamp: datetime
    header_sections_dict: dict
    sensor_history_list: list
    header_line_ctr: int 
    """
    # initialize
    self.header_sections_dict = {}
    self.sensor_history_list = []
    self.header_line_ctr = 0
    self.timeseries_data = pd.DataFrame()
    self.first_timestamp = None
    self.last_timestamp = None

    # do parameter check here, if file exist, file not empty

    try:
        # read headers start
        line_ctr = 0

        logger.debug(f"opening file: {txt_filepath}")
        with open(txt_filepath, "rt") as file_data:
            for line in file_data:
                line_ctr += 1
                line = line.strip()

                if line == "Data":
                    break
                if line == "Sensor History":
                    continue

                # initialize section settings:
                section_name = line
                section = {}

                # sensor history -- channel definition
                if line.startswith("Channel:"):
                    split_arr = line.split(":", 1)
                    section[split_arr[0].strip()] = split_arr[1].strip()
                    section_name = "Ch_" + split_arr[1].strip()

                # header section loop
                for line in file_data:
                    line_ctr += 1
                    str_line = line.strip()
                    if (
                        str_line == "Data"
                    ):  # means timeseries section, break from section loop
                        break
                    elif (
                        str_line == ""
                    ):  # if empty, it means new section break from section loop
                        break
                    elif str_line.find(":") != -1:  # normal header data
                        split_arr = str_line.split(":", 1)
                        section[split_arr[0].strip()] = util_func.try_string(
                            split_arr[1].strip()
                        )

                if section_name.startswith("Ch_"):
                    self.sensor_history_list.append(section)
                else:
                    self.header_sections_dict[section_name] = section

        self.header_line_ctr = line_ctr

        # read headers end

        # read timeseries data
        try:
            self.timeseries_data = pd.read_csv(
                txt_filepath,
                skiprows=self.header_line_ctr,
                sep="\t",
                encoding="utf-8",
            )
        except Exception:
            logger.debug(traceback.format_exc())
            pass  # file has no data

        # get first and last timestamp if timeseries data is available
        if len(self.timeseries_data) > 0:
            self.timeseries_data["Timestamp"] = pd.to_datetime(
                self.timeseries_data["Timestamp"], errors="coerce"
            )
            self.first_timestamp = self.timeseries_data["Timestamp"].min()
            self.last_timestamp = self.timeseries_data["Timestamp"].max()

    except Exception as e:
        # handle any possible exception
        logging.error(f"Failed to load file {txt_filepath} error: str{e}")
get_header(header_section_name=None, header=None)

Helper method that returns symphonie pro txt file in dictionary format if header is not given, else return header value.

Parameters:

Name Type Description Default
header_section_name

section headers name in symphonie pro txt file

 None
header

headers name in symphonie pro txt file

 None
Source code in src/nrg_parser/mast/logger_model/nrg/reader/symphoniepro_txt_reader.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def get_header(self, header_section_name=None, header=None):
    """
    Helper method that returns symphonie pro txt file in dictionary format if
    header is not given, else return header value.

    Args:
        header_section_name: section headers name in symphonie pro txt file
        header: headers name in symphonie pro txt file
    """

    if header_section_name is None:
        return self.header_sections_dict  # return all headers in dict

    if header_section_name in self.header_sections_dict:
        section = self.header_sections_dict[header_section_name]
        if header is not None:
            if header in section:
                self.header = section[header]
                return section[header]
            else:
                logging.debug(f"{header} header does not exists.")
                self.header = None
                return None
        else:
            self.header = section
            return section

    else:
        logging.debug(f"{header_section_name} section does not exists")
        self.header = None
        return None
get_iea_logger_main_config()

Helper method that returns the logger main configuration in IEATask43 format

Returns:

Name Type Description
logger_main_config_dict

logger main configuration in dictionary format
Source code in src/nrg_parser/mast/logger_model/nrg/reader/symphoniepro_txt_reader.py 
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
def get_iea_logger_main_config(self):
    """
    Helper method that returns the logger main configuration in IEATask43 format
    Returns:
        logger_main_config_dict: logger main configuration in dictionary format
    """

    logger_main_config_dict = {}

    # header sections
    logger_history = "Logger History"
    exp_params = "Export Parameters"
    site_properties = "Site Properties"

    logger_main_config_dict["logger_serial_number"] = self.get_header(
        header_section_name=logger_history, header="Serial Number"
    )
    logger_main_config_dict["logger_model_name"] = self.get_header(
        header_section_name=logger_history, header="Model"
    )
    logger_main_config_dict["logger_id"] = self.get_header(
        header_section_name=exp_params, header="Site Number"
    )
    logger_main_config_dict["logger_name"] = self.get_header(
        header_section_name=site_properties, header="Project"
    )

    logger_main_config_dict["date_from"] = self.first_timestamp
    logger_main_config_dict["date_to"] = self.last_timestamp

    logger_main_config_dict["latitude_ddeg"] = util_func.try_float(
        self.get_header(header_section_name=site_properties, header="Latitude")
    )
    logger_main_config_dict["longitude_ddeg"] = util_func.try_float(
        self.get_header(header_section_name=site_properties, header="Longitude")
    )
    logger_main_config_dict["measurement_station_type_id"] = "mast"
    logger_main_config_dict["offset_from_utc_hrs"] = util_func.try_float(
        self.get_header(header_section_name=site_properties, header="Time Zone")
        .replace("UTC", "")
        .replace(":", ".")
    )
    logger_main_config_dict["firmware_version"] = self.get_header(
        header_section_name=logger_history, header="Firmware"
    )
    self.logger_main_config_dict = logger_main_config_dict
get_iea_sensor_config()

Helper method that returns the measurement point and sensor configuration in IEATask43 format

Returns:

Name Type Description
sensor_config_dict_list

list of measurement points and sensor configurations (dict)
Source code in src/nrg_parser/mast/logger_model/nrg/reader/symphoniepro_txt_reader.py 
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
def get_iea_sensor_config(self):
    """
    Helper method that returns the measurement point and sensor configuration in IEATask43 format
    Returns:
        sensor_config_dict_list: list of measurement points and sensor configurations (dict)
    """
    sensor_config_dict_list = []
    column_headers = list(self.timeseries_data.columns.values)

    # mast sensor configuration
    for ch_info in self.sensor_history_list:
        sensor_config_dict = {}
        match ch_info["Type"]:
            case "Calculated":
                pass
                # sensor_config_dict["sensor_serial_number"] = None
                # sensor_config_dict["logger_height"] = None
                # sensor_config_dict["logger_slope"] = None
                # sensor_config_dict["boom_orientation_deg"] = None
            case _:
                sensor_config_dict["connection_channel"] = util_func.try_int(ch_info["Channel"])
                sensor_config_dict["sensor_serial_number"] = ch_info["Serial Number"]
                sensor_config_dict["logger_height"] = util_func.try_float(ch_info["Height"])
                sensor_config_dict["logger_slope"] = util_func.try_float(
                    ch_info["Scale Factor"]
                )
                sensor_config_dict["logger_offset"] = util_func.try_float(ch_info["Offset"])
                sensor_config_dict["boom_orientation_deg"] = util_func.try_float(
                    ch_info["Bearing"]
                )
                sensor_config_dict["boom_orientation_deg"] = util_func.try_float(
                    ch_info["Bearing"]
                )
        if ch_info["Units"] is not None:
            if ch_info["Units"].lower() in iea_mapping_config.IEA_UNIT_DICT:
                # map unit to iea task43
                sensor_config_dict[
                    "measurement_unit_id"
                ] = iea_mapping_config.IEA_UNIT_DICT[ch_info["Units"].lower()]
            else:
                sensor_config_dict["measurement_unit_id"] = "-"
        else:
            sensor_config_dict["measurement_unit_id"] = None

        if ch_info["Type"] == "Vane":
            sensor_config_dict[
                "vane_dead_band_orientation_deg"
            ] = util_func.try_float(ch_info["Vane Mounting Angle"])

        match sensor_config_dict["measurement_unit_id"]:
            case "m/s":
                sensor_config_dict["measurement_type"] = "wind_speed"
                sensor_config_dict["sensor_type"] = "anemometer"
            case "deg":
                sensor_config_dict["measurement_type"] = "wind_direction"
                sensor_config_dict["sensor_type"] = "wind_vane"
            case "deg_C":
                sensor_config_dict["measurement_type"] = "air_temperature"
                sensor_config_dict["sensor_type"] = "thermometer"
            case "deg_F":
                sensor_config_dict["measurement_type"] = "air_temperature"
                sensor_config_dict["sensor_type"] = "thermometer"
            case "K":
                sensor_config_dict["measurement_type"] = "air_temperature"
                sensor_config_dict["sensor_type"] = "thermometer"
            case "%":
                sensor_config_dict["measurement_type"] = "relative_humidity"
                sensor_config_dict["sensor_type"] = "hygrometer"
            case "mbar":
                sensor_config_dict["measurement_type"] = "air_pressure"
                sensor_config_dict["sensor_type"] = "barometer"
            case "hPa":
                sensor_config_dict["measurement_type"] = "air_pressure"
                sensor_config_dict["sensor_type"] = "barometer"
            case "atm":
                sensor_config_dict["measurement_type"] = "air_pressure"
                sensor_config_dict["sensor_type"] = "barometer"
            case "mmHg":
                sensor_config_dict["measurement_type"] = "air_pressure"
                sensor_config_dict["sensor_type"] = "barometer"
            case "inHg":
                sensor_config_dict["measurement_type"] = "air_pressure"
                sensor_config_dict["sensor_type"] = "barometer"
            case "kg/m^2":
                sensor_config_dict["measurement_type"] = "air_pressure"
                sensor_config_dict["sensor_type"] = "barometer"
            case "kg/m^3":
                sensor_config_dict["measurement_type"] = "air_density"
                sensor_config_dict["sensor_type"] = "thermometer"
            case "V":
                sensor_config_dict["measurement_type"] = "voltage"
                sensor_config_dict["sensor_type"] = "voltmeter"
            case "mA":
                sensor_config_dict["measurement_type"] = "current"
                sensor_config_dict["sensor_type"] = "ammeter"
            case "A":
                sensor_config_dict["measurement_type"] = "current"
                sensor_config_dict["sensor_type"] = "ammeter"
            case "ohm":
                sensor_config_dict["measurement_type"] = "resistance"
                sensor_config_dict["sensor_type"] = "ice_detection_sensor"
            case "mm":
                sensor_config_dict["measurement_type"] = "precipitation"
                sensor_config_dict["sensor_type"] = "rain_gauge"
            case "W/m^2":
                sensor_config_dict[
                    "measurement_type"
                ] = "global_horizontal_irradiance"
                sensor_config_dict["sensor_type"] = "pyranometer"
            case _:
                sensor_config_dict["sensor_type"] = "other"
                sensor_config_dict["measurement_type"] = "other"

        sensor_config_dict["date_from"] = self.first_timestamp
        sensor_config_dict["date_to"] = self.last_timestamp

        column_name_list = []
        ch_header = [
            columns
            for columns in column_headers
            if columns.startswith("Ch" + ch_info["Channel"] + "_")
        ]

        for column in ch_header:
            # parse statistic type
            chunks = column.split("_")
            statistic_type = chunks[len(chunks) - 2].lower()

            if statistic_type == "gustdir":
                statistic_type = "gust"

            if statistic_type in iea_mapping_config.IEA_STATISTIC_TYPE:
                statistic_type = statistic_type
            else:
                statistic_type = "avg"

            column_name = column + "#" + statistic_type
            column_name_list.append(column_name)

        sensor_config_dict["column_name_list"] = ";".join(column_name_list)
        sensor_config_dict_list.append(sensor_config_dict)

    self.sensor_config_dict_list = sensor_config_dict_list
get_site_info()

Helper method that returns SymphonieProSiteInfo SymphoniePRO basic site information from the headers

Returns:

Type Description

site_info - Symphonie Pro site information
Source code in src/nrg_parser/mast/logger_model/nrg/reader/symphoniepro_txt_reader.py 
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
def get_site_info(self):
    """
    Helper method that returns SymphonieProSiteInfo
    SymphoniePRO basic site information from the headers

    Returns:
        site_info - Symphonie Pro site information
    """
    site_info = SymphonieProSiteInfo()

    # Export Parameters
    exp_params = "Export Parameters"
    site_info.site_number = self.get_header(
        header_section_name=exp_params, header="Site Number"
    )

    # site properties
    site_properties = "Site Properties"
    site_info.site_description = self.get_header(
        header_section_name=site_properties, header="Site Description"
    )
    site_info.project = self.get_header(
        header_section_name=site_properties, header="Project"
    )
    site_info.tower = self.get_header(
        header_section_name=site_properties, header="Tower"
    )
    site_info.location = self.get_header(
        header_section_name=site_properties, header="Location"
    )
    site_info.coordinate_system = self.get_header(
        header_section_name=site_properties, header="Coordinate System"
    )
    site_info.latitude = util_func.try_float(
        self.get_header(header_section_name=site_properties, header="Latitude")
    )
    site_info.longitude = util_func.try_float(
        self.get_header(header_section_name=site_properties, header="Longitude")
    )
    site_info.elevation = util_func.try_int(
        self.get_header(header_section_name=site_properties, header="Elevation")
    )
    site_info.timezone = self.get_header(
        header_section_name=site_properties, header="Time Zone"
    )
    site_info.site_units = self.get_header(
        header_section_name=site_properties, header="Site Units"
    )

    # logger history
    logger_history = "Logger History"
    site_info.logger_date = self.get_header(
        header_section_name=logger_history, header="Date"
    )
    site_info.logger_model = self.get_header(
        header_section_name=logger_history, header="Model"
    )
    site_info.logger_sn = self.get_header(
        header_section_name=logger_history, header="Serial Number"
    )
    site_info.logger_firmware = self.get_header(
        header_section_name=logger_history, header="Firmware"
    )

    # iPack History
    ipack_history = "iPack History"
    site_info.ipack_model = self.get_header(
        header_section_name=ipack_history, header="Model"
    )
    site_info.ipack_sn = self.get_header(
        header_section_name=ipack_history, header="Serial Number"
    )
    site_info.ipack_firmware = self.get_header(
        header_section_name=ipack_history, header="Firmware"
    )

    self.site_info = site_info

utilities

iea_json

convert_to_iea_json
calculate_distance_m(lat1, lon1, lat2, lon2)

Returns distance, in metere, between one set of longitude/latitude coordinates and another

Source code in src/nrg_parser/mast/utilities/iea_json/convert_to_iea_json.py 
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
def calculate_distance_m(lat1, lon1, lat2, lon2):
    """Returns distance, in metere, between one set of longitude/latitude coordinates and another"""
    lon1, lat1, lon2, lat2 = map(np.radians, [lon1, lat1, lon2, lat2])

    newlon = lon2 - lon1
    newlat = lat2 - lat1

    haver_formula = (
        np.sin(newlat / 2.0) ** 2
        + np.cos(lat1) * np.cos(lat2) * np.sin(newlon / 2.0) ** 2
    )

    dist = 2 * np.arcsin(np.sqrt(haver_formula))
    m = (6367 * dist) * 1000  # 6367 for distance in m for miles use 3958

    return m
create_iea_json_nrg
main()

Creates IEA-task43 json from NRG Symphonie Pro logger data files SymPRO Desktop PRO must be installed. Please specify installation location in nrg_config.py

Parameters:

Name Type Description Default
project_name(str)

ex: Morgan-County

 required
local_path(str)

local path directory where logger data files(.rld) are located

 required

Output: iea_task43.json file under local path

Source code in src/nrg_parser/mast/utilities/iea_json/create_iea_json_nrg.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def main():
    """Creates IEA-task43 json from NRG Symphonie Pro logger data files
    SymPRO Desktop PRO must be installed. Please specify installation location in nrg_config.py
    Args:
        project_name(str): ex: Morgan-County
        local_path(str): local path directory where logger data files(.rld) are located

    Output: iea_task43.json file under local path
    """

    logging.info("IN execute - NRG Systems")
    project_name = sys.argv[1]
    local_path = sys.argv[2]

    # check if path exists

    if not os.path.isdir(local_path):
        logging.error(
            f"Local path {local_path} does not exists. Please confirm rld path directory."
        )
        return

    local_path_dir_converted = os.path.join(local_path, "converted")

    # converted folder doesn't exist, create
    if not os.path.isdir(local_path_dir_converted):
        os.mkdir(local_path_dir_converted)
    else:
        util_func.remove_files_in_dir(local_path_dir_converted)

    process_start = time.perf_counter()

    # convert rld binary - txt conversion
    convert_start = time.perf_counter()

    # get symphonie pro desktop app locations
    if nrg_config.SYMPHONIE_APP is not None:
        symphro_path = nrg_config.SYMPHONIE_APP
    else:
        symphro_path = "C:/Program Files (x86)/Renewable NRG Systems/SymPRO Desktop/SymPRODesktop.exe"

    # # convert .rld file to txt
    converter = nrgpy.local_rld(
        rld_dir=local_path, out_dir=local_path_dir_converted, sympro_path=symphro_path
    )
    converter.convert()

    convert_end = time.perf_counter()
    logging.warning(
        f"PERF: Conversion {round(convert_end - convert_start, 2)} second(s)"
    )

    # read and parse
    logger_config_dict_list, sensor_config_dict_list = read_and_parse(
        local_path_dir_converted
    )
    if len(logger_config_dict_list) > 0 and len(sensor_config_dict_list) > 0:
        create_iea_json(
            logger_config_dict_list,
            sensor_config_dict_list,
            local_path,
            "NRG SymphoniePro",
            project_name,
        )

    # remove all converted files after the run
    util_func.remove_files_in_dir(local_path_dir_converted)

    process_end = time.perf_counter()
    logging.warning(
        f"PERF: PROCESS EXEC OVERALL: {round(process_end - process_start, 2)} second(s)"
    )
    logging.info(f"Please find the output file in {local_path}/iea_task43.json")
    logging.info("OUT execute - NRG Systems")