resources/tools/dash/app/pal/data/data.py

   1 # Copyright (c) 2022 Cisco and/or its affiliates.
   2 # Licensed under the Apache License, Version 2.0 (the "License");
   3 # you may not use this file except in compliance with the License.
   4 # You may obtain a copy of the License at:
   5 #
   6 #     http://www.apache.org/licenses/LICENSE-2.0
   7 #
   8 # Unless required by applicable law or agreed to in writing, software
   9 # distributed under the License is distributed on an "AS IS" BASIS,
  10 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  11 # See the License for the specific language governing permissions and
  12 # limitations under the License.
  13
  14 """Prepare data for Plotly Dash applications.
  15 """
  16
  17 import logging
  18 import awswrangler as wr
  19
  20 from yaml import load, FullLoader, YAMLError
  21 from datetime import datetime, timedelta
  22 from time import time
  23 from pytz import UTC
  24 from pandas import DataFrame
  25 from awswrangler.exceptions import EmptyDataFrame, NoFilesFound
  26
  27
  28 class Data:
  29     """Gets the data from parquets and stores it for further use by dash
  30     applications.
  31     """
  32
  33     def __init__(self, data_spec_file: str, debug: bool=False) -> None:
  34         """Initialize the Data object.
  35
  36         :param data_spec_file: Path to file specifying the data to be read from
  37             parquets.
  38         :param debug: If True, the debuf information is printed to stdout.
  39         :type data_spec_file: str
  40         :type debug: bool
  41         :raises RuntimeError: if it is not possible to open data_spec_file or it
  42             is not a valid yaml file.
  43         """
  44
  45         # Inputs:
  46         self._data_spec_file = data_spec_file
  47         self._debug = debug
  48
  49         # Specification of data to be read from parquets:
  50         self._data_spec = None
  51
  52         # Data frame to keep the data:
  53         self._data = None
  54
  55         # Read from files:
  56         try:
  57             with open(self._data_spec_file, "r") as file_read:
  58                 self._data_spec = load(file_read, Loader=FullLoader)
  59         except IOError as err:
  60             raise RuntimeError(
  61                 f"Not possible to open the file {self._data_spec_file,}\n{err}"
  62             )
  63         except YAMLError as err:
  64             raise RuntimeError(
  65                 f"An error occurred while parsing the specification file "
  66                 f"{self._data_spec_file,}\n"
  67                 f"{err}"
  68             )
  69
  70     @property
  71     def data(self):
  72         return self._data
  73
  74     def _get_columns(self, parquet: str) -> list:
  75         """Get the list of columns from the data specification file to be read
  76         from parquets.
  77
  78         :param parquet: The parquet's name.
  79         :type parquet: str
  80         :raises RuntimeError: if the parquet is not defined in the data
  81             specification file or it does not have any columns specified.
  82         :returns: List of columns.
  83         :rtype: list
  84         """
  85
  86         try:
  87             return self._data_spec[parquet]["columns"]
  88         except KeyError as err:
  89             raise RuntimeError(
  90                 f"The parquet {parquet} is not defined in the specification "
  91                 f"file {self._data_spec_file} or it does not have any columns "
  92                 f"specified.\n{err}"
  93             )
  94
  95     def _get_path(self, parquet: str) -> str:
  96         """Get the path from the data specification file to be read from
  97         parquets.
  98
  99         :param parquet: The parquet's name.
 100         :type parquet: str
 101         :raises RuntimeError: if the parquet is not defined in the data
 102             specification file or it does not have the path specified.
 103         :returns: Path.
 104         :rtype: str
 105         """
 106
 107         try:
 108             return self._data_spec[parquet]["path"]
 109         except KeyError as err:
 110             raise RuntimeError(
 111                 f"The parquet {parquet} is not defined in the specification "
 112                 f"file {self._data_spec_file} or it does not have the path "
 113                 f"specified.\n{err}"
 114             )
 115
 116     def _create_dataframe_from_parquet(self,
 117         path, partition_filter=None,
 118         columns=None,
 119         validate_schema=False,
 120         last_modified_begin=None,
 121         last_modified_end=None,
 122         days=None) -> DataFrame:
 123         """Read parquet stored in S3 compatible storage and returns Pandas
 124         Dataframe.
 125
 126         :param path: S3 prefix (accepts Unix shell-style wildcards)
 127             (e.g. s3://bucket/prefix) or list of S3 objects paths
 128             (e.g. [s3://bucket/key0, s3://bucket/key1]).
 129         :param partition_filter: Callback Function filters to apply on PARTITION
 130             columns (PUSH-DOWN filter). This function MUST receive a single
 131             argument (Dict[str, str]) where keys are partitions names and values
 132             are partitions values. Partitions values will be always strings
 133             extracted from S3. This function MUST return a bool, True to read
 134             the partition or False to ignore it. Ignored if dataset=False.
 135         :param columns: Names of columns to read from the file(s).
 136         :param validate_schema: Check that individual file schemas are all the
 137             same / compatible. Schemas within a folder prefix should all be the
 138             same. Disable if you have schemas that are different and want to
 139             disable this check.
 140         :param last_modified_begin: Filter the s3 files by the Last modified
 141             date of the object. The filter is applied only after list all s3
 142             files.
 143         :param last_modified_end: Filter the s3 files by the Last modified date
 144             of the object. The filter is applied only after list all s3 files.
 145         :type path: Union[str, List[str]]
 146         :type partition_filter: Callable[[Dict[str, str]], bool], optional
 147         :type columns: List[str], optional
 148         :type validate_schema: bool, optional
 149         :type last_modified_begin: datetime, optional
 150         :type last_modified_end: datetime, optional
 151         :returns: Pandas DataFrame or None if DataFrame cannot be fetched.
 152         :rtype: DataFrame
 153         """
 154         df = None
 155         start = time()
 156         if days:
 157             last_modified_begin = datetime.now(tz=UTC) - timedelta(days=days)
 158         try:
 159             df = wr.s3.read_parquet(
 160                 path=path,
 161                 path_suffix="parquet",
 162                 ignore_empty=True,
 163                 validate_schema=validate_schema,
 164                 use_threads=True,
 165                 dataset=True,
 166                 columns=columns,
 167                 partition_filter=partition_filter,
 168                 last_modified_begin=last_modified_begin,
 169                 last_modified_end=last_modified_end
 170             )
 171             if self._debug:
 172                 df.info(verbose=True, memory_usage='deep')
 173                 logging.info(
 174                     u"\n"
 175                     f"Creation of dataframe {path} took: {time() - start}"
 176                     u"\n"
 177                 )
 178         except NoFilesFound as err:
 179             logging.error(f"No parquets found.\n{err}")
 180         except EmptyDataFrame as err:
 181             logging.error(f"No data.\n{err}")
 182
 183         self._data = df
 184         return df
 185
 186     def read_stats(self, days: int=None) -> tuple:
 187         """Read statistics from parquet.
 188
 189         It reads from:
 190         - Suite Result Analysis (SRA) partition,
 191         - NDRPDR trending partition,
 192         - MRR trending partition.
 193
 194         :param days: Number of days back to the past for which the data will be
 195             read.
 196         :type days: int
 197         :returns: tuple of pandas DataFrame-s with data read from specified
 198             parquets.
 199         :rtype: tuple of pandas DataFrame-s
 200         """
 201
 202         l_stats = lambda part: True if part["stats_type"] == "sra" else False
 203         l_mrr = lambda part: True if part["test_type"] == "mrr" else False
 204         l_ndrpdr = lambda part: True if part["test_type"] == "ndrpdr" else False
 205
 206         return (
 207             self._create_dataframe_from_parquet(
 208                 path=self._get_path("statistics"),
 209                 partition_filter=l_stats,
 210                 columns=self._get_columns("statistics"),
 211                 days=days
 212             ),
 213             self._create_dataframe_from_parquet(
 214                 path=self._get_path("statistics-trending-mrr"),
 215                 partition_filter=l_mrr,
 216                 columns=self._get_columns("statistics-trending-mrr"),
 217                 days=days
 218             ),
 219             self._create_dataframe_from_parquet(
 220                 path=self._get_path("statistics-trending-ndrpdr"),
 221                 partition_filter=l_ndrpdr,
 222                 columns=self._get_columns("statistics-trending-ndrpdr"),
 223                 days=days
 224             )
 225         )
 226
 227     def read_trending_mrr(self, days: int=None) -> DataFrame:
 228         """Read MRR data partition from parquet.
 229
 230         :param days: Number of days back to the past for which the data will be
 231             read.
 232         :type days: int
 233         :returns: Pandas DataFrame with read data.
 234         :rtype: DataFrame
 235         """
 236
 237         lambda_f = lambda part: True if part["test_type"] == "mrr" else False
 238
 239         return self._create_dataframe_from_parquet(
 240             path=self._get_path("trending-mrr"),
 241             partition_filter=lambda_f,
 242             columns=self._get_columns("trending-mrr"),
 243             days=days
 244         )
 245
 246     def read_trending_ndrpdr(self, days: int=None) -> DataFrame:
 247         """Read NDRPDR data partition from iterative parquet.
 248
 249         :param days: Number of days back to the past for which the data will be
 250             read.
 251         :type days: int
 252         :returns: Pandas DataFrame with read data.
 253         :rtype: DataFrame
 254         """
 255
 256         lambda_f = lambda part: True if part["test_type"] == "ndrpdr" else False
 257
 258         return self._create_dataframe_from_parquet(
 259             path=self._get_path("trending-ndrpdr"),
 260             partition_filter=lambda_f,
 261             columns=self._get_columns("trending-ndrpdr"),
 262             days=days
 263         )
 264
 265     def read_iterative_mrr(self, release: str) -> DataFrame:
 266         """Read MRR data partition from iterative parquet.
 267
 268         :param release: The CSIT release from which the data will be read.
 269         :type release: str
 270         :returns: Pandas DataFrame with read data.
 271         :rtype: DataFrame
 272         """
 273
 274         lambda_f = lambda part: True if part["test_type"] == "mrr" else False
 275
 276         return self._create_dataframe_from_parquet(
 277             path=self._get_path("iterative-mrr").format(release=release),
 278             partition_filter=lambda_f,
 279             columns=self._get_columns("iterative-mrr")
 280         )
 281
 282     def read_iterative_ndrpdr(self, release: str) -> DataFrame:
 283         """Read NDRPDR data partition from parquet.
 284
 285         :param release: The CSIT release from which the data will be read.
 286         :type release: str
 287         :returns: Pandas DataFrame with read data.
 288         :rtype: DataFrame
 289         """
 290
 291         lambda_f = lambda part: True if part["test_type"] == "ndrpdr" else False
 292
 293         return self._create_dataframe_from_parquet(
 294             path=self._get_path("iterative-ndrpdr").format(release=release),
 295             partition_filter=lambda_f,
 296             columns=self._get_columns("iterative-ndrpdr")
 297         )