########################################################################### # # Copyright (c) 2019-2021 Diality Inc. - All Rights Reserved. # # THIS CODE MAY NOT BE COPIED OR REPRODUCED IN ANY FORM, IN PART OR IN # WHOLE, WITHOUT THE EXPLICIT PERMISSION OF THE COPYRIGHT OWNER. # # @file utils.py # # @author (last) Joseph varghese # @date (last) 15-Jan-2022 # ############################################################################ import csv import test from datetime import timezone from datetime import datetime def get_current_date_and_time_in_utc(date_format='%Y/%m/%d - %H:%M:%S'): """ Method to get current date and time. @input date_format (str) - format of date to be retrieved. @return (str) - date in specified format. """ date = datetime.now(timezone.utc) return str(date.strftime(date_format)) def get_extracted_input_buf_file(): """ This function is the handler for getting file from cloudsync folder. This handler will go inside cloudsync folder and looks for newly added input buf file based on current time. if it satisfied that condition, it will return the exact path of newly created input buf file. Application input buf file is automatically created in '/home/denali/Desktop/sd-card/cloudsync/ {current_date}_inp.buf' @return latest_file - (string) returns latest file that append in cloudsync folder """ try: current_date = get_current_date_and_time_in_utc(date_format = "%Y_%m_%d") latest_file = '/home/denali/Desktop/sd-card/cloudsync/'+current_date+'_inp.buf' return latest_file except: return False def retrive_log_data(readline_count): """ This function is the handler for getting file from cloudsync folder. Application log data is automatically appended on '/home/denali/Desktop/sd-card/cloudsync/ {current_date}_inp.buf' @param: (int) readline_count @return: (list) cloudsync_data returns latest file that append in cloudsync folder """ cloudsync_data = [] count = 0 file_name = get_extracted_input_buf_file() try: with open(file_name,mode = 'r') as filereader: contents = csv.reader(filereader) try: for reader in reversed(list(contents)): if readline_count == count: return cloudsync_data cloudsync_data.append(reader) count = count + 1 except: test.fail("application log data is corrupted") except: test.fail("Log file is not created or log file is not created based on standard log naming format.") def get_epoch_value_consistancy(current_epoch_value, expected_epoch_value): """ This function is verify consistancy of epoch value. @input current_epoch_value (float) - current epoch time @input expected_epoch_value (str) - epoch time from the cloud-sync log. @return (bool) - True/False """ expected_epoch_value = int(expected_epoch_value) maximum_epoch_value = int(current_epoch_value + 25) minimum_epoch_value = int(current_epoch_value - 25) if expected_epoch_value > minimum_epoch_value and expected_epoch_value < maximum_epoch_value : return True return False