From 3bd8f7349055e0e2af1b108d24251aad809a4a93 Mon Sep 17 00:00:00 2001 From: eirikhelsethequinor Date: Tue, 13 Jun 2023 15:06:24 +0200 Subject: [PATCH 1/8] WIP, added multiread_tag to PIWeb, cache need work --- tagreader/clients.py | 213 ++++++++++++++++++++++++++++++++ tagreader/web_handlers.py | 253 +++++++++++++++++++++++++++++++++++--- 2 files changed, 451 insertions(+), 15 deletions(-) diff --git a/tagreader/clients.py b/tagreader/clients.py index cb8f8903..f93a9d72 100644 --- a/tagreader/clients.py +++ b/tagreader/clients.py @@ -459,6 +459,161 @@ def _read_single_tag( df = df.rename(columns={"value": tag}) return df + + def _multi_read_tags(self, tag_list, start_time, stop_time, ts, read_type, get_status, cache=None): + + if read_type in [ReaderType.SNAPSHOT, ReaderType.RAW]: + + missing_data = {} + for tag in tag_list: + missing_data[tag] = {} + missing_data[tag]['0'] = {'start_time': start_time, 'stop_time': stop_time} + + df = self.handler.read_multi_tag( + tag_list=missing_data, + sample_time=ts, + read_type=read_type, + get_status=get_status + ) + + elif read_type in [ReaderType.INT]: + + missing_data = {} + + df = pd.DataFrame(columns=tag_list, index=pd.date_range(start_time.tz_convert("UTC"), + stop_time.tz_convert("UTC"), freq=pd.Timedelta(ts))) + + if isinstance(cache, SmartCache): + + for tag in tag_list: + + df_tag = cache.fetch( + tagname=tag, + readtype=read_type, + ts=ts, + start_time=start_time, + stop_time=stop_time + ) + + missing_intervals = get_missing_intervals( + df_tag, start_time, stop_time, ts, read_type) + + if missing_intervals: + + missing_data[tag] = {} + + for interval in range(len(missing_intervals)): + missing_data[tag][interval] = {'start_time': missing_intervals[interval][0], 'stop_time': missing_intervals[interval][1]} + + df = df.fillna(df_tag) + + if not missing_data: + return df.tz_convert(self.tz).sort_index() + + else: + for tag in tag_list: + missing_data[tag] = {} + missing_data[tag]['0'] = {'start_time': start_time, 'stop_time': stop_time} + + missing_df = self.handler.read_multi_tag( + tag_list=missing_data, + sample_time=ts, + read_type=read_type, + get_status=get_status + ) + + if read_type in [read_type.INT]: + df = df.fillna(missing_df) + else: + df = missing_df + + if len(missing_df.index) > 0: + if ( + cache is not None + and not get_status + ): + for column in missing_df: + + cache.store( + df=missing_df[column].to_frame(), + readtype=read_type, + ts=ts + ) + + df = df[~df.index.duplicated(keep="first")] # Deduplicate on index + + else: + + missing_data = {} + + df = pd.DataFrame(columns=tag_list, index=pd.date_range(stop_time.tz_convert("UTC"), + stop_time.tz_convert("UTC"), freq=pd.Timedelta(ts))) + + if isinstance(cache, SmartCache): + + for tag in tag_list: + + df_tag = cache.fetch( + tagname=tag, + readtype=read_type, + ts=ts, + start_time=start_time, + stop_time=stop_time + ) + + missing_intervals = get_missing_intervals( + df_tag, start_time, stop_time, ts, read_type) + + if missing_intervals: + + missing_data[tag] = {} + + for interval in range(len(missing_intervals)): + missing_data[tag][interval] = {'start_time': missing_intervals[interval][0], 'stop_time': missing_intervals[interval][1]} + + df = df.fillna(df_tag) + + if not missing_data: + return df.tz_convert(self.tz).sort_index() + + else: + for tag in tag_list: + missing_data[tag] = {} + missing_data[tag]['0'] = {'start_time': start_time, 'stop_time': stop_time} + + missing_df = self.handler.read_multi_tag( + tag_list=missing_data, + sample_time=ts, + read_type=read_type, + get_status=get_status + ) + + if read_type in [read_type.INT]: + df = df.fillna(missing_df) + else: + df = missing_df + + if len(missing_df.index) > 0: + if ( + cache is not None + and not get_status + ): + for column in missing_df: + + cache.store( + df=missing_df[column].to_frame(), + readtype=read_type, + ts=ts, + start_time=start_time, + stop_time=stop_time, + ) + + df = df[~df.index.duplicated(keep="first")] # Deduplicate on index + + df = df.tz_convert(self.tz).sort_index() + + return df + def get_units(self, tags: Union[str, List[str]]): if isinstance(tags, str): tags = [tags] @@ -517,6 +672,64 @@ def read_tags( get_status=get_status, ) + + def multi_read_tags( + self, + tags: list, + start_time=None, + end_time=None, + ts=60, + read_type=ReaderType.INT, + get_status=False, + ): + """Reads values for the specified [tags] from the IMS server for the + time interval from [start_time] to [stop_time] in intervals [ts]. + + The interval [ts] can be specified as pd.Timedelta or as an integer, + in which case it will be interpreted as seconds. + + Default value for [read_type] is ReaderType.INT, which interpolates + the raw data. + All possible values for read_type are defined in the ReaderType class, + which can be imported as follows: + from utils import ReaderType + + Values for ReaderType.* that should work for all handlers are: + INT, RAW, MIN, MAX, RNG, AVG, VAR, STD and SNAPSHOT + """ + + if read_type in [ReaderType.RAW] and len(tags) > 1: + raise RuntimeError( + "Unable to read raw/sampled data for multiple tags since they don't " + "share time vector. Read one at a time." + ) + + if read_type != ReaderType.SNAPSHOT: + start_time = ensure_datetime_with_tz(start_time, tz=self.tz) + if end_time: + end_time = ensure_datetime_with_tz(end_time, tz=self.tz) + if not isinstance(ts, pd.Timedelta): + ts = pd.Timedelta(ts, unit="s") + + if len(tags) != len(set(tags)): + duplicates = [x for n, x in enumerate(tags) if x in tags[:n]] + tag_list = [*set(tags)] + warnings.warn( + f"Duplicate tags found, removed duplicates: {', '.join(duplicates)}") + + res = self._multi_read_tags( + tag_list=tags, + start_time=start_time, + stop_time=end_time, + ts=ts, + read_type=read_type, + get_status=get_status, + cache=self.cache + ) + + return res + + def read( self, tags: Union[str, List[str]], diff --git a/tagreader/web_handlers.py b/tagreader/web_handlers.py index 2e74fbf8..0e2cc098 100644 --- a/tagreader/web_handlers.py +++ b/tagreader/web_handlers.py @@ -126,11 +126,19 @@ def __init__( urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) self.session.verify = verifySSL if verifySSL is not None else get_verifySSL() - def fetch(self, url, params=None) -> Response: + def fetch(self, url, params=None, json=None) -> Response: if not self.session.verify: urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) - res = self.session.get(url, params=params) + res = self.session.get(url, json=json, params=params) + res.raise_for_status() + return res + + def post(self, url, json=None, headers=None) -> Response: + if not self.session.verify: + urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) + + res = self.session.post(url, json=json, headers=headers) res.raise_for_status() return res @@ -676,7 +684,7 @@ def generate_search_query( def generate_read_query( self, - tag: str, + tag: None, start_time: pd.Timestamp, stop_time: pd.Timestamp, sample_time: Optional[Union[int, pd.Timestamp]], @@ -684,6 +692,7 @@ def generate_read_query( metadata: Optional[Dict[str, str]] = None, get_status: bool = False, ) -> Tuple[str, Dict[str, str]]: + if read_type in [ ReaderType.COUNT, ReaderType.GOOD, @@ -694,22 +703,26 @@ def generate_read_query( ]: raise NotImplementedError - webid = tag + if tag: + webid = tag - seconds = 0 - if read_type != ReaderType.SNAPSHOT: - seconds = int(sample_time.total_seconds()) + get_action = { + ReaderType.INT: "interpolated", + ReaderType.RAW: "recorded", + ReaderType.SNAPSHOT: "value", + ReaderType.SHAPEPRESERVING: "plot", + }.get(read_type, "summary") - get_action = { - ReaderType.INT: "interpolated", - ReaderType.RAW: "recorded", - ReaderType.SNAPSHOT: "value", - ReaderType.SHAPEPRESERVING: "plot", - }.get(read_type, "summary") + url = f"streams/{webid}/{get_action}" + else: + url = None - url = f"streams/{webid}/{get_action}" params = {} + seconds = 0 + if read_type != ReaderType.SNAPSHOT: + seconds = int(sample_time.total_seconds()) + if read_type != ReaderType.SNAPSHOT: params["startTime"] = self._time_to_UTC_string(start_time) params["endTime"] = self._time_to_UTC_string(stop_time) @@ -885,7 +898,7 @@ def read_tag( if not webid: return pd.DataFrame() - (url, params) = self.generate_read_query( + url, params = self.generate_read_query( tag=webid, start_time=start_time, stop_time=stop_time, @@ -969,5 +982,215 @@ def read_tag( return df.rename(columns={"Value": tag, "Status": tag + "::status"}) + + def read_multi_tag( + self, + tag_list: dict, + start_time: Optional[pd.Timestamp] = None, + stop_time: Optional[pd.Timestamp] = None, + sample_time: Optional[Union[int, pd.Timestamp]] = None, + read_type: ReaderType = ReaderType.INTERPOLATED, + metadata: Optional[Dict[str, str]] = None, + get_status: bool = False, + ): + + if len(tag_list) > 950: + raise RuntimeError( + f"ERROR: Trying to read too many tags ({len(tag_list)}). Server will return a HTTP 429 Too Many " + f"Requests error. Max. is set to 950. Please split the request." + ) + + urls = {} + + for tag in tag_list: + + for interval in tag_list[tag]: + + _, params = self.generate_read_query( + start_time=tag_list[tag][interval]['start_time'], + stop_time=tag_list[tag][interval]['stop_time'], + sample_time=sample_time, + read_type=read_type, + tag=None, + get_status=get_status + ) + + query_filter = self.create_filter(params) + + webid = self.tag_to_webid(tag) + + if not webid: + return pd.DataFrame() + + get_action = { + ReaderType.INT: "interpolated", + ReaderType.RAW: "recorded", + ReaderType.SNAPSHOT: "value", + ReaderType.SHAPEPRESERVING: "plot", + }.get(read_type, "summary") + + url = f"streams/{webid}/{get_action}" + + url = urljoin(self.base_url, f"{url}{query_filter}") + + urls[tag] = {"Method": "Get", "Resource": url} + + headers = {"Content-Type": "Application/json", "X-Requested-With": ""} + + res = self.post(url=self.base_url + "/batch", json=urls, headers=headers) + + j = res.json() + + df = pd.DataFrame() + + data = {} + + for tag in j: + timestamps = [] + + data[tag] = [] + if get_status: + data[f"{tag}::status"] = [] + + if read_type == ReaderType.SNAPSHOT: + + if type(j[tag]['Content']['Value']) == float: + data[tag].append(j[tag]['Content']['Value']) + else: + data[tag].append(np.nan) + + if get_status: + data[tag + '::status'].append(j[tag]['Content']["Questionable"] + 2 * (1 - j[tag]['Content']["Good"]) + + 4 * j[tag]['Content']["Substituted"]) + + timestamps.append(j[tag]['Content']['Timestamp']) + df_tag = pd.DataFrame(data, index=timestamps) + df = pd.concat([df, df_tag], axis=1) + data = {} + timestamps = [] + + elif read_type in [ReaderType.INT, ReaderType.INTERPOLATED, ReaderType.INTERPOLATE, ReaderType.RAW]: + for item in j[tag]['Content']['Items']: + + if type(item['Value']) == float: + data[tag].append(item['Value']) + else: + data[tag].append(np.nan) + + if get_status: + data[tag + '::status'].append(item["Questionable"] + 2 * (1 - item["Good"]) + + 4 * item["Substituted"]) + + timestamps.append(item['Timestamp']) + + df_tag = pd.DataFrame(data, index=timestamps) + df = pd.concat([df, df_tag], axis=1) + data = {} + timestamps = [] + + else: + for item in j[tag]['Content']['Items']: + + if type(item['Value']['Value']) == float: + data[tag].append(item['Value']['Value']) + else: + data[tag].append(np.nan) + + if get_status: + data[tag + '::status'].append(item['Value']["Questionable"] + 2 * (1 - item['Value']["Good"]) + + 4 * item['Value']["Substituted"]) + + timestamps.append(item['Value']['Timestamp']) + + df_tag = pd.DataFrame(data, index=timestamps) + df = pd.concat([df, df_tag], axis=1) + data = {} + timestamps = [] + + # Can happen for RAW reads w/o data in interval + if df.empty: + return df + + try: + if read_type == ReaderType.RAW or read_type == ReaderType.SNAPSHOT: + # Sub-second timestamps are common + df.index = pd.to_datetime(df.index, format="%Y-%m-%dT%H:%M:%S.%fZ", utc=True) + else: + # Sub-second timestamps are uncommon + df.index = pd.to_datetime(df.index, format="%Y-%m-%dT%H:%M:%SZ", utc=True) + + except ValueError: + df.index = pd.to_datetime(df.index, utc=True) + + if read_type == ReaderType.VAR: + df = df ** 2 + + df.index.name = "time" + + # Correct weird bug in PI Web API where MAX timestamps end of interval while + # all the other summaries stamp start of interval by shifting all timestamps + # one interval down. + if read_type == ReaderType.MAX: + import pytz + min_start_time = datetime.datetime.now().astimezone(pytz.utc) + for tag in tag_list: + for interval in tag_list[tag]: + if tag_list[tag][interval]['start_time'] < min_start_time: + min_start_time = tag_list[tag][interval]['start_time'] + if df.index[0] > min_start_time: + df.index = df.index - sample_time + + return df + def query_sql(self, query: str, parse: bool = True) -> pd.DataFrame: raise NotImplementedError + + def create_filter(self, params): + """ + Creates URL filter from parameters. Returns a string that will be appended to the request URL. + + :param params: + :return: + """ + filter_list = [] + if "interval" in params.keys(): + filter_list.append(f"interval={params['interval']}") + + if "summaryType" in params.keys(): + filter_list.append(f"summaryType={params['summaryType']}") + + if "startTime" in params.keys(): + url_start_time_ = self.date_to_urldate(params["startTime"]) + filter_list.append(f"starttime={url_start_time_}") + + if "endTime" in params.keys(): + url_stop_time = self.date_to_urldate(params["endTime"]) + filter_list.append(f"endtime={url_stop_time}") + + n = 0 + if filter_list: + for filters in filter_list: + if n == 0: + filter = f"?{filters}" + else: + filter = f"{filter}&{filters}" + n += 1 + else: + filter = "" + + return filter + + def date_to_urldate(self, date): + """ + Converts timestamp to url date format. + + :param date: Date and time in timestamp format + :return: + """ + + d = date.split("-")[0] + m = datetime.datetime.strptime(date.split("-")[1], '%b').month + y = date.split("-")[2][:2] + t = date.split(" ")[1] + + return f"20{y}-{m}-{d}T{t}Z" \ No newline at end of file From 9b809a4a6e770413e1396c1bd217e5590f3f945b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=85smund=20V=C3=A5ge=20Fannemel?= <34712686+asmfstatoil@users.noreply.github.com> Date: Thu, 21 Aug 2025 19:47:44 +0200 Subject: [PATCH 2/8] style: format with black --- tagreader/clients.py | 18 ++++-------------- tagreader/utils.py | 11 +++++++---- 2 files changed, 11 insertions(+), 18 deletions(-) diff --git a/tagreader/clients.py b/tagreader/clients.py index 5cfc0b73..53ab8d18 100644 --- a/tagreader/clients.py +++ b/tagreader/clients.py @@ -364,7 +364,6 @@ def _multi_read_tags( ): if read_type in [ReaderType.SNAPSHOT, ReaderType.RAW]: - missing_data = {} for tag in tag_list: missing_data[tag] = {} @@ -381,7 +380,6 @@ def _multi_read_tags( ) elif read_type in [ReaderType.INT]: - missing_data = {} df = pd.DataFrame( @@ -394,15 +392,14 @@ def _multi_read_tags( ) if isinstance(cache, SmartCache): - for tag in tag_list: - df_tag = cache.fetch( tagname=tag, - readtype=read_type, + read_type=read_type, ts=ts, - start_time=start_time, - stop_time=stop_time, + start=start_time, + end=stop_time, + get_status=get_status, ) missing_intervals = get_missing_intervals( @@ -410,7 +407,6 @@ def _multi_read_tags( ) if missing_intervals: - missing_data[tag] = {} for interval in range(len(missing_intervals)): @@ -447,7 +443,6 @@ def _multi_read_tags( if len(missing_df.index) > 0: if cache is not None and not get_status: for column in missing_df: - cache.store( df=missing_df[column].to_frame(), readtype=read_type, ts=ts ) @@ -468,9 +463,7 @@ def _multi_read_tags( ) if isinstance(cache, SmartCache): - for tag in tag_list: - df_tag = cache.fetch( tagname=tag, readtype=read_type, @@ -484,9 +477,7 @@ def _multi_read_tags( ) if missing_intervals: - missing_data[tag] = {} - for interval in range(len(missing_intervals)): missing_data[tag][interval] = { "start_time": missing_intervals[interval][0], @@ -521,7 +512,6 @@ def _multi_read_tags( if len(missing_df.index) > 0: if cache is not None and not get_status: for column in missing_df: - cache.store( df=missing_df[column].to_frame(), readtype=read_type, diff --git a/tagreader/utils.py b/tagreader/utils.py index 6c9b0a6f..746c2a03 100644 --- a/tagreader/utils.py +++ b/tagreader/utils.py @@ -55,10 +55,13 @@ def ensure_datetime_with_tz( ) -> datetime: date_stamp = convert_to_pydatetime(date_stamp) - if not date_stamp.tzinfo: - date_stamp = tz.localize(date_stamp) - - return date_stamp + if date_stamp.tzinfo: + return date_stamp + else: + if hasattr(tz, "localize"): + return tz.localize(date_stamp) + else: + return date_stamp.replace(tzinfo=tz) def urljoin(*args) -> str: From 6f2a3cdc020883de3c72bc0f507eb67833838a67 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=85smund=20V=C3=A5ge=20Fannemel?= <34712686+asmfstatoil@users.noreply.github.com> Date: Tue, 6 Oct 2026 17:18:38 +0200 Subject: [PATCH 3/8] refactor: streamline parameter handling in AspenHandlerWeb for improved readability --- tagreader/web_handlers.py | 26 +++++++++++--------------- 1 file changed, 11 insertions(+), 15 deletions(-) diff --git a/tagreader/web_handlers.py b/tagreader/web_handlers.py index 5ae6cbd7..265eaa65 100644 --- a/tagreader/web_handlers.py +++ b/tagreader/web_handlers.py @@ -692,21 +692,18 @@ def read_tag( if map_name is None: map_name = "" - if tag_name is not None: - params = self.generate_read_query( - tagname=tag_name, - mapname=map_name, - start=start, - end=end, - sample_time=sample_time, - read_type=read_type, - metadata={}, - ) + params = self.generate_read_query( + tagname=tag_name, + mapname=map_name, + start=start, + end=end, + sample_time=sample_time, + read_type=read_type, + metadata={}, + ) - # Consider use_post = use_post or not tag.isascii() - data = self.fetch(url, params=params, use_post=use_post) - else: - data = {} + # Consider use_post = use_post or not tag.isascii() + data = self.fetch(url, params=params, use_post=use_post) if len(data) == 0: # Normally for timestamps in future return pd.DataFrame(columns=[tag]) @@ -899,7 +896,6 @@ def generate_read_query( metadata: Optional[Dict[str, str]], get_status: bool = False, ) -> Tuple[str, Dict[str, str]]: - if read_type in [ ReaderType.COUNT, ReaderType.GOOD, From 42dc26b709d6c6737b697fc6239ceb42679f8b66 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=85smund=20V=C3=A5ge=20Fannemel?= <34712686+asmfstatoil@users.noreply.github.com> Date: Tue, 6 Oct 2026 17:21:40 +0200 Subject: [PATCH 4/8] feat: tag not optional --- tagreader/web_handlers.py | 24 ++++++++++++------------ 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/tagreader/web_handlers.py b/tagreader/web_handlers.py index 265eaa65..a429b359 100644 --- a/tagreader/web_handlers.py +++ b/tagreader/web_handlers.py @@ -888,7 +888,7 @@ def generate_search_params( def generate_read_query( self, - tag: Optional[str], + tag: str, start: pd.Timestamp, end: pd.Timestamp, sample_time: Optional[Union[int, pd.Timestamp]], @@ -906,19 +906,19 @@ def generate_read_query( ]: raise NotImplementedError - if tag: - web_id = tag + if not tag: + raise ValueError("Tag must be provided.") - get_action = { - ReaderType.INT: "interpolated", - ReaderType.RAW: "recorded", - ReaderType.SNAPSHOT: "value", - ReaderType.SHAPEPRESERVING: "plot", - }.get(read_type, "summary") + web_id = tag - url = f"streams/{web_id}/{get_action}" - else: - url = "" + get_action = { + ReaderType.INT: "interpolated", + ReaderType.RAW: "recorded", + ReaderType.SNAPSHOT: "value", + ReaderType.SHAPEPRESERVING: "plot", + }.get(read_type, "summary") + + url = f"streams/{web_id}/{get_action}" params = {} From b8e3a2f07b67d56663100eccd02f805b250e6ef9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=85smund=20V=C3=A5ge=20Fannemel?= <34712686+asmfstatoil@users.noreply.github.com> Date: Tue, 6 Oct 2026 17:23:59 +0200 Subject: [PATCH 5/8] revert --- tagreader/web_handlers.py | 15 +++++++-------- 1 file changed, 7 insertions(+), 8 deletions(-) diff --git a/tagreader/web_handlers.py b/tagreader/web_handlers.py index a429b359..a32d5acb 100644 --- a/tagreader/web_handlers.py +++ b/tagreader/web_handlers.py @@ -889,9 +889,9 @@ def generate_search_params( def generate_read_query( self, tag: str, - start: pd.Timestamp, - end: pd.Timestamp, - sample_time: Optional[Union[int, pd.Timestamp]], + start: datetime, + end: datetime, + sample_time: timedelta, read_type: ReaderType, metadata: Optional[Dict[str, str]], get_status: bool = False, @@ -911,21 +911,20 @@ def generate_read_query( web_id = tag + seconds = 0 + if read_type != ReaderType.SNAPSHOT: + seconds = int(sample_time.total_seconds()) + get_action = { ReaderType.INT: "interpolated", ReaderType.RAW: "recorded", ReaderType.SNAPSHOT: "value", ReaderType.SHAPEPRESERVING: "plot", }.get(read_type, "summary") - url = f"streams/{web_id}/{get_action}" params = {} - seconds = 0 - if read_type != ReaderType.SNAPSHOT: - seconds = int(sample_time.total_seconds()) - if read_type != ReaderType.SNAPSHOT: params["startTime"] = self._time_to_UTC_string(start) params["endTime"] = self._time_to_UTC_string(end) From e66bf06f82cd31c73b230e92533285deac7c6db3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=85smund=20V=C3=A5ge=20Fannemel?= <34712686+asmfstatoil@users.noreply.github.com> Date: Tue, 6 Oct 2026 17:24:45 +0200 Subject: [PATCH 6/8] f --- tagreader/web_handlers.py | 1 - 1 file changed, 1 deletion(-) diff --git a/tagreader/web_handlers.py b/tagreader/web_handlers.py index a32d5acb..20821ffd 100644 --- a/tagreader/web_handlers.py +++ b/tagreader/web_handlers.py @@ -922,7 +922,6 @@ def generate_read_query( ReaderType.SHAPEPRESERVING: "plot", }.get(read_type, "summary") url = f"streams/{web_id}/{get_action}" - params = {} if read_type != ReaderType.SNAPSHOT: From 944f140ba97b48ed581f5a1da3e2bcdf24a69f28 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=85smund=20V=C3=A5ge=20Fannemel?= <34712686+asmfstatoil@users.noreply.github.com> Date: Tue, 6 Oct 2026 18:02:04 +0200 Subject: [PATCH 7/8] feat: implement multi-read functionality for tags in IMSClient and add corresponding tests --- tagreader/clients.py | 9 +++--- tagreader/web_handlers.py | 22 +++----------- tests/test_PIHandlerREST_connect.py | 28 +++++++++++++++++ tests/test_clients.py | 47 +++++++++++++++++++++++++++++ 4 files changed, 84 insertions(+), 22 deletions(-) diff --git a/tagreader/clients.py b/tagreader/clients.py index 9425276a..2aa9f8cd 100644 --- a/tagreader/clients.py +++ b/tagreader/clients.py @@ -405,8 +405,8 @@ def _multi_read_tags( df = pd.DataFrame( columns=tag_list, index=pd.date_range( - start_time.tz_convert("UTC"), - stop_time.tz_convert("UTC"), + start_time.astimezone(timezone.utc), + stop_time.astimezone(timezone.utc), freq=pd.Timedelta(ts), ), ) @@ -470,14 +470,13 @@ def _multi_read_tags( df = df[~df.index.duplicated(keep="first")] # Deduplicate on index else: - missing_data = {} df = pd.DataFrame( columns=tag_list, index=pd.date_range( - stop_time.tz_convert("UTC"), - stop_time.tz_convert("UTC"), + start_time.astimezone(timezone.utc), + stop_time.astimezone(timezone.utc), freq=pd.Timedelta(ts), ), ) diff --git a/tagreader/web_handlers.py b/tagreader/web_handlers.py index 20821ffd..fca8ba70 100644 --- a/tagreader/web_handlers.py +++ b/tagreader/web_handlers.py @@ -256,7 +256,6 @@ def fetch( self, url, params: Optional[Union[bytes, str, Dict[str, str]]] = None, - json_str: Optional[str] = None, timeout: Optional[int] = None, use_post: bool = False, ) -> Dict: @@ -283,7 +282,6 @@ def fetch( res = self.session.get( url, params=params, - json=json_str, timeout=(None, timeout), ) # Noqa. Read timeout, No connect timeout. @@ -307,11 +305,11 @@ def fetch( txt = res.text.replace('"v":nan', '"v":NaN').replace('"v":-nan', '"v":NaN') return json.loads(txt) - def post(self, url, json=None, headers=None) -> requests.Response: + def post(self, url, json_str=None, headers=None) -> requests.Response: if not self.session.verify: urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) - res = self.session.post(url, json=json, headers=headers) + res = self.session.post(url, json=json_str, headers=headers) res.raise_for_status() return res @@ -1202,7 +1200,6 @@ def read_multi_tag( metadata: Optional[Dict[str, str]] = None, get_status: bool = False, ): - if len(tag_list) > 950: raise RuntimeError( f"ERROR: Trying to read too many tags ({len(tag_list)}). Server will return a HTTP 429 Too Many " @@ -1210,7 +1207,6 @@ def read_multi_tag( ) urls = {} - for tag in tag_list: for interval in tag_list[tag]: _, params = self.generate_read_query( @@ -1218,7 +1214,7 @@ def read_multi_tag( end=tag_list[tag][interval]["stop_time"], sample_time=sample_time, read_type=read_type, - tag=None, + tag=tag, metadata=metadata, get_status=get_status, ) @@ -1238,30 +1234,22 @@ def read_multi_tag( }.get(read_type, "summary") url = f"streams/{webid}/{get_action}" - url = urljoin(self.base_url, f"{url}{query_filter}") - urls[tag] = {"Method": "Get", "Resource": url} headers = {"Content-Type": "Application/json", "X-Requested-With": ""} - - res = self.post(url=self.base_url + "/batch", json=urls, headers=headers) - - j = res.json() + res = self.post(url=self.base_url + "/batch", json_str=urls, headers=headers) df = pd.DataFrame() - data = {} - + j = res.json() for tag in j: timestamps = [] - data[tag] = [] if get_status: data[f"{tag}::status"] = [] if read_type == ReaderType.SNAPSHOT: - if isinstance(j[tag]["Content"]["Value"], float): data[tag].append(j[tag]["Content"]["Value"]) else: diff --git a/tests/test_PIHandlerREST_connect.py b/tests/test_PIHandlerREST_connect.py index 0c433f2f..b9bae974 100644 --- a/tests/test_PIHandlerREST_connect.py +++ b/tests/test_PIHandlerREST_connect.py @@ -2,6 +2,7 @@ from datetime import timedelta from typing import Generator +import pandas as pd import pytest from tagreader.cache import SmartCache @@ -182,6 +183,33 @@ def test_read(client: IMSClient, read_type: str, size: int) -> None: assert df.index[-1] <= ensure_datetime_with_tz(STOP_TIME) +def test_multi_read_tags(client: IMSClient) -> None: + tags = [TAGS["Float32"], "SINUSOID"] + result = client.multi_read_tags( + tags=tags, + start_time=START_TIME, + end_time=STOP_TIME, + ts=SAMPLE_TIME, + read_type=ReaderType.INT, + ) + expected = client.read( + tags=tags, + start_time=START_TIME, + end_time=STOP_TIME, + ts=SAMPLE_TIME, + read_type=ReaderType.INT, + ) + + assert result.shape == (61, 2) + assert list(result.columns) == tags + assert result.notna().any().all() + assert result.index.is_unique + assert result.index.is_monotonic_increasing + pd.testing.assert_frame_equal( + result, expected, check_dtype=False, check_names=False, check_freq=False + ) + + def test_read_with_status(client: IMSClient) -> None: df = client.read( tags=TAGS["Float32"], diff --git a/tests/test_clients.py b/tests/test_clients.py index 17c38c27..e6846a16 100644 --- a/tests/test_clients.py +++ b/tests/test_clients.py @@ -3,7 +3,9 @@ import pandas as pd import pytest +from requests.auth import AuthBase +from tagreader.cache import SmartCache from tagreader.clients import IMSClient, get_missing_intervals, get_next_timeslice from tagreader.utils import IMSType, ReaderType @@ -13,6 +15,51 @@ def test_init_client_without_cache() -> None: assert not client.cache +def test_multi_read_tags_smartcache_hits(cache: SmartCache) -> None: + tags = ["tag1", "tag2"] + sample_time = timedelta(seconds=60) + cached_data = pd.DataFrame( + {"tag1": [1.0, 2.0, 3.0, 4.0], "tag2": [5.0, 6.0, 7.0, 8.0]}, + index=pd.date_range("2020-04-01 09:05:00", periods=4, freq="60s", tz="UTC"), + ) + for tag in tags: + cache.store( + df=cached_data[[tag]], + tagname=tag, + read_type=ReaderType.INT, + ts=sample_time, + get_status=False, + ) + cache[tag] = f"cached-webid-{tag}" + + client = IMSClient( + datasource="cache-test", + imstype=IMSType.PIWEBAPI, + url="cache-only://pi", + auth=AuthBase(), + verify_ssl=True, + cache=cache, + tz="Europe/Oslo", + ) + cache.enable_cache_statistics() + cache.stats(reset=True) + expected = cached_data.tz_convert(client.tz) + + try: + for read_number in (1, 2): + result = client.multi_read_tags( + tags=tags, + start_time=expected.index[0], + end_time=expected.index[-1], + ts=60, + read_type=ReaderType.INT, + ) + pd.testing.assert_frame_equal(result, expected, check_dtype=False) + assert cache.stats() == (len(tags) * read_number, 0) + finally: + client.handler.session.close() + + def test_init_client_with_tzinfo() -> None: """ Currently testing valid timezone From a5ba20a386179123f352645962ee77b5fa934a56 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=85smund=20V=C3=A5ge=20Fannemel?= <34712686+asmfstatoil@users.noreply.github.com> Date: Tue, 6 Oct 2026 18:27:49 +0200 Subject: [PATCH 8/8] fix: improve cache --- tagreader/clients.py | 9 ++-- tagreader/web_handlers.py | 84 ++++------------------------- tests/test_PIHandlerREST_connect.py | 46 +++++++++++++++- 3 files changed, 60 insertions(+), 79 deletions(-) diff --git a/tagreader/clients.py b/tagreader/clients.py index 2aa9f8cd..888564e8 100644 --- a/tagreader/clients.py +++ b/tagreader/clients.py @@ -1,7 +1,7 @@ from datetime import datetime, timedelta, timezone, tzinfo from itertools import groupby from operator import itemgetter -from typing import Any, Dict, List, Optional, Tuple, Union +from typing import Any, Dict, List, Optional, Tuple, Union, cast from zoneinfo import ZoneInfo, ZoneInfoNotFoundError import numpy as np @@ -464,7 +464,11 @@ def _multi_read_tags( if cache is not None and not get_status: for column in missing_df: cache.store( - df=missing_df[column].to_frame(), readtype=read_type, ts=ts + df=missing_df[column].to_frame(), + tagname=cast(str, column), + read_type=read_type, + ts=ts, + get_status=get_status, ) df = df[~df.index.duplicated(keep="first")] # Deduplicate on index @@ -507,7 +511,6 @@ def _multi_read_tags( if not missing_data: return df.tz_convert(self.tz).sort_index() - else: for tag in tag_list: missing_data[tag] = {} diff --git a/tagreader/web_handlers.py b/tagreader/web_handlers.py index 3fa8c2a5..6a120b31 100644 --- a/tagreader/web_handlers.py +++ b/tagreader/web_handlers.py @@ -1206,33 +1206,23 @@ def read_multi_tag( urls = {} for tag in tag_list: + web_id = self.tag_to_web_id(tag) + if not web_id: + return pd.DataFrame() + for interval in tag_list[tag]: - _, params = self.generate_read_query( + url, params = self.generate_read_query( + tag=web_id, start=tag_list[tag][interval]["start_time"], end=tag_list[tag][interval]["stop_time"], sample_time=sample_time, read_type=read_type, - tag=tag, metadata=metadata, get_status=get_status, ) - query_filter = self.create_filter(params) - - webid = self.tag_to_web_id(tag) - - if not webid: - return pd.DataFrame() - - get_action = { - ReaderType.INT: "interpolated", - ReaderType.RAW: "recorded", - ReaderType.SNAPSHOT: "value", - ReaderType.SHAPEPRESERVING: "plot", - }.get(read_type, "summary") - - url = f"streams/{webid}/{get_action}" - url = urljoin(self.base_url, f"{url}{query_filter}") + query = urllib.parse.urlencode(params) + url = f"{urljoin(self.base_url, url)}?{query}" urls[tag] = {"Method": "Get", "Resource": url} headers = {"Content-Type": "Application/json", "X-Requested-With": ""} @@ -1265,7 +1255,6 @@ def read_multi_tag( df = pd.concat([df, df_tag], axis=1) data = {} timestamps = [] - elif read_type in [ ReaderType.INT, ReaderType.INTERPOLATED, @@ -1273,7 +1262,6 @@ def read_multi_tag( ReaderType.RAW, ]: for item in j[tag]["Content"]["Items"]: - if isinstance(item["Value"], float): data[tag].append(item["Value"]) else: @@ -1292,7 +1280,6 @@ def read_multi_tag( df = pd.concat([df, df_tag], axis=1) data = {} timestamps = [] - else: for item in j[tag]["Content"]["Items"]: if isinstance(item["Value"]["Value"], float): @@ -1329,7 +1316,6 @@ def read_multi_tag( df.index = pd.to_datetime( df.index, format="%Y-%m-%dT%H:%M:%SZ", utc=True ) - except ValueError: df.index = pd.to_datetime(df.index, utc=True) @@ -1342,9 +1328,7 @@ def read_multi_tag( # all the other summaries stamp start of interval by shifting all timestamps # one interval down. if read_type == ReaderType.MAX: - import pytz - - min_start_time = datetime.now().astimezone(pytz.utc) + min_start_time = datetime.now(timezone.utc) for tag in tag_list: for interval in tag_list[tag]: if tag_list[tag][interval]["start_time"] < min_start_time: @@ -1356,53 +1340,3 @@ def read_multi_tag( def query_sql(self, query: str, parse: bool = True) -> pd.DataFrame: raise NotImplementedError - - def create_filter(self, params): - """ - Creates URL filter from parameters. Returns a string that will be appended to the request URL. - - :param params: - :return: - """ - filter_list = [] - if "interval" in params.keys(): - filter_list.append(f"interval={params['interval']}") - - if "summaryType" in params.keys(): - filter_list.append(f"summaryType={params['summaryType']}") - - if "startTime" in params.keys(): - url_start_time_ = self.date_to_urldate(params["startTime"]) - filter_list.append(f"starttime={url_start_time_}") - - if "endTime" in params.keys(): - url_stop_time = self.date_to_urldate(params["endTime"]) - filter_list.append(f"endtime={url_stop_time}") - - n = 0 - if filter_list: - for filters in filter_list: - if n == 0: - filter = f"?{filters}" - else: - filter = f"{filter}&{filters}" - n += 1 - else: - filter = "" - - return filter - - def date_to_urldate(self, date): - """ - Converts timestamp to url date format. - - :param date: Date and time in timestamp format - :return: - """ - - d = date.split("-")[0] - m = datetime.strptime(date.split("-")[1], "%b").month - y = date.split("-")[2][:2] - t = date.split(" ")[1] - - return f"20{y}-{m}-{d}T{t}Z" diff --git a/tests/test_PIHandlerREST_connect.py b/tests/test_PIHandlerREST_connect.py index b9bae974..aaaecc51 100644 --- a/tests/test_PIHandlerREST_connect.py +++ b/tests/test_PIHandlerREST_connect.py @@ -1,9 +1,10 @@ import os from datetime import timedelta -from typing import Generator +from typing import Any, Generator import pandas as pd import pytest +from requests import Response from tagreader.cache import SmartCache from tagreader.clients import IMSClient, list_sources @@ -210,6 +211,49 @@ def test_multi_read_tags(client: IMSClient) -> None: ) +def test_multi_read_tags_smartcache_miss(client: IMSClient, cache: SmartCache) -> None: + tags = [TAGS["Float32"], "SINUSOID"] + request_methods = [] + + def record_response(response: Response, **kwargs: Any) -> None: + request_methods.append(response.request.method) + + assert list(cache.iterkeys()) == [] + client.cache = cache + cache.enable_cache_statistics() + cache.stats(reset=True) + client.handler.session.hooks["response"].append(record_response) + + try: + first = client.multi_read_tags( + tags=tags, + start_time=START_TIME, + end_time=STOP_TIME, + ts=SAMPLE_TIME, + read_type=ReaderType.INT, + ) + assert first.shape == (61, 2) + assert first.notna().any().all() + assert "POST" in request_methods + assert cache.stats() == (0, len(tags)) + assert len(list(cache.iterkeys())) == len(tags) + requests_after_first = len(request_methods) + + second = client.multi_read_tags( + tags=tags, + start_time=START_TIME, + end_time=STOP_TIME, + ts=SAMPLE_TIME, + read_type=ReaderType.INT, + ) + pd.testing.assert_frame_equal(second, first) + assert cache.stats() == (len(tags), len(tags)) + assert len(request_methods) == requests_after_first + finally: + client.handler.session.hooks["response"].remove(record_response) + client.cache = None + + def test_read_with_status(client: IMSClient) -> None: df = client.read( tags=TAGS["Float32"],