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 _get_list_of_files(self,
 117         path,
 118         last_modified_begin=None,
 119         last_modified_end=None,
 120         days=None) -> list:
 121         """Get list of interested files stored in S3 compatible storage and
 122         returns it.
 123
 124         :param path: S3 prefix (accepts Unix shell-style wildcards)
 125             (e.g. s3://bucket/prefix) or list of S3 objects paths
 126             (e.g. [s3://bucket/key0, s3://bucket/key1]).
 127         :param last_modified_begin: Filter the s3 files by the Last modified
 128             date of the object. The filter is applied only after list all s3
 129             files.
 130         :param last_modified_end: Filter the s3 files by the Last modified date
 131             of the object. The filter is applied only after list all s3 files.
 132         :param days: Number of days to filter.
 133         :type path: Union[str, List[str]]
 134         :type last_modified_begin: datetime, optional
 135         :type last_modified_end: datetime, optional
 136         :type days: integer, optional
 137         :returns: List of file names.
 138         :rtype: List
 139         """
 140         if days:
 141             last_modified_begin = datetime.now(tz=UTC) - timedelta(days=days)
 142         try:
 143             file_list = wr.s3.list_objects(
 144                 path=path,
 145                 suffix="parquet",
 146                 last_modified_begin=last_modified_begin,
 147                 last_modified_end=last_modified_end
 148             )
 149             if self._debug:
 150                 logging.info("\n".join(file_list))
 151         except NoFilesFound as err:
 152             logging.error(f"No parquets found.\n{err}")
 153         except EmptyDataFrame as err:
 154             logging.error(f"No data.\n{err}")
 155
 156         return file_list
 157
 158     def _create_dataframe_from_parquet(self,
 159         path, partition_filter=None,
 160         columns=None,
 161         validate_schema=False,
 162         last_modified_begin=None,
 163         last_modified_end=None,
 164         days=None) -> DataFrame:
 165         """Read parquet stored in S3 compatible storage and returns Pandas
 166         Dataframe.
 167
 168         :param path: S3 prefix (accepts Unix shell-style wildcards)
 169             (e.g. s3://bucket/prefix) or list of S3 objects paths
 170             (e.g. [s3://bucket/key0, s3://bucket/key1]).
 171         :param partition_filter: Callback Function filters to apply on PARTITION
 172             columns (PUSH-DOWN filter). This function MUST receive a single
 173             argument (Dict[str, str]) where keys are partitions names and values
 174             are partitions values. Partitions values will be always strings
 175             extracted from S3. This function MUST return a bool, True to read
 176             the partition or False to ignore it. Ignored if dataset=False.
 177         :param columns: Names of columns to read from the file(s).
 178         :param validate_schema: Check that individual file schemas are all the
 179             same / compatible. Schemas within a folder prefix should all be the
 180             same. Disable if you have schemas that are different and want to
 181             disable this check.
 182         :param last_modified_begin: Filter the s3 files by the Last modified
 183             date of the object. The filter is applied only after list all s3
 184             files.
 185         :param last_modified_end: Filter the s3 files by the Last modified date
 186             of the object. The filter is applied only after list all s3 files.
 187         :param days: Number of days to filter.
 188         :type path: Union[str, List[str]]
 189         :type partition_filter: Callable[[Dict[str, str]], bool], optional
 190         :type columns: List[str], optional
 191         :type validate_schema: bool, optional
 192         :type last_modified_begin: datetime, optional
 193         :type last_modified_end: datetime, optional
 194         :type days: integer, optional
 195         :returns: Pandas DataFrame or None if DataFrame cannot be fetched.
 196         :rtype: DataFrame
 197         """
 198         df = None
 199         start = time()
 200         if days:
 201             last_modified_begin = datetime.now(tz=UTC) - timedelta(days=days)
 202         try:
 203             df = wr.s3.read_parquet(
 204                 path=path,
 205                 path_suffix="parquet",
 206                 ignore_empty=True,
 207                 validate_schema=validate_schema,
 208                 use_threads=True,
 209                 dataset=True,
 210                 columns=columns,
 211                 partition_filter=partition_filter,
 212                 last_modified_begin=last_modified_begin,
 213                 last_modified_end=last_modified_end
 214             )
 215             if self._debug:
 216                 df.info(verbose=True, memory_usage='deep')
 217                 logging.info(
 218                     u"\n"
 219                     f"Creation of dataframe {path} took: {time() - start}"
 220                     u"\n"
 221                 )
 222         except NoFilesFound as err:
 223             logging.error(f"No parquets found.\n{err}")
 224         except EmptyDataFrame as err:
 225             logging.error(f"No data.\n{err}")
 226
 227         self._data = df
 228         return df
 229
 230     def check_datasets(self, days: int=None):
 231         """Read structure from parquet.
 232
 233         :param days: Number of days back to the past for which the data will be
 234             read.
 235         :type days: int
 236         """
 237         self._get_list_of_files(path=self._get_path("trending"), days=days)
 238         self._get_list_of_files(path=self._get_path("statistics"), days=days)
 239
 240     def read_stats(self, days: int=None) -> tuple:
 241         """Read statistics from parquet.
 242
 243         It reads from:
 244         - Suite Result Analysis (SRA) partition,
 245         - NDRPDR trending partition,
 246         - MRR trending partition.
 247
 248         :param days: Number of days back to the past for which the data will be
 249             read.
 250         :type days: int
 251         :returns: tuple of pandas DataFrame-s with data read from specified
 252             parquets.
 253         :rtype: tuple of pandas DataFrame-s
 254         """
 255
 256         l_stats = lambda part: True if part["stats_type"] == "sra" else False
 257         l_mrr = lambda part: True if part["test_type"] == "mrr" else False
 258         l_ndrpdr = lambda part: True if part["test_type"] == "ndrpdr" else False
 259
 260         return (
 261             self._create_dataframe_from_parquet(
 262                 path=self._get_path("statistics"),
 263                 partition_filter=l_stats,
 264                 columns=self._get_columns("statistics"),
 265                 days=days
 266             ),
 267             self._create_dataframe_from_parquet(
 268                 path=self._get_path("statistics-trending-mrr"),
 269                 partition_filter=l_mrr,
 270                 columns=self._get_columns("statistics-trending-mrr"),
 271                 days=days
 272             ),
 273             self._create_dataframe_from_parquet(
 274                 path=self._get_path("statistics-trending-ndrpdr"),
 275                 partition_filter=l_ndrpdr,
 276                 columns=self._get_columns("statistics-trending-ndrpdr"),
 277                 days=days
 278             )
 279         )
 280
 281     def read_trending_mrr(self, days: int=None) -> DataFrame:
 282         """Read MRR data partition from parquet.
 283
 284         :param days: Number of days back to the past for which the data will be
 285             read.
 286         :type days: int
 287         :returns: Pandas DataFrame with read data.
 288         :rtype: DataFrame
 289         """
 290
 291         lambda_f = lambda part: True if part["test_type"] == "mrr" else False
 292
 293         return self._create_dataframe_from_parquet(
 294             path=self._get_path("trending-mrr"),
 295             partition_filter=lambda_f,
 296             columns=self._get_columns("trending-mrr"),
 297             days=days
 298         )
 299
 300     def read_trending_ndrpdr(self, days: int=None) -> DataFrame:
 301         """Read NDRPDR data partition from iterative parquet.
 302
 303         :param days: Number of days back to the past for which the data will be
 304             read.
 305         :type days: int
 306         :returns: Pandas DataFrame with read data.
 307         :rtype: DataFrame
 308         """
 309
 310         lambda_f = lambda part: True if part["test_type"] == "ndrpdr" else False
 311
 312         return self._create_dataframe_from_parquet(
 313             path=self._get_path("trending-ndrpdr"),
 314             partition_filter=lambda_f,
 315             columns=self._get_columns("trending-ndrpdr"),
 316             days=days
 317         )
 318
 319     def read_iterative_mrr(self, release: str) -> DataFrame:
 320         """Read MRR data partition from iterative parquet.
 321
 322         :param release: The CSIT release from which the data will be read.
 323         :type release: str
 324         :returns: Pandas DataFrame with read data.
 325         :rtype: DataFrame
 326         """
 327
 328         lambda_f = lambda part: True if part["test_type"] == "mrr" else False
 329
 330         return self._create_dataframe_from_parquet(
 331             path=self._get_path("iterative-mrr").format(release=release),
 332             partition_filter=lambda_f,
 333             columns=self._get_columns("iterative-mrr")
 334         )
 335
 336     def read_iterative_ndrpdr(self, release: str) -> DataFrame:
 337         """Read NDRPDR data partition from parquet.
 338
 339         :param release: The CSIT release from which the data will be read.
 340         :type release: str
 341         :returns: Pandas DataFrame with read data.
 342         :rtype: DataFrame
 343         """
 344
 345         lambda_f = lambda part: True if part["test_type"] == "ndrpdr" else False
 346
 347         return self._create_dataframe_from_parquet(
 348             path=self._get_path("iterative-ndrpdr").format(release=release),
 349             partition_filter=lambda_f,
 350             columns=self._get_columns("iterative-ndrpdr")
 351         )