import pymssql import time import os import pytz from datetime import datetime from datetime import timedelta import traceback from collections import OrderedDict import src.settings as settings from src.containers.DataContainer import DataContainer from src.dbconnectors.db import DBConnector from pymssql import ProgrammingError class DataDB(DBConnector): fmt = '%Y-%m-%d %H:%M:%S' correction_map = {'INVERTER': { 'F_AC': { 'min': 0.0, 'max': 60}, 'E_TOTAL': { 'min': 1, 'max': 2**63 - 1}, 'T_INV': { 'min': -60, 'max': 150}, 'I_DC_TOTAL': { 'min': -10, 'max': 4000}, 'P_DC': { 'min': -10, 'max': 5000 * 1000}, 'U_DC': { 'min': -10, 'max': 2000}, 'I_DC_': { 'min': -10, 'max': 2000}, 'P_DC_': { 'min': -10, 'max': 200 * 1000}, 'U_DC_': { 'min': -10, 'max': 2000}, 'Q_AC': { 'min': -10 * 1000 * 1000, 'max': 10 * 1000 * 1000}, 'P_AC_L1': { 'min': -20000, 'max': 20000}, 'P_AC_L2': { 'min': -20000, 'max': 20000}, 'P_AC_L3': { 'min': -20000, 'max': 20000} }, 'METEO': { 'P_RAD': { 'min': -10, 'max': 2000}, 'T_M': { 'min': -60, 'max': 100}, 'T_A': { 'min': -60, 'max': 60} }, 'STRINGBOX': { 'I_': { 'min': -10, 'max': 200}, 'U_DC': { 'min': -10, 'max': 2000} } } datatype_filter = {'real': { 'min': [-3.4e+38, -1.18e-38], 'max': [1.18e-38, 3.40e+38], 'precision': 38}, 'float': { 'min': [-1.79e+308, -2.23e-308], 'max': [2.23e-308, 1.79e+308], 'precision': 53}, 'int': { 'min': -2**31, 'max': 2**31 - 1}, 'bigint': { 'min': -2**63, 'max': 2**63 - 1} } def __init__(self, host, user, password, database): DBConnector.__init__(self, host, user, password, database) self.logger = settings.logger # retry count self.retry_count = 0 # hold device information self.device_info = {} # private plant time zone self.__plant_timezone = None self.__plant_res = None # day yield cache self.__day_min_yield = {} self.nominal_acp = None def __create_etoday_for_inverter(self, cur, device_type, tbl): if type == 'INVERTER': sql = """IF NOT EXISTS ( SELECT * FROM sys.columns WHERE object_id = OBJECT_ID(N'[dbo].[{_tbl}]') AND name = 'COL_E_TODAY_CALC') ALTER TABLE {_tbl1} ADD COL_E_TODAY_CALC bigint""".format(_tbl=tbl, _tbl1=tbl) # create COL_E_TODAY_CALC value for inverter cur.executes(sql) self._connection.commit() @staticmethod def __check_device_version(device_id, device_info, device_data, list_status_code): if device_data['version'] != "ignore": if device_info is None: list_status_code[device_id] = 'UNKNOWN_DEVICE' return False elif device_info['version'] != device_data['version']: list_status_code[device_id] = 'UNKNOWN_DEVICE_VERSION' return False return True @staticmethod def __get_device_version(cur, db, tbl): sql = """USE {_db}; SELECT TOP 1 device_value FROM xxx_RAW_Device_Properties WHERE table_name=%s AND device_name='Config_Version' ORDER BY ID ASC""".format(_db=db) cur.execute(sql, (tbl,)) res = cur.fetchall() for row in res: return row[0] @staticmethod def __chunk_arr(arr, chunk_size): for i in range(0, len(arr), chunk_size): yield arr[i:i + chunk_size] @staticmethod def __get_device_values(cur, db, tbl): values = {} sql = """USE {_db}; SELECT A.device_value, A.column_name, A.device_vpc, B.value_type, B.datatype FROM xxx_RAW_Device_Properties AS A JOIN commondb.dbo.value_type AS B ON A.device_vpc=B.vpc WHERE table_name=%s AND device_vpc is not NULL""".format(_db=db) cur.execute(sql, (tbl, )) res = cur.fetchall() for row in res: vid, colname, vpc, value_type, data_type = row values[vid] = {'column_name': colname, 'vpc': vpc, 'value_type': value_type, 'data_type': data_type} return values @staticmethod def __scale_data(device_info, data): if device_info['type'].upper() == 'INVERTER': vid_to_scale = {} # find values which need to be scaled values = device_info['values'] for vid in values: value = values[vid] if value['value_type'] in ['E_DAY_KWH', 'E_TOTAL_KWH']: vid_to_scale[vid] = 1000 for ts in data: d = data[ts] for vid in vid_to_scale: value = d.get(vid) if value is not None and value != 'NULL': d[vid] = str(int(float(value) * vid_to_scale[vid])) def __get_device_info(self, cur, logger_id, device_id, list_status_codes): device_info = {} sql = """USE {_db}; SELECT TOP 1 A.Table_name, B.Table_Name FROM xxx_Device_Mappings AS B INNER JOIN xxx_RAW_Device_Properties AS A ON B.RAW_Table_Name=A.TABLE_NAME WHERE A.DEVICE_NAME='ldid' AND A.DEVICE_VALUE=%s""".format(_db=self.database) ldid = "{_lid}_{_did}".format(_lid=logger_id, _did=device_id) cur.execute(sql, (ldid,)) res = cur.fetchone() if res is None: list_status_codes[device_id] = 'UNKNOWN_DEVICE' return device_info['tbl'] = res[0] device_info['dest_tbl'] = res[1] # fetch device class for device sql = """SELECT DEVICE_VALUE FROM xxx_Raw_Device_Properties WHERE TABLE_NAME=%s AND DEVICE_NAME='Device_Class'""" cur.execute(sql, (device_info['tbl'],)) res = cur.fetchone() device_info['type'] = res[0] """ If the device type is an inverter check if there is an etoday value and if not create one.""" self.__create_etoday_for_inverter( cur, device_info['type'], device_info['tbl']) device_info['version'] = DataDB.__get_device_version( cur, self.database, device_info['tbl']) # get values for device device_info['values'] = DataDB.__get_device_values( cur, self.database, device_info['tbl']) return device_info def __get_plant_timezone(self, cur): if self.__plant_timezone is None: sql = """SELECT Value FROM xxx_Plant_Device_Properties WHERE Value_key='plant.timezone.identifier'""" cur.execute(sql) res = cur.fetchall() tzIdentifier = 'Europe/Berlin' for row in res: tzIdentifier = row[0] self.__plant_timezone = pytz.timezone(tzIdentifier) return self.__plant_timezone def __get_plant_res(self, cur, ): if self.__plant_res is None: sql = """SELECT Value FROM xxx_Plant_Device_Properties WHERE Value_key='plant.data.resolution'""" cur.execute(sql) res = cur.fetchone() if res is not None: self.__plant_res = int(res[0]) return self.__plant_res def __get_last_etoday(self, cur, device, ts): device_id = device.get('device_id') device_info = device.get('device_info') dtts = datetime.strptime(ts, DataDB.fmt) ts_UTC_localized = pytz.utc.localize(dtts) ts_plant = ts_UTC_localized.astimezone(self.__get_plant_timezone(cur)) ts_midnight = ts_plant.replace(hour=0, minute=0, second=0) #ts_end_of_day = ts_plant.replace(hour=23, minute=59, second=59) start_time_dtts = ts_midnight.astimezone(pytz.utc) start_time = datetime.strftime(start_time_dtts, DataDB.fmt) #end_time = ts_end_of_day.astimezone(pytz.utc) sql = """SELECT TOP 1 COL_E_TODAY_CALC FROM dbo.{_tbl} WHERE Date_and_time >= %s AND Date_and_time < %s AND COL_E_TODAY_CALC IS NOT NULL ORDER BY Date_and_time DESC""".format(_tbl=device_info['dest_tbl']) cur.execute(sql, (start_time, ts)) res = cur.fetchone() if res is not None: last_etoday_yield = res[0] else: last_etoday_yield = 0 return last_etoday_yield def __get_first_yield_of_the_day(self, cur, device, ts, col_total_yield): device_id = device.get('device_id') device_info = device.get('device_info') # transform timestamp dtts = datetime.strptime(ts, DataDB.fmt) ts_UTC_localized = pytz.utc.localize(dtts) ts_plant = ts_UTC_localized.astimezone(self.__get_plant_timezone(cur)) # extract day of the date ts_midnight = ts_plant.replace(hour=0, minute=0, second=0) ts_end_of_day = ts_plant.replace(hour=23, minute=59, second=59) device_yield = self.__day_min_yield.get(device_id) if device_yield: midnight_yield = device_yield.get(ts_midnight.strftime(DataDB.fmt)) if midnight_yield: return midnight_yield start_time = ts_midnight.astimezone(pytz.utc) end_time = ts_end_of_day.astimezone(pytz.utc) if device_info is None: self.logger.warning( "DataDB: can't get first yield of the day since there is no device info set.") return None """If there is no yield for this day we need to fetch this from the database. We are trying to fetch the TOTAL YIELD value form the inverter""" try: '''special case: for some inverters there is an error in the midnight value for etotal, so we want to have the highest value around midnight -> TOP 3 instead of TOP 1''' sql = """SELECT TOP 3 {_col} FROM dbo.{_tbl} WHERE Date_and_time >= %s AND Date_and_time <= %s AND {_col} IS NOT NULL ORDER BY Date_and_time ASC""".format(_tbl=device_info['dest_tbl'], _col=col_total_yield) cur.execute(sql, (start_time, end_time)) except Exception, err: if "Invalid column name" in err[1]: #self.logger.debug("Could not find COL_SR_196767 in device for fetching the etoday. {_did}".format(_did=device_id)) # self.logger.debug(err) """ We might need to adjust values here in case the total yield of the inverter comes from different importer like skytron or meteocontrol. syktron: class: E class2: E_LIFE""" sql = """SELECT Column_Name FROM dbo.xxx_virtual_devices WHERE table_name=%s AND ((Device_Class='E' OR Device_Class='E_TOTAL') AND (Device_Class_2 = 'E_LIFE' OR Device_Class_2='E_TOTAL'))""" cur.execute(sql, (device_info.get('dest_tbl'))) column_name = None res = cur.fetchall() for row in res: column_name = row[0] """Try to fetch data with the column""" if column_name: #self.logger.debug("Found another vpc providing e_total: {_vid}".format(_vid=column_name)) sql = """SELECT TOP 3 {_column} FROM dbo.{_tbl} WHERE Date_and_time >= %s AND Date_and_time <= %s AND {_column} IS NOT NULL ORDER BY Date_and_time ASC""".format(_column=column_name, _tbl=device_info['dest_tbl']) cur.execute(sql, (start_time, end_time)) else: self.logger.debug( "Could not find E_TOTAL value for {_did} (Searched: COL_SR_196767 and class_2 E_TOTAL)".format(_did=device_id)) else: """Return None since we can't find a data point for the first yield of the day.""" self.logger.error(err) return None res = cur.fetchall() res_yield = None for row in res: '''special case: for some inverters there is an error in the midnight value for etotal, so we want to have the highest value around midnight''' # if res_yield is None or res_yield >= row[0]: # continue res_yield = row[0] if not res_yield: self.logger.warning( "DataDB: no first day yield available from database. Searching in data now") data = device.get('data') yield_vid = None values = device_info['values'] for vid in values: value = values[vid] if value['value_type'] == 'E_TOTAL_KWH': yield_vid = vid break for next_ts in data: if next_ts > ts: self.logger.warning( "DataDB: no first yield of the day found for this timestamp: {_ts}".format(_ts=ts)) return None d = data.get(next_ts) res_yield = d.get(yield_vid) if res_yield is None: self.logger.warning( "DataDB: can't get first yield of the day since there is datum available for value type E_TOTAL_KWH in data.") else: self.logger.info("DataDB: found day yield from set data") break if not self.__day_min_yield.get(device_id): self.__day_min_yield[device_id] = OrderedDict() self.__day_min_yield[device_id][ts_midnight.strftime( DataDB.fmt)] = res_yield return res_yield def __calc_etoday(self, cur, device, list_status_codes): device_info = device.get('device_info') vid_total_yield = None vid_pac = None # find values which need to be scaled values = device_info['values'] for vid in values: value = values[vid] if value['value_type'] in ['E_TOTAL_KWH', 'E_TOTAL']: vid_total_yield = vid col_total_yield = value.get('col_name') break for vid in values: value = values[vid] if value['value_type'] == 'P_AC_TOTAL': vid_pac = vid break if vid_total_yield is not None: self.__calc_etoday_from_total_yield( cur, device, list_status_codes, vid_total_yield, col_total_yield) elif vid_pac is not None: self.__calc_etoday_from_acp( cur, device, list_status_codes, vid_pac) def __calc_etoday_from_total_yield(self, cur, device, list_status_codes, vid_total_yield, col_total_yield): # self.logger.debug('__calc_etoday_from_total_yield') device_info = device.get('device_info') data = device.get('data') tz = self.__get_plant_timezone(cur) for ts in data: midnight_yield = self.__get_first_yield_of_the_day( cur, device, ts, col_total_yield) if midnight_yield is not None: d = data.get(ts) current_total_yield = d.get(vid_total_yield) if current_total_yield is not None and current_total_yield != 'NULL': # add yield today to the inverter #d['COL_E_TODAY_CALC'] = str(long(current_total_yield) - long(midnight_yield)) # COL_E_TODAY_CALC correction calculation try: col_e_today_calc_val = long( current_total_yield) - long(midnight_yield) except ValueError as e: col_e_today_calc_val = long( float(current_total_yield)) - long(midnight_yield) if col_e_today_calc_val >= 0: d['COL_E_TODAY_CALC'] = str(col_e_today_calc_val) else: self.logger.warning("Calculated negative value " + str( col_e_today_calc_val) + " for COL_E_TODAY_CALC. Midnight yield: " + str(midnight_yield)) self.logger.warning( "Current total yield: " + str(current_total_yield)) ''' col_e_today_calc_val = long(current_total_yield) - long(midnight_yield) if col_e_today_calc_val>=0: d['COL_E_TODAY_CALC'] = str(col_e_today_calc_val) else: self.logger.begug('Calculated negative value ') d['COL_E_TODAY_CALC'] = None ''' ################################### values = device_info['values'] if values.get('COL_E_TODAY_CALC') is None: values['COL_E_TODAY_CALC'] = {'value_type': u'COL_E_TODAY_CALC', 'data_type': u'bigint', 'vpc': u'COL_E_TODAY_CALC', 'column_name': u'COL_E_TODAY_CALC'} def __calc_etoday_from_acp(self, cur, device, list_status_codes, vid_pac): self.logger.debug('__calc_etoday_from_acp') device_info = device.get('device_info') res = self.__get_plant_res(cur) if res is None or res == 'NULL': self.logger.warning( "No resolution defined for plant. Can't calculate etoday_from_acp") return data = device.get('data') """ this line is needed in case that the timestamps are not ordered chronologically """ ts_arr = list(sorted(data.keys())) for ts in ts_arr: d = data.get(ts) acp = d.get(vid_pac) prev_dtts = datetime.strptime( ts, self.fmt) - timedelta(seconds=res * 60) prev_ts = datetime.strftime(prev_dtts, self.fmt) if acp is not None and acp != 'NULL': if data.get(prev_ts) is not None: prev_yield_today = data[prev_ts].get('COL_E_TODAY_CALC') else: prev_yield_today = self.__get_last_etoday(cur, device, ts) try: current_yield = float(acp) * (float(res) / 60) col_e_today_calc_val = str( float(prev_yield_today) + float(current_yield)) except Exception as e: continue if col_e_today_calc_val >= 0: d['COL_E_TODAY_CALC'] = str(col_e_today_calc_val) else: self.logger.warning("Calculated negative value {_val}".format( _val=col_e_today_calc_val)) values = device_info['values'] if values.get('COL_E_TODAY_CALC') is None: values['COL_E_TODAY_CALC'] = {'value_type': u'COL_E_TODAY_CALC', 'data_type': u'bigint', 'vpc': u'COL_E_TODAY_CALC', 'column_name': u'COL_E_TODAY_CALC'} @staticmethod def __prepare_sql_string(device, date_filter_list=None, storage_delay=True): data = device.get('data') device_info = device.get('device_info') if storage_delay: sql_column_name = "Date_and_time, Storage_time, Storage_delay" sql_value_data_placeholder = "%s, %s, %s" else: sql_column_name = "Date_and_time, Storage_time" sql_value_data_placeholder = "%s, %s" values = device_info.get('values') sql_data = OrderedDict() now_obj = datetime.utcnow() for vid in values: value = values.get(vid) sql_column_name += ", {_col}".format(_col=value['column_name']) sql_value_data_placeholder += ", %s" for ts in data: ts_obj = datetime.strptime(ts, DataDB.fmt) if ts_obj in date_filter_list: continue delay = now_obj - ts_obj dsec = int(delay.total_seconds()) now = now_obj.strftime('%Y-%m-%d %H:%M:%S') data_dict = data.get(ts) if sql_data.get(ts) is None: if storage_delay: sql_data[ts] = ["'{_ts}'".format(_ts=ts), "'{_ts}'".format( _ts=now), "'{_sec}'".format(_sec=dsec)] else: sql_data[ts] = ["'{_ts}'".format( _ts=ts), "'{_ts}'".format(_ts=now)] v = data_dict.get(vid, "NULL") if value.get('data_type').lower() in ['text', 'string', 'nvarchar(32)', 'varchar(32)', 'varchar(200)']: sql_data[ts].append("'{_v}'".format(_v=v)) else: sql_data[ts].append(v) sql_row = "" first_row = True prepared_sql_insert_queries = [] for idx, ts in enumerate(sql_data): row = sql_data.get(ts) if first_row: first_row = False sql_row += "({_d})".format(_d=",".join(row)) else: sql_row += ",({_d})".format(_d=",".join(row)) # build sql statement since we only can do 1000 inserts at once if (idx + 1) % 1000 == 0: sql = "INSERT INTO {_tbl} ({_col}) VALUES {_val}".format(_tbl=device_info['dest_tbl'], _col=sql_column_name, _val=sql_row) # reset sql_row and first_row to start the sql query over again sql_row = "" first_row = True if len(sql) > 0: prepared_sql_insert_queries.append(sql) # add last query if len(sql_row) > 0: sql = "INSERT INTO {_tbl} ({_col}) VALUES {_val}".format(_tbl=device_info['dest_tbl'], _col=sql_column_name, _val=sql_row) prepared_sql_insert_queries.append(sql) return prepared_sql_insert_queries @staticmethod def __prepare_sql_overwrite(device, date_filter_list=None): data = device.get('data') device_info = device.get('device_info') sql_column_name = "Date_and_time, Storage_time" values = device_info.get('values') sql_data = OrderedDict() now = datetime.now().strftime('%Y-%m-%d %H:%M:%S') update_sql = OrderedDict() for vid in values: value = values.get(vid) sql_column_name += ", {_col}".format(_col=value['column_name']) for ts in data: data_dict = data.get(ts) if sql_data.get(ts) is None: sql_data[ts] = ["'{_ts}'".format( _ts=ts), "'{_ts}'".format(_ts=now)] v = data_dict.get(vid, "NULL") if value.get('data_type').lower() in ['text', 'string', 'nvarchar(32)', 'varchar(32)', 'varchar(200)']: sql_data[ts].append("'{_v}'".format(_v=v)) else: sql_data[ts].append(v) if update_sql.get(ts) is None: update_sql[ts] = "" update_sql[ts] += "{_col}=".format(_col=value['column_name']) if value.get('data_type').lower() in ['text', 'string', 'nvarchar(32)', 'varchar(32)', 'varchar(200)']: update_sql[ts] += "'{_v}',".format(_v=v) else: update_sql[ts] += "{_v},".format(_v=v) prepared_sql_queries = [] chunk_size = 50 ts_arr = sql_data.keys() ts_chunks = list(DataDB.__chunk_arr(ts_arr, chunk_size)) for chunk in ts_chunks: sql = "" for ts in chunk: row = sql_data.get(ts) sql_insert_row = "({_d})".format(_d=",".join(row)) sql += """IF NOT EXISTS(SELECT * FROM {_tbl} WHERE Date_and_time='{_ts}') BEGIN INSERT INTO {_tbl} ({_col}) VALUES {_val} END ELSE BEGIN UPDATE {_tbl} SET {_update_sql} WHERE Date_and_time='{_ts}' END;""".format(_tbl=device_info['dest_tbl'], _col=sql_column_name, _val=sql_insert_row, _ts=ts, _update_sql=update_sql[ts][:-1]) prepared_sql_queries.append(sql) return prepared_sql_queries @staticmethod def __get_date_data_from_device(cur, device): data = device.get('data') dates = sorted(data.keys()) start_date_utc = datetime.strptime(next(iter(dates)), DataDB.fmt) end_date_utc = datetime.strptime(next(reversed(dates)), DataDB.fmt) device_info = device.get('device_info') sql = """SELECT Date_and_time AS [date] FROM dbo.{_tbl} WHERE Date_and_time>=%s AND Date_and_time <= %s""".format(_tbl=device_info['dest_tbl']) cur.execute(sql, (start_date_utc, end_date_utc)) res = cur.fetchall() date_filter = [] for row in res: date_filter.append(row[0]) return date_filter def __set_data(self, cur, container, device, list_status_codes, last_data_values): """Returns the timestamps which has been inserted into the database for the device""" self.logger.debug("DataDB: {_pid} {_lid} {_did}".format(_pid=os.getpid(), _lid=container.loggerId, _did=device.get('device_id'))) # return if there is no data to add if len(device['data']) == 0: self.logger.debug("DataDB: {_did} device has no data to insert.".format( _did=device.get('device_id'))) return device_id = device.get('device_id') device_info = self.__get_device_info( cur, container.loggerId, device_id, list_status_codes) device['device_info'] = device_info # if the version failed return if not container.ignore_version: if not DataDB.__check_device_version(device_id, device_info, device, list_status_codes): return storage_delay = True if not self.__add_storage_delay_col(device_info['tbl'], device_info['type'].upper(), cur): storage_delay = False # scale data DataDB.__scale_data(device_info, device.get('data')) # data point filter try: self.__check_datatype_range(device_id, device) if device_info['type'].upper() == 'INVERTER': self.__filter_acp(device, container.loggerId, cur) self.__filter_etotal(device, container.loggerId, cur) self.__filter_generic_values(device, device_info['type'].upper()) except Exception as e: self.logger.error( 'An error occurred in the data point filter code') self.logger.error(str(e)) ################################## # calculate etoday for device if device_info['type'].upper() == 'INVERTER': try: self.__calc_etoday(cur, device, list_status_codes) except Exception as e: self.logger.error( "An error occurred in __calc_etoday: " + str(e)) try: if not container.overwrite: # remove all duplicated timestamps date_filter_list = DataDB.__get_date_data_from_device( cur, device) sql_query_list = self.__prepare_sql_string( device, date_filter_list, storage_delay) else: date_filter_list = [] self.logger.info("Overwriting data") sql_query_list = self.__prepare_sql_overwrite(device) except Exception as e: self.logger.warning("Error in overwrite code") date_filter_list = DataDB.__get_date_data_from_device(cur, device) sql_query_list = self.__prepare_sql_string( device, date_filter_list, storage_delay) for sql_query in sql_query_list: cur.execute(sql_query) if len(sql_query_list) > 0: self._connection.commit() device['status'] = 'SUCCESS' @staticmethod def cmp_dirty_ts(a, b): a_dt = datetime.strptime(a, DataDB.fmt) b_dt = datetime.strptime(b, DataDB.fmt) if a_dt < b_dt: return -1 if a_dt > b_dt: return 1 return 0 def set_data(self, container): self.logger.debug( "DataDB: {_pid} entered set_data".format(_pid=os.getpid())) if self._connection != None: cur = self._connection.cursor() cur.execute('USE {_db}'.format(_db=self.database)) cur.execute('select db_name()') curr_db = cur.fetchone()[0] if curr_db != self.database: self.logger.error('Mismatch between current db and routing db. Exiting') raise Exception list_tsstatus_codes = {} devices = {} dirty_ts = [] # transform data into timestamps per device for data in container.timestampData: for dev in data.deviceData: device = devices.get(dev.deviceId) if not device: device = OrderedDict({'device_id': dev.deviceId, 'data': OrderedDict(), 'version': None}) devices[dev.deviceId] = device values = OrderedDict() for v in dev.valueData: values[v.id] = v.value device['data'][data.timestamp] = values device['version'] = dev.version if not data.timestamp in dirty_ts: dirty_ts.append(data.timestamp) dirty_ts.sort(cmp=DataDB.cmp_dirty_ts) # hidden device hidden_devices = [] try: self.__create_hidden_devs_table(cur) self._connection.commit() hidden_devices = self.__get_hidden_devices( cur, container.loggerId) except Exception as e: self.logger.error( 'Something went wrong with the hidden device code') ################################### """Activate PAS service for this plant if not done yet.""" self.__activate_pas_service(cur) list_status_codes = {} for device_id in devices: if device_id in hidden_devices: list_status_codes[device_id] = 'SUCCESS' self.logger.debug( 'DataDB: device {_did} is marked as hidden. Going to skip'.format(_did=device_id)) continue self.logger.debug( "DataDB processing data for {_did}".format(_did=device_id)) last_data_values = {} device = devices.get(device_id) if len(device) == 0: continue try_count = 0 while(try_count <= self.retry_count): try: self.__set_data(cur, container, device, list_status_codes, last_data_values) try_count = self.retry_count + 1 except Exception as e: self._connection.rollback() if str(e).find("deadlock") > -1: self.logger.error( "DataDB: deadlock {_did}".format(_did=device_id)) list_status_codes[dev.deviceId] = 'SERVER_SIDE_ERROR' try_count += 1 elif str(e).find("Violation of PRIMARY KEY constraint") > -1: self.logger.info( "DataDB: duplicated data found and ignored for device: {_d}".format(_d=device_id)) list_status_codes[dev.deviceId] = 'SUCCESS' try_count = self.retry_count + 1 else: self.logger.error('DataDB: Exception raised(' + str(os.getpid()) + ') ' + container.loggerId + ' ' + device_id + ' ' + str(traceback.format_exc())) self.logger.error(str(e)) list_status_codes[device_id] = 'SERVER_SIDE_ERROR' try_count = self.retry_count + 1 #ts = device['data'].keys()[0] if len(list_status_codes) > 0: for ts in device['data'].keys(): list_tsstatus_codes[ts] = list_status_codes #if len(list_status_codes) > 0: # dirty_ts.remove(datetime.strptime(ts, DataDB.fmt)) # insert dirty timestamps into table. try: self.__insert_dirty_ts(cur, dirty_ts) except ProgrammingError as err: if "Invalid object name" in err[1]: try: self.__create_dirty_ts_table(cur) self._connection.commit() except Exception as err1: self.logger.error("ERROR - could not create xxx_dirty_timestamps table") self.logger.error(str(e)) self.__insert_dirty_ts(cur, dirty_ts) except Exception as e: self.logger.error('An error ocurred while trying to insert dirty timestamps: ' + str(e)) # do not update the last data tables for devices if not container.bulk_insert: self.__update_last_data_for_devices(cur, devices) return list_tsstatus_codes def setData(self, container): list_status_codes = {} list_status_codes = self.set_data(container) return list_status_codes def __update_last_data_for_devices(self, cur, devices): delete_device_tbl_list = [] insert_device_tbl_list = [] for device_id in devices: device = devices.get(device_id) device_info = device.get('device_info') if device.get('status') == 'SUCCESS': tbl = device_info.get('dest_tbl') if tbl and device_info.get('type') == 'INVERTER': delete_device_tbl_list.append("'{_tbl}'".format(_tbl=tbl)) data = device.get('data') last_added_ts = next(reversed(data)) data_values = data.get(last_added_ts) # add COL_E_TODAY_CALC to insert list try: insert_device_tbl_list.append("('{_ts}', '{_tbl}', {_val}, '{_col}', 'Energy_Yield')" .format(_ts=last_added_ts, _tbl=tbl, _val=data_values['COL_E_TODAY_CALC'], _col='COL_E_TODAY_CALC')) except KeyError, err: self.logger.warning( "[{_did}] - Device has no COL_E_TODAY_CALC value".format(_did=tbl)) # print "Key ERROR" # print err # print "device_id: {_did}".format(_did=device_id) # find inverter ac power COL_SR_196620 () values = device_info['values'] for vid in values: value = values[vid] data_value = data_values.get(vid) if value['vpc'] == '196620' and data_value: insert_device_tbl_list.append("('{_ts}', '{_tbl}', {_val}, '{_col}', 'Inverter_Power_AC')" .format(_ts=last_added_ts, _tbl=tbl, _val=data_value, _col=value['column_name'])) break # find inverter dc power COL_SR_196626 () values = device_info['values'] for vid in values: value = values[vid] data_value = data_values.get(vid) if value['vpc'] == '196626' and data_value: insert_device_tbl_list.append("('{_ts}', '{_tbl}', {_val}, '{_col}', 'Inverter_Power_DC')" .format(_ts=last_added_ts, _tbl=tbl, _val=data_value, _col=value['column_name'])) break if len(delete_device_tbl_list) > 0: sql_tbl = "({_tbl})".format(_tbl=",".join(delete_device_tbl_list)) sql = """DELETE FROM dbo.xxx_Last_Data WHERE Table_Name in {_tbl} """.format( _tbl=sql_tbl) # + toBeDeletedString cur.execute(sql) if len(insert_device_tbl_list) > 0: sql_val = "{_val}".format(_val=",".join(insert_device_tbl_list)) sql = """INSERT INTO dbo.xxx_Last_Data (Date_and_Time, Table_Name, Value_Value, Value_Type, Value_Name) VALUES {_val}""".format( _val=sql_val) cur.execute(sql) # print cur.rowcount self._connection.commit() def __create_dirty_ts_table(self, cur): sql = """ IF NOT EXISTS (SELECT * FROM dbo.sysobjects where id = object_id(N'dbo.[xxx_Dirty_Timestamps]') and OBJECTPROPERTY(id, N'IsTable') = 1) BEGIN CREATE TABLE dbo.[xxx_Dirty_Timestamps] ( data_ts datetime, state nvarchar(10) NOT NULL DEFAULT 'dirty', modified datetime CONSTRAINT dirty_timestamps_pk PRIMARY KEY (data_ts) ) END """ cur.execute(sql) def __add_storage_delay_col(self, table_name, dev_category, cur): '''get device alias, if no alias is found - no need to add column ''' sql = """ SELECT Virtual_Device_Name FROM xxx_Virtual_Devices WHERE TABLE_NAME='{_table}'""".format(_table=table_name) cur.execute(sql) res = cur.fetchall() if res != []: alias = res[0] else: return False sql = """ IF NOT EXISTS (SELECT * FROM dbo.syscolumns where id = object_id(N'dbo.[{_table}]') AND name ='Storage_delay') BEGIN ALTER TABLE dbo.[{_table}] ADD Storage_delay int END """.format(_table=table_name) cur.execute(sql) sql = """IF NOT EXISTS (SELECT Table_Name FROM xxx_Virtual_Devices (READUNCOMMITTED) WHERE Table_Name=%s AND Column_Name=%s) INSERT INTO xxx_Virtual_Devices (Table_Name, Column_Name, Device_Type, Device_Class, Device_Class_2, Virtual_Device_Name, Translation_Key, Station_ID, Unit_String) VALUES(%s, %s, %s, %s, %s, %s, %s, '-1', %s) """ params = (table_name, 'Storage_delay', table_name, 'Storage_delay', dev_category, 'DELAY', 'DELAY', alias, 'DELAY', 'seconds') cur.execute(sql, params) return True def __insert_dirty_ts(self, cur, dirty_ts): """inserts or updates new dirty timestamps. This function should be called after all data of the setData has been inserted into the database.""" """separate ts to insert and update separatly""" utc_now = datetime.utcnow() """ if there is no datetime """ if len(dirty_ts) <= 0: return first_ts = dirty_ts[0] last_ts = dirty_ts[len(dirty_ts) - 1] sql_query = """SELECT data_ts AS date FROM xxx_Dirty_Timestamps WHERE data_ts >= convert(datetime, %s, 21) AND data_ts <= convert(datetime, %s, 21)""" cur.execute(sql_query, (first_ts, last_ts)) res = cur.fetchall() dirty_ts_update = [] for row in res: date = row[0] dirty_ts_update.append(date) if date.strftime(DataDB.fmt) in dirty_ts: dirty_ts.remove(date.strftime(DataDB.fmt)) """ update all dirty timestamps """ if len(dirty_ts_update) > 0: # may be this can be improved and put into one query but since this case should be seldom we # can keep it this way for now. sql = """UPDATE xxx_Dirty_Timestamps SET state='dirty', modified=GETUTCDATE() WHERE data_ts=%s""" for date in dirty_ts_update: cur.execute(sql, (date)) self._connection.commit() """if there are no more dirty timestamps return.""" if len(dirty_ts) == 0: return sql_list = [] is_first = True for ts in dirty_ts: if is_first: is_first = False sql_list.append( "(convert(datetime, '{_dt}', 21), 'dirty', GETUTCDATE())".format(_dt=ts)) else: sql_list.append( ",(convert(datetime, '{_dt}', 21), 'dirty', GETUTCDATE())".format(_dt=ts)) sql_many = """IF NOT EXISTS (SELECT data_ts FROM xxx_Dirty_Timestamps WHERE data_ts=convert(datetime, '{_dt}', 21)) INSERT INTO xxx_Dirty_Timestamps (data_ts, state, modified) VALUES""".format(_dt=ts) sql_many += " ".join(sql_list) try: cur.execute(sql_many) self._connection.commit() except ProgrammingError as err: if "Invalid object name" in err[1]: try: self.__create_dirty_ts_table(cur) self._connection.commit() except Exception as err1: self.logger.error( "ERROR - could not create xxx_dirty_timestamps table") self.logger.error(str(e)) cur.execute(sql_many) self._connection.commit() except Exception as err: self.logger.error("ERROR while inserting dirty timestamps") self.logger.error(str(err)) self.logger.error(sql_many) self.logger.error("res") self.logger.error(res) self.logger.error(sql_query) self.logger.error(first_ts) self.logger.error(last_ts) def __activate_pas_service(self, cur): """Function activates the PAS service for this plant if not set yet. This should only happen if the titanium is importing the data""" sql = """SELECT * FROM xxx_Plant_Device_Properties WHERE Value_Key='calc_plant_yield'""" cur.execute(sql) res = cur.fetchall() if len(res) == 0: sql = """INSERT INTO xxx_Plant_Device_Properties (Value_Key, Value) VALUES('calc_plant_yield', '1')""" try: cur.execute(sql) self._connection.commit() except Exception as err: self.logger.error( "ERROR - could not create plant device property: 'calc_plant_yield'") self.logger.error(str(e)) def __create_hidden_devs_table(self, cur): sql = """USE {_db}; IF NOT EXISTS (SELECT * FROM dbo.sysobjects where id = object_id(N'dbo.[xxx_Hidden_Devices]') and OBJECTPROPERTY(id, N'IsTable') = 1) BEGIN CREATE TABLE dbo.[xxx_Hidden_Devices] (sdid nvarchar(100), lid nvarchar(100) CONSTRAINT hidden_devices_pk PRIMARY KEY (sdid, lid) ) END""".format(_db=self.database) cur.execute(sql) def __get_hidden_devices(self, cur, lid): hidden_devs = [] sql = """SELECT sdid FROM xxx_Hidden_Devices WHERE lid=%s""" cur.execute(sql, (lid,)) res = cur.fetchall() for row in res: hidden_devs.append(row[0]) return hidden_devs def __filter_acp(self, dev, logger, cur): """This function filters power value which are 1.5 x nominal power of the inverter. It also filters values which are below 0 W. TODO: Discuss if small negative values should be filtered as well.""" self.logger.debug('__filter_acp()') if self.nominal_acp is None: self.nominal_acp = self.__get_all_nom_acp(logger, cur) device_id = dev.get('device_id') if device_id not in self.nominal_acp.keys(): return pac_vid = None for vid, vdict in dev['device_info']['values'].items(): if not vdict.get('value_type') == 'P_AC_TOTAL': continue pac_vid = vid if not pac_vid: return for ts in dev.get('data'): pac_val = dev['data'][ts].get(pac_vid) if pac_val is None: continue pac_val = float(pac_val) ac_limit = float(self.nominal_acp[device_id]) * 1.5 if pac_val is not None and pac_val > ac_limit: self.logger.warning( 'DataDB: Value P_AC_TOTAL exceeds nominal power * 1.5. Discarding value') dev['data'][ts].pop(pac_vid) if pac_val is not None and pac_val < 0: self.logger.warning( 'DataDB: Value P_AC_TOTAL has a negative value. Discarding value') dev['data'][ts].pop(pac_vid) def __filter_etotal(self, dev, logger, cur): self.logger.debug('__filter_etotal()') device_id = dev.get('device_id') if device_id not in self.nominal_acp.keys(): return etotal_vid = None for vid, vdict in dev['device_info']['values'].items(): if vdict.get('value_type') not in ['E_TOTAL_KWH', 'E_TOTAL']: continue etotal_vid = vid if not etotal_vid: return for ts in dev.get('data'): etotal_val = dev['data'][ts].get(etotal_vid) if etotal_val is None: continue try: etotal_val = long(etotal_val) except Exception: etotal_val = long(float(etotal_val)) """ the limit for etotal is calculated using following formula: nom_acp * 1000 * number of years (20) * 110% """ etotal_limit = ( int(float(self.nominal_acp[device_id]))) * 1000 * 20 * 110 / 100 if etotal_val is not None and long(etotal_val) > long(etotal_limit): self.logger.warning('DataDB: Value E_TOTAL exceeds the limit {_etotal_limit}. Discarding value {_val} for device {_dev_id}, value id: {_val_id}'.format(_etotal_limit=etotal_limit, _val=etotal_val, _dev_id=device_id, _val_id=etotal_vid)) dev['data'][ts].pop(etotal_vid) if etotal_val is not None and etotal_val < 0: self.logger.warning( 'DataDB: Value E_TOTAL has a negative value. Discarding value') dev['data'][ts].pop(etotal_vid) def __filter_generic_values(self, dev_data, dtype): self.logger.debug('__filter_generic_values()') if dtype == 'SENSOR': dtype = 'METEO' if dtype not in DataDB.correction_map.keys(): return for ts in dev_data.get('data'): for val, data_point in dev_data['data'][ts].items(): # cut the last number e.g P_DC_1 becomes P_DC_ if val in dev_data['device_info']['values'].keys(): val_vid = dev_data['device_info']['values'][val].get( 'value_type') if val_vid is None: continue else: continue cropped_val = val_vid if dtype == 'INVERTER': if val_vid.count('_') > 1 and val_vid[-1].isdigit(): cropped_val = val_vid[:5] cropped_val = cropped_val.rstrip('_KWH') if dtype == "STRINGBOX": if 'I_' in val_vid: cropped_val = 'I_' if DataDB.correction_map[dtype].get(cropped_val) is None: continue else: max_limit = DataDB.correction_map[dtype][cropped_val]['max'] min_limit = DataDB.correction_map[dtype][cropped_val]['min'] if data_point == 'NULL': continue data_point = float(data_point) if data_point > max_limit: self.logger.warning('DataDB: Measurement {val} exceeds the allowed maximum {max_limit} with {data_point}. Discarding value'.format(val=val_vid, max_limit=max_limit, data_point=data_point)) dev_data['data'][ts].pop(val) elif data_point < min_limit: self.logger.warning('DataDB: Measurement {val} is below the allowed minimum {min_limit} with {data_point}. Discarding value'.format(val=val_vid, min_limit=min_limit, data_point=data_point)) dev_data['data'][ts].pop(val) def __get_all_nom_acp(self, logger, cur): """ get installed ac_p for all inverters of the given logger""" sql = """use {db}; SELECT ISNULL(Fix_Value, Maximum_Value), Table_Name FROM xxx_Virtual_Devices WHERE Device_Class='P_AC' AND Device_Type='INVERTER' AND Table_Name like '%[_]{logger}[_]%'""".format(db=self.database, logger=logger) cur.execute(sql) res = cur.fetchall() dev_acp = {} for row in res: if row[0] is not None: dev_id = row[1].split(logger + '_')[1] dev_id = dev_id.replace('dash', '-') dev_id = dev_id.replace('slash', '/') dev_acp.update({dev_id: row[0]}) return dev_acp def __check_datatype_range(self, dev_id, dev_data): for ts in dev_data.get('data'): for val_id, data_point in dev_data['data'][ts].items(): if dev_data.get('device_info').get('values') is None: return if dev_data['device_info']['values'].get(val_id) is None: continue datatype = dev_data['device_info']['values'][val_id]['data_type'] if data_point is None or "NULL" in data_point: continue if 'nvarchar(200)' in datatype and len(data_point) > 200: self.logger.warning('DataDB: Measurement {vid} of device {dev_id} is out of range, value {dp}. Discarding value'.format(vid=val_id, dp=data_point, dev_id=dev_id)) dev_data['data'][ts].pop(val_id) return if datatype == 'integer': datatype = 'int' dt_range = DataDB.datatype_filter.get(datatype) if dt_range is None: return try: str_dp = data_point data_point = float(data_point) except Exception as e: self.logger.warning('DataDB: Measurement {vid} of device {dev_id} is out of range, value: {dp}. Discarding value'.format(vid=val_id, dp=data_point, dev_id=dev_id)) dev_data['data'][ts].pop(val_id) return if type(dt_range['min']) is list: if not ((dt_range['min'][0] <= data_point <= dt_range['min'][1] or dt_range['max'][0] <= data_point <= dt_range['max'][1] or data_point == 0) and len(str_dp) <= dt_range['precision']): self.logger.warning('DataDB: Measurement {vid} of device {dev_id} is out of range, value: {dp}. Discarding value'.format(vid=val_id, dp=data_point, dev_id=dev_id)) dev_data['data'][ts].pop(val_id) else: if not (dt_range['min'] <= data_point <= dt_range['max']): self.logger.warning('DataDB: Measurement {vid} of device {dev_id} is out of range, value: {dp}. Discarding value'.format(vid=val_id, dp=data_point, dev_id=dev_id)) dev_data['data'][ts].pop(val_id)