feat: [sc-219335] [pi-system plugin] Add boundary type selector to recorded type data

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## [Version 1.2.4](https://github.com/dataiku/dss-plugin-pi-server/releases/tag/v1.2.4) - Feature release - 2024-12-12
+
+- Add boundary type selector to recorded data type
+- Add boundary type selector to attribute search connector
+
 ## [Version 1.2.3](https://github.com/dataiku/dss-plugin-pi-server/releases/tag/v1.2.3) - Feature release - 2024-09-26
 
 - Add summaryDuration input (duration of each summary interval)

custom-recipes/pi-system-retrieve-list/recipe.json

@@ -194,7 +194,7 @@
         {
             "name": "boundary_type",
             "label": "Boundary type",
-            "visibilityCondition": "['InterpolatedData'].includes(model.data_type)",
+            "visibilityCondition": "['InterpolatedData','RecordedData'].includes(model.data_type)",
             "type": "SELECT",
             "selectChoices":[
                 {"value": "Inside", "label": "Inside"},

custom-recipes/pi-system-retrieve-list/recipe.py

@@ -48,6 +48,7 @@
 end_time_column = config.get("end_time_column")
 server_url_column = config.get("server_url_column")
 interval, sync_time, boundary_type = get_interpolated_parameters(config)
+record_boundary_type = config.get("boundary_type") if data_type == "RecordedData" else None
 summary_type, summary_duration = get_summary_parameters(config)
 
 network_timer = PerformanceTimer()
@@ -102,6 +103,7 @@
                 interval=interval,
                 sync_time=sync_time,
                 boundary_type=boundary_type,
+                record_boundary_type=record_boundary_type,
                 max_count=max_count,
                 can_raise=False,
                 object_id=object_id,
@@ -117,6 +119,7 @@
                 interval=interval,
                 sync_time=sync_time,
                 boundary_type=boundary_type,
+                record_boundary_type=record_boundary_type,
                 max_count=max_count,
                 can_raise=False,
                 endpoint_type="AF",

plugin.json

@@ -1,6 +1,6 @@
 {
     "id": "pi-system",
-    "version": "1.2.3",
+    "version": "1.2.4",
     "meta": {
         "label": "PI System",
         "description": "Retrieve data from your OSIsoft PI System servers",

python-connectors/pi-system_attribute-search/connector.json

@@ -360,6 +360,17 @@
             ],
             "visibilityCondition": "((model.must_retrieve_metrics) && (model.data_type == 'SummaryData'))"
         },
+        {
+            "name": "boundary_type",
+            "label": "Boundary type",
+            "visibilityCondition": "((model.must_retrieve_metrics) && ['InterpolatedData','RecordedData'].includes(model.data_type))",
+            "type": "SELECT",
+            "selectChoices":[
+                {"value": "Inside", "label": "Inside"},
+                {"value": "Outside", "label": "Outside"}
+            ],
+            "defaultValue": "Inside"
+        },
         {
             "name": "summary_duration",
             "label": "Summary duration",

python-connectors/pi-system_attribute-search/connector.py

@@ -6,7 +6,8 @@
 from osisoft_plugin_common import (
     PISystemConnectorError, RecordsLimit, get_credentials, assert_time_format,
     remove_unwanted_columns, format_output, filter_columns_from_schema, is_child_attribute_path,
-    check_debug_mode, PerformanceTimer, get_max_count, get_summary_parameters, fields_selector
+    check_debug_mode, PerformanceTimer, get_max_count, get_summary_parameters, fields_selector,
+    get_interpolated_parameters
 )
 from osisoft_constants import OSIsoftConstants
 
@@ -36,9 +37,7 @@ def __init__(self, config, plugin_config):
         self.start_time = self.client.parse_pi_time(self.start_time)
         self.end_time = config.get("end_time")
         self.end_time = self.client.parse_pi_time(self.end_time)
-        is_interpolated_data = config.get("data_type", "").endswith("InterpolatedData")
-        self.interval = config.get("interval") if is_interpolated_data else None
-        self.sync_time = config.get("sync_time") if is_interpolated_data else None
+        self.interval, self.sync_time, self.boundary_type = get_interpolated_parameters(config)
         self.sync_time = self.client.parse_pi_time(self.sync_time)
         assert_time_format(self.start_time, error_source="start time")
         assert_time_format(self.end_time, error_source="end time")
@@ -57,6 +56,8 @@ def __init__(self, config, plugin_config):
         self.config = config
         self.summary_type, self.summary_duration = get_summary_parameters(config)
 
+        self.record_boundary_type = config.get("boundary_type") if self.data_type == "RecordedData" else None
+
     def extract_database_webid(self, database_endpoint):
         return database_endpoint.split("/")[-1]
 
@@ -114,8 +115,9 @@ def generate_rows(self, dataset_schema=None, dataset_partitioning=None,
                         selected_fields=fields_selector(self.data_type),
                         max_count=self.max_count,
                         summary_type=self.summary_type,
-                        summary_duration=self.summary_duration
-                        # boundary_type=self.boundary_type
+                        summary_duration=self.summary_duration,
+                        boundary_type=self.boundary_type,
+                        record_boundary_type=self.record_boundary_type
                     ):
                         if limit.is_reached():
                             return

python-lib/osisoft_client.py

@@ -48,7 +48,7 @@ def get_auth(self, auth_type, username, password):
             return None
 
     def recursive_get_rows_from_webid(self, webid, data_type, start_date=None, end_date=None,
-                                      interval=None, sync_time=None, boundary_type=None, selected_fields=None,
+                                      interval=None, sync_time=None, boundary_type=None, record_boundary_type=None, selected_fields=None,
                                       can_raise=True, endpoint_type="event_frames", search_full_hierarchy=None,
                                       max_count=None, summary_type=None, summary_duration=None):
         # Split the time range until no more HTTP 400
@@ -57,7 +57,8 @@ def recursive_get_rows_from_webid(self, webid, data_type, start_date=None, end_d
         while not done:
             logger.info("Attempting download webids from {} to {}".format(start_date, end_date))
             rows = self.get_rows_from_webid(webid, data_type, start_date=start_date, end_date=end_date,
-                                            interval=interval, sync_time=sync_time, boundary_type=boundary_type, selected_fields=selected_fields,
+                                            interval=interval, sync_time=sync_time, boundary_type=boundary_type,
+                                            record_boundary_type=record_boundary_type, selected_fields=selected_fields,
                                             can_raise=can_raise, endpoint_type=endpoint_type, search_full_hierarchy=search_full_hierarchy,
                                             max_count=max_count, summary_type=summary_type, summary_duration=summary_duration)
             counter = 0
@@ -79,7 +80,8 @@ def recursive_get_rows_from_webid(self, webid, data_type, start_date=None, end_d
                     start_timestamp, end_timestamp, half_time_iso = self.halve_time_range(start_date, end_date)
                     first_half_rows = self.recursive_get_rows_from_webid(
                         webid, data_type, start_date=start_timestamp, end_date=half_time_iso,
-                        interval=interval, sync_time=sync_time, boundary_type=boundary_type, selected_fields=selected_fields,
+                        interval=interval, sync_time=sync_time, boundary_type=boundary_type,
+                        record_boundary_type=record_boundary_type, selected_fields=selected_fields,
                         can_raise=can_raise, endpoint_type=endpoint_type, search_full_hierarchy=search_full_hierarchy, max_count=max_count,
                         summary_type=summary_type, summary_duration=summary_duration
                     )
@@ -88,7 +90,8 @@ def recursive_get_rows_from_webid(self, webid, data_type, start_date=None, end_d
                     logger.info("Successfully retrieved first half ({} to {})".format(start_timestamp, half_time_iso))
                     second_half_rows = self.recursive_get_rows_from_webid(
                         webid, data_type, start_date=half_time_iso, end_date=end_timestamp,
-                        interval=interval, sync_time=sync_time, boundary_type=boundary_type, selected_fields=selected_fields,
+                        interval=interval, sync_time=sync_time, boundary_type=boundary_type,
+                        record_boundary_type=record_boundary_type, selected_fields=selected_fields,
                         can_raise=can_raise, endpoint_type=endpoint_type, search_full_hierarchy=search_full_hierarchy, max_count=max_count,
                         summary_type=summary_type, summary_duration=summary_duration
                     )
@@ -109,7 +112,7 @@ def recursive_get_rows_from_webid(self, webid, data_type, start_date=None, end_d
                 done = True
 
     def recursive_get_rows_from_item(self, item, data_type, start_date=None, end_date=None,
-                                     interval=None, sync_time=None, boundary_type=None,
+                                     interval=None, sync_time=None, boundary_type=None, record_boundary_type=None,
                                      can_raise=True, object_id=None, endpoint_type="event_frames", search_full_hierarchy=None,
                                      max_count=None, summary_type=None, summary_duration=None):
         # item can be an pi tag, a path to an element or event frame
@@ -119,7 +122,8 @@ def recursive_get_rows_from_item(self, item, data_type, start_date=None, end_dat
         while not done:
             logger.info("Attempting download items from {} to {}".format(start_date, end_date))
             rows = self.get_rows_from_item(item, data_type, start_date=start_date, end_date=end_date, interval=interval,
-                                           sync_time=sync_time, boundary_type=boundary_type, can_raise=True, object_id=object_id,
+                                           sync_time=sync_time, boundary_type=boundary_type, record_boundary_type=record_boundary_type,
+                                           can_raise=True, object_id=object_id,
                                            search_full_hierarchy=search_full_hierarchy, max_count=max_count,
                                            summary_type=summary_type, summary_duration=summary_duration)
             counter = 0
@@ -141,15 +145,17 @@ def recursive_get_rows_from_item(self, item, data_type, start_date=None, end_dat
                     start_timestamp, end_timestamp, half_time_iso = self.halve_time_range(start_date, end_date)
                     first_half_rows = self.recursive_get_rows_from_item(
                         item, data_type, start_date=start_timestamp, end_date=half_time_iso,
-                        interval=interval, sync_time=sync_time, boundary_type=boundary_type, can_raise=True, object_id=object_id,
+                        interval=interval, sync_time=sync_time, boundary_type=boundary_type,
+                        record_boundary_type=record_boundary_type, can_raise=True, object_id=object_id,
                         search_full_hierarchy=search_full_hierarchy, max_count=max_count, summary_type=summary_type, summary_duration=summary_duration
                     )
                     for row in first_half_rows:
                         yield row
                     logger.info("Successfully retrieved first half ({} to {})".format(start_timestamp, half_time_iso))
                     second_half_rows = self.recursive_get_rows_from_item(
                         item, data_type, start_date=half_time_iso, end_date=end_timestamp,
-                        interval=interval, sync_time=sync_time, boundary_type=boundary_type, can_raise=True, object_id=object_id,
+                        interval=interval, sync_time=sync_time, boundary_type=boundary_type,
+                        record_boundary_type=record_boundary_type, can_raise=True, object_id=object_id,
                         search_full_hierarchy=search_full_hierarchy, max_count=max_count, summary_type=summary_type, summary_duration=summary_duration
                     )
                     for row in second_half_rows:
@@ -219,7 +225,7 @@ def parse_pi_time(self, pi_time, to_epoch=False):
         return iso_timestamp
 
     def get_rows_from_webid(self, webid, data_type, start_date=None, end_date=None,
-                            interval=None, sync_time=None, boundary_type=None, selected_fields=None,
+                            interval=None, sync_time=None, boundary_type=None, record_boundary_type=None, selected_fields=None,
                             can_raise=True, endpoint_type="event_frames", search_full_hierarchy=None,
                             max_count=None, summary_type=None, summary_duration=None):
 
@@ -234,6 +240,7 @@ def get_rows_from_webid(self, webid, data_type, start_date=None, end_date=None,
                 interval=interval,
                 sync_time=sync_time,
                 boundary_type=boundary_type,
+                record_boundary_type=record_boundary_type,
                 selected_fields=selected_fields,
                 search_full_hierarchy=search_full_hierarchy,
                 max_count=max_count,
@@ -252,7 +259,7 @@ def get_rows_from_webid(self, webid, data_type, start_date=None, end_date=None,
                     yield item
 
     def get_rows_from_webids(self, input_rows, data_type, start_date=None, end_date=None,
-                             interval=None, sync_time=None, boundary_type=None, selected_fields=None, search_full_hierarchy=None,
+                             interval=None, sync_time=None, boundary_type=None, record_boundary_type=None, selected_fields=None, search_full_hierarchy=None,
                              max_count=None, can_raise=True, endpoint_type="event_frames", batch_size=500, summary_type=None, summary_duration=None):
         batch_requests_parameters = []
         number_processed_webids = 0
@@ -318,7 +325,7 @@ def _batch_requests(self, batch_requests_parameters):
             yield batch_section.get("Content", {})
 
     def generic_get_kwargs(self, start_date=None, end_date=None, interval=None, sync_time=None,
-                           boundary_type=None, selected_fields=None, search_full_hierarchy=None, max_count=None,
+                           boundary_type=None, record_boundary_type=None, selected_fields=None, search_full_hierarchy=None, max_count=None,
                            summary_type=None, summary_duration=None, can_raise=None):
         headers = self.get_requests_headers()
         params = self.get_requests_params(
@@ -327,6 +334,7 @@ def generic_get_kwargs(self, start_date=None, end_date=None, interval=None, sync
             interval=interval,
             sync_time=sync_time,
             boundary_type=boundary_type,
+            record_boundary_type=record_boundary_type,
             selected_fields=selected_fields,
             search_full_hierarchy=search_full_hierarchy,
             max_count=max_count,
@@ -339,7 +347,7 @@ def generic_get_kwargs(self, start_date=None, end_date=None, interval=None, sync
         }
 
     def generic_get(self, url, start_date=None, end_date=None, interval=None, sync_time=None,
-                    boundary_type=None, selected_fields=None, search_full_hierarchy=None, max_count=None,
+                    boundary_type=None, record_boundary_type=None, selected_fields=None, search_full_hierarchy=None, max_count=None,
                     can_raise=None, summary_type=None, summary_duration=None):
         headers = self.get_requests_headers()
         params = self.get_requests_params(
@@ -348,6 +356,7 @@ def generic_get(self, url, start_date=None, end_date=None, interval=None, sync_t
             interval=interval,
             sync_time=sync_time,
             boundary_type=boundary_type,
+            record_boundary_type=record_boundary_type,
             selected_fields=selected_fields,
             search_full_hierarchy=search_full_hierarchy,
             max_count=max_count,
@@ -363,7 +372,7 @@ def generic_get(self, url, start_date=None, end_date=None, interval=None, sync_t
         return json_response
 
     def get_rows_from_item(self, item, data_type, start_date=None, end_date=None, interval=None,
-                           sync_time=None, boundary_type=None, can_raise=True, object_id=None,
+                           sync_time=None, boundary_type=None, record_boundary_type=None, can_raise=True, object_id=None,
                            search_full_hierarchy=None, max_count=None, summary_type=None,
                            summary_duration=None):
         # item can be an pi tag, a path to an element or event frame
@@ -378,6 +387,7 @@ def get_rows_from_item(self, item, data_type, start_date=None, end_date=None, in
                 interval=interval,
                 sync_time=sync_time,
                 boundary_type=boundary_type,
+                record_boundary_type=record_boundary_type,
                 max_count=max_count,
                 search_full_hierarchy=search_full_hierarchy,
                 can_raise=can_raise,
@@ -392,7 +402,7 @@ def get_rows_from_item(self, item, data_type, start_date=None, end_date=None, in
                 yield self.loop_sub_items(item)
 
     def get_link_from_item(self, item, data_type, start_date, end_date, interval=None,
-                           sync_time=None, boundary_type=None, search_full_hierarchy=None,
+                           sync_time=None, boundary_type=None, record_boundary_type=None, search_full_hierarchy=None,
                            max_count=None, can_raise=True, summary_type=None,
                            summary_duration=None):
         url = self.extract_link_with_key(item, data_type)
@@ -404,7 +414,8 @@ def get_link_from_item(self, item, data_type, start_date, end_date, interval=Non
         headers = self.get_requests_headers()
         params = build_requests_params(
             start_time=start_date, end_time=end_date, interval=interval,
-            sync_time=sync_time, sync_time_boundary_type=boundary_type, search_full_hierarchy=search_full_hierarchy,
+            sync_time=sync_time, sync_time_boundary_type=boundary_type, record_boundary_type=record_boundary_type,
+            search_full_hierarchy=search_full_hierarchy,
             max_count=max_count, summary_type=summary_type, summary_duration=summary_duration
         )
         json_response = self.get(
@@ -625,7 +636,7 @@ def get_requests_headers(self):
         }
 
     def get_requests_params(self, start_date=None, end_date=None, interval=None, sync_time=None,
-                            boundary_type=None, selected_fields=None, search_full_hierarchy=None,
+                            boundary_type=None, record_boundary_type=None, selected_fields=None, search_full_hierarchy=None,
                             max_count=None, summary_type=None, summary_duration=None):
         params = {}
         if start_date:
@@ -638,6 +649,8 @@ def get_requests_params(self, start_date=None, end_date=None, interval=None, syn
             params.update({"syncTime": sync_time})
         if boundary_type:
             params.update({"syncTimeBoundaryType": boundary_type})
+        if record_boundary_type:
+            params.update({"boundaryType": record_boundary_type})
         if selected_fields:
             params.update({"selectedFields": selected_fields})
         if search_full_hierarchy:

python-lib/osisoft_constants.py

@@ -403,7 +403,7 @@ class OSIsoftConstants(object):
         "Security": "{base_url}/eventframes/{webid}/security",
         "SecurityEntries": "{base_url}/eventframes/{webid}/securityentries"
     }
-    PLUGIN_VERSION = "1.2.3"
+    PLUGIN_VERSION = "1.2.4-beta.1"
     VALUE_COLUMN_SUFFIX = "_val"
     WEB_API_PATH = "piwebapi"
     WRITE_HEADERS = {'X-Requested-With': 'XmlHttpRequest'}