Wolfgang Hottgenroth c54b335e5f
All checks were successful
ci/woodpecker/push/woodpecker Pipeline was successful
queries and migrations
2024-01-03 20:57:26 +01:00

80 lines
4.6 KiB
Python

import psycopg2
from loguru import logger
import os
srcPgHost = os.environ["SRC_PGHOST"]
srcPgUser = os.environ["SRC_PGUSER"]
srcPgPassword = os.environ["SRC_PGPASSWORD"]
srcPgDatabase = os.environ["SRC_PGDATABASE"]
destPgHost = os.environ["DEST_PGHOST"]
destPgUser = os.environ["DEST_PGUSER"]
destPgPassword = os.environ["DEST_PGPASSWORD"]
destPgDatabase = os.environ["DEST_PGDATABASE"]
try:
srcConn = psycopg2.connect(
host=srcPgHost,
dbname=srcPgDatabase,
user=srcPgUser,
password=srcPgPassword,
sslmode='require'
)
srcConn.autocommit = False
destConn = psycopg2.connect(
host=destPgHost,
dbname=destPgDatabase,
user=destPgUser,
password=destPgPassword,
sslmode='require'
)
destConn.autocommit = False
with srcConn.cursor() as srcCur, destConn.cursor() as destCur:
srcCur.execute("select time, deviceid, status, state, importenergyactive, importenergyreactive, exportenergyactive, exportenergyreactive, powerapparent, poweractive, powerreactive, powerdemandpositive, powerdemandreverse, factor, angle, voltage, current, powerdemand from pv_power_measurement order by time")
for srcObj in srcCur:
timestamp = srcObj[0]
deviceName = srcObj[1]
status = srcObj[2]
state = srcObj[3]
importenergyactive = srcObj[4]
importenergyreactive = srcObj[5]
exportenergyactive = srcObj[6]
exportenergyreactive = srcObj[7]
powerapparent = srcObj[8]
poweractive = srcObj[9]
powerreactive = srcObj[10]
powerdemandpositive = srcObj[11]
powerdemandreverse = srcObj[12]
factor = srcObj[13]
angle = srcObj[14]
voltage = srcObj[15]
current = srcObj[16]
powerdemand = srcObj[17]
logger.info(f"{timestamp=}, {deviceName=}")
destTime = timestamp
destApplication = "PV"
destDevice = "Powermeter"
destAttributes = f"{\"ApplicationId\":\"PV\", \"Status\":\"{status}\",\"Hint\": \"Migrated\"}'
destValues = f"{\"Cnt\": {\"unit\": \"\", \"label\": \"\", \"value\": \"-1\", \"variable\": \"Cnt\"}, \"Angle\": {\"unit\": \"degree\", \"label\": \"\", \"value\": \"{angle}\", \"variable\": \"Angle\"}, \"State\": {\"unit\": \"\", \"label\": \"\", \"value\": \"{state}\", \"variable\": \"State\"}, \"Factor\": {\"unit\": \"\", \"label\": \"\", \"value\": \"{factor}\", \"variable\": \"Factor\"}, \"Current\": {\"unit\": \"A\", \"label\": \"\", \"value\": \"{current}\", \"variable\": \"Current\"}, \"Voltage\": {\"unit\": \"V\", \"label\": \"\", \"value\": \"{voltage}\", \"variable\": \"Voltage\"}, \"PowerActive\": {\"unit\": \"W\", \"label\": \"\", \"value\": \"{poweractive}\", \"variable\": \"PowerActive\"}, \"PowerApparent\": {\"unit\": \"VA\", \"label\": \"\", \"value\": \"{powerapparent}\", \"variable\": \"PowerApparent\"}, \"PowerReactive\": {\"unit\": \"VA\", \"label\": \"\", \"value\": \"{powerreactive}\", \"variable\": \"PowerReactive\"}, \"ExportEnergyActive\": {\"unit\": \"Wh\", \"label\": \"\", \"value\": \"{exportenergyactive}\", \"variable\": \"ExportEnergyActive\"}, \"ImportEnergyActive\": {\"unit\": \"Wh\", \"label\": \"\", \"value\": \"{importenergyactive}\", \"variable\": \"ImportEnergyActive\"}, \"PowerDemandReverse\": {\"unit\": \"W\", \"label\": \"\", \"value\": \"{powerdemandreverse}\", \"variable\": \"PowerDemandReverse\"}, \"PowerDemandPositive\": {\"unit\": \"W\", \"label\": \"\", \"value\": \"{powerdemandpositive}\", \"variable\": \"PowerDemandPositive\"}, \"ExportEnergyReactive\": {\"unit\": \"VAh\", \"label\": \"\", \"value\": \"{exportenergyreactive}\", \"variable\": \"ExportEnergyReactive\"}, \"ImportEnergyReactive\": {\"unit\": \"VAh\", \"label\": \"\", \"value\": \"{importenergyreactive}\", \"variable\": \"ImportEnergyReactive\"}}"
logger.info(f"{destTime=}, {destApplication=}, {destDevice=}, {destAttributes=}, {destValues=}")
try:
destCur.execute("insert into measurements (time, application, device, attributes, values) values(%s, %s, %s, %s, %s)",
(destTime, destApplication, destDevice, destAttributes, destValues))
destConn.commit()
except Exception as e:
destConn.rollback()
logger.error(f"Error {e} when inserted time {destTime}")
finally:
if srcConn:
srcConn.close()
if destConn:
destConn.close()