Source code for nrgpy.quality.quality

from datetime import datetime


[docs]def check_intervals(df, verbose=True, return_info=False, show_all_missing_timestamps=False, interval=''): """checks for missing intervals in a pandas dataframe with a "Timestamp" column Parameters ---------- df : object the dataframe to be checked interval : int [deprecated] the averaging interval in seconds verbose : bool print results to terminal; False to skip return_info : bool set to True to return dict with below values show_all_missing_timestamps : bool set to True to show all missing timestamps in verbose option. otherwise, shows first and last 3. Returns ---------- dict actual_rows : int actual number of rows in data section of export file (1 subtracted for column headers) expected_rows : int expected number of rows (assumes 10 min. AVG), converts result to whole integer time_range : str range of time represented in export file first_interval : str file starting timestamp last_interval : str file ending timestamp missing_timestamps : list a list of missing timestamps Examples ---------- ex. pass a reader.data dataframe for an interval check: >>> reader = nrgpy.sympro_txt_read() instance created, no filename specified >>> reader.concat_txt(txt_dir="C:/data/sympro_data/000110/") ... >>> nrgpy.check_intervals(reader.data, interval=600) Starting timestamp : 2019-01-01 00:00:00 Ending timestamp : 2019-07-01 04:50:00 Data set Duration : 181 days, 4:50:00 Expected rows in data set : 26093 Actual rows in data set : 26093 Data set complete. """ if "horz" in "".join(df.columns).lower() or isinstance(df['Timestamp'][0], datetime): df2 = df.copy() # df2.Timestamp = df2.Timestamp.apply(lambda x: x.strftime("%Y-%m-%d %H:%M:%S")) df2.reset_index(level=0, inplace=True) _df = df2 first_interval = _df['Timestamp'].min() last_interval = _df['Timestamp'].max() else: _df = df.copy() time_fmt = "%Y-%m-%d %H:%M:%S" first_interval = datetime.strptime(_df['Timestamp'].min(), time_fmt) last_interval = datetime.strptime(_df['Timestamp'].max(), time_fmt) interval = select_interval_length(_df) time_range = last_interval - first_interval expected_rows = int(time_range.total_seconds() / interval) actual_rows = len(df) - 1 loss_pct = round( 100 * (expected_rows - actual_rows) / expected_rows ) if expected_rows != actual_rows: missing_timestamps, _df = find_missing_intervals(_df, interval) if verbose == True: print('Statistical interval : {0} seconds'.format(interval)) print('Starting timestamp : {0}'.format(first_interval)) print('Ending timestamp : {0}'.format(last_interval)) print('Data set Duration : {0}'.format(time_range)) print('Expected rows in data set : {0}'.format(expected_rows)) print('Actual rows in data set : {0}'.format(actual_rows)) if expected_rows == actual_rows: print('\nData set complete.') else: print('Interval loss percentage : {0}'.format(loss_pct)) print('\nMissing {0} timestamps:'.format(len(missing_timestamps))) if len(missing_timestamps) <= 8 or show_all_missing_timestamps == True: for i, timestamp in enumerate(missing_timestamps): print("\t{0}\t{1}".format(i+1,timestamp)) else: for timestamp in missing_timestamps[0:3] + ["..."] + missing_timestamps[-3:]: print("\t{0}\t{1}".format(" ",timestamp)) if return_info == True: interval_info = {} interval_info['actual_rows'] = actual_rows interval_info['expected_rows'] = expected_rows interval_info['first_interval'] = first_interval interval_info['last_interval'] = last_interval interval_info['time_range'] = time_range interval_info['loss_pct'] = loss_pct try: interval_info['missing_timestamps'] = missing_timestamps except: interval_info['missing_timestamps'] = None return interval_info
[docs]def find_missing_intervals(__df, interval): """find gaps in data dataframe returns ---------- list a list of all missing intervals """ _df = __df.copy() import pandas as pd _df['data'] = True _df['Timestamp'] = pd.to_datetime(_df['Timestamp']) _df.set_index('Timestamp', inplace=True) _df = _df.reindex(pd.date_range(start=_df.index[0], end=_df.index[-1], freq='{0}s'.format(interval))) missing_timestamps = [] for index, row in _df.iterrows(): if row['data'] != True: missing_timestamps.append(index) return missing_timestamps, _df
[docs]def select_interval_length(df, seconds=True): """returns the mode of the first 10 intervals of the data set parameters ---------- reader : nrgpy reader object seconds : bool (True) set to False to get interval length in minutes returns ------- int """ from datetime import datetime formatter = "%Y-%m-%d %H:%M:%S" interval = [] for i in range(10): try: interval.append( int( (datetime.strptime(df['Timestamp'].loc[i+1], formatter) - datetime.strptime(df['Timestamp'].loc[i], formatter) ).seconds) ) except: formatter = "%Y-%m-%d %H:%M:%S.%f" interval.append( int( (df['Timestamp'][i+1] - df['Timestamp'][i]).seconds) ) # except: # pass interval_s = select_mode_from_list(interval) interval_m = interval_s/60 try: if seconds: return select_mode_from_list(interval) return select_mode_from_list(interval)/60 except: return False
[docs]def select_mode_from_list(lst): return max(set(lst), key=lst.count)
[docs]def check_for_missing_txt_files(txt_file_names): """ check list of files for missing file numbers parameters ---------- txt_file_names : list list of SymphoniePRO text file exports returns ------- list "missing" text file numbers """ missing_file_numbers = [] for i, f in enumerate(sorted(txt_file_names)): file_number = int(f.split("_")[-2]) if i > 0: if file_number - _file_number > 1: missing_file_numbers.append(f) _file_number = file_number return missing_file_numbers