diff --git a/tagreader/clients.py b/tagreader/clients.py index d8e7ff7..888564e 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 @@ -379,6 +379,175 @@ 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.astimezone(timezone.utc), + stop_time.astimezone(timezone.utc), + freq=pd.Timedelta(ts), + ), + ) + + if isinstance(cache, SmartCache): + for tag in tag_list: + df_tag = cache.fetch( + tagname=tag, + read_type=read_type, + ts=ts, + start=start_time, + end=stop_time, + get_status=get_status, + ) + + 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(), + tagname=cast(str, column), + read_type=read_type, + ts=ts, + get_status=get_status, + ) + + df = df[~df.index.duplicated(keep="first")] # Deduplicate on index + + else: + missing_data = {} + + df = pd.DataFrame( + columns=tag_list, + index=pd.date_range( + start_time.astimezone(timezone.utc), + stop_time.astimezone(timezone.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] @@ -447,6 +616,63 @@ 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)] + logger.warning( + 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 81a09e7..6a120b3 100644 --- a/tagreader/web_handlers.py +++ b/tagreader/web_handlers.py @@ -282,10 +282,7 @@ def fetch( res = self.session.get( url, params=params, - timeout=( - None, - timeout, - ), + timeout=(None, timeout), ) # Noqa. Read timeout, No connect timeout. res.raise_for_status() @@ -308,6 +305,14 @@ def fetch( txt = res.text.replace('"v":nan', '"v":NaN').replace('"v":-nan', '"v":NaN') return json.loads(txt) + 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_str, headers=headers) + res.raise_for_status() + return res + def connect(self): if self.datasource == "": raise ValueError("Datasource must be specified") @@ -1183,5 +1188,155 @@ 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: + web_id = self.tag_to_web_id(tag) + if not web_id: + return pd.DataFrame() + + for interval in tag_list[tag]: + 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, + metadata=metadata, + get_status=get_status, + ) + + 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": ""} + 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: + 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 isinstance(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 isinstance(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: + 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: + 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 diff --git a/tests/test_PIHandlerREST_connect.py b/tests/test_PIHandlerREST_connect.py index 0c433f2..aaaecc5 100644 --- a/tests/test_PIHandlerREST_connect.py +++ b/tests/test_PIHandlerREST_connect.py @@ -1,8 +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 @@ -182,6 +184,76 @@ 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_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"], diff --git a/tests/test_clients.py b/tests/test_clients.py index 17c38c2..e6846a1 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