
For the record, I had some programming lessons at university (Turbo Pascal, C++), and I am proficient enough to write Powershell and Python scripts. Is it the best? No. Could it be done better? Probably yes! Does it do the trick? Yes Ma'am!

Requirements : a Linux OS (eg a Pi), Python3, Cron and Panda (pip install panda).
Provided : Charge percentage, PV power, Grid power, total consumed power. The first in percentage, all power devices in kWh. Lastly, a text device is needed to hold data. It just needs to be there, no need to display it.
I wanted to to gather usage data only. Things like how much PV energy has been used etc. There is a lot more possible, but I did not need that. If you wish, take the script as an example and extend it to your needs.
A few quirks : The tank does not provide a counter for energy consumed. It has to be calculated and sent to Domoticz. Totals also have to be calculated. Each time I let the script update Domoticz, the current value is fetched, consumption is added and the lot is written back to Domoticz. The counter value holds an Epoch number which is the Linux interpretation of our date and time. The last record is saved in Domoticz, and used as a value again to find new records. Due to timing a record might be missed. This solves that issue.
How to use :
- Create a mixergy.py file and copy the contents into it. Location : home folder of the account (eg \home\pi)
- Make it executable (chmod +x)
- Fill in your details and Domoticz settings in the top of the script
- If you want to test : comment out the last few lines
- Create a cron job to execute the script. Mine runs every 5 mins but set it as you like
Code: Select all
#!/usr/bin/python3
# Import functions
import sys
import requests
import urllib.request
import json
from datetime import datetime
import pandas as pd #Panda needs to be installed with the "PIP install panda" commmand
import calendar
import time
# ============================= CHANGE VALUES BELOW TO REFLECT YOUR SETUP ==================================================
# Declare variables
username = 'username' # Your Mixergy username
password = '"password' # Your Mixergy password
serial_number = 'MX011111' # the serial number of your Mixergy tank
urldomo = "http://192.168.1.19:8080/json.htm?" # The address of your Domoticz server, change the IP address only
semicolumn = "\u003B"
# Domoticz Indexes for Devices
DomIndexCharge = 394 # Domoticz Index for the current charge (percentage device)
DomIndexPV = 395 # Domoticz Index for the PV power used (kWh electric + counter)
DomIndexGrid = 396 # Domoticz Index for the Grid power used (kWh electric + counter)
DomIndexTotal = 397 # Domoticz Index for the total power used (kWh electric + counter)
DomIndexCount = 398 # Domoticz Index for the date counter (text)
# ============================= CHANGE VALUES ABOVE TO REFLECT YOUR SETUP ==================================================
# Get login URL for the Mixergy portal
result = requests.get("https://www.mixergy.io/api/v2")
root_result = result.json()
account_url = root_result["_links"]["account"]["href"]
result = requests.get(account_url)
account_result = result.json()
login_url = account_result["_links"]["login"]["href"]
result = requests.post(login_url, json = {'username': username, 'password': password})
if result.status_code != 201:
print("Authentication failure. Check your credentials and try again!")
exit()
print("Authentication successful!")
login_result = result.json()
login_token = login_result["token"]
headers = {'Authorization': f'Bearer {login_token}'}
result = requests.get("https://www.mixergy.io/api/v2", headers=headers)
root_result = result.json()
tanks_url = root_result["_links"]["tanks"]["href"]
result = requests.get(tanks_url, headers=headers)
tanks_result = result.json()
tanks = tanks_result['_embedded']['tankList']
if serial_number == '':
for i, subjobj in enumerate(tanks):
print("** Found a tank with serial number", subjobj['serialNumber'])
exit()
for i, subjobj in enumerate(tanks):
if serial_number == subjobj['serialNumber']:
print("Found tanks serial number", subjobj['serialNumber'])
tank_url = subjobj["_links"]["self"]["href"]
firmware_version = subjobj["firmwareVersion"]
print("Tank Url:", tank_url)
print("Firmware:",firmware_version)
print("Fetching details...")
result = requests.get(tank_url, headers=headers)
tank_result = result.json()
# Get the JSON with alle the measurements
measurement_url = tank_result["_links"]["measurements"]["href"]
# Get last recorded timestamp from Domoticz. This timestamp is the timestamp of the last record at the Mixergy portal
url = urllib.request.urlopen(urldomo + "type=command¶m=getdevices&rid=" + str(DomIndexCount))
data = json.load(url)
Last_recorded_timestamp = data["result"][0]["Data"]
if Last_recorded_timestamp == "Hello World" : # Device has just been created and still has default value
Last_recorded_timestamp = calendar.timegm(time.gmtime()) # Prepopulate value with current date/time in GMT
else :
Last_recorded_timestamp = int(Last_recorded_timestamp)
print("Last timestamp in domoticz : ", Last_recorded_timestamp)
# URL to fetch the JSON data
response = requests.get(measurement_url, headers=headers) # Make a GET request to fetch the data
# Check if the request was successful
if response.status_code == 200:
try:
# Try to parse the response as JSON
data = response.json()
# Extract the content list from the data
records = data['content']
# Filter records that are newer than the last recorded timestamp
new_records = [record for record in records if record['recordedTime'] > Last_recorded_timestamp]
# Prepare the records, handle missing 'pvEnergy' and 'energy' fields, and calculate 'GridEnergy'
for record in new_records:
record['recordedTimeEpoch'] = (record['recordedTime'])
record['recordedTime'] = datetime.fromtimestamp(record['recordedTime'] / 1000).strftime('%Y-%m-%d %H:%M:%S')
record['receivedTime'] = datetime.fromtimestamp(record['receivedTime'] / 1000).strftime('%Y-%m-%d %H:%M:%S')
record['pvEnergy'] = round(record.get('pvEnergy', 0) / 60, 2) # Convert to Watt and round to 2 decimal places
record['energy'] = round(record.get('energy', 0) / 60, 2) # Convert to Watt and round to 2 decimal places
record['GridEnergy'] = round(record['energy'] - record['pvEnergy'], 2) # Calculate and round GridEnergy in Watt
# Convert the new records to a DataFrame
df = pd.DataFrame(new_records)
if not df.empty:
# Determine the amount of records in the dataset. Due to timing issues this could differ from what is expected.
# If there are eg 5 records, there is data for 5 minutes in the dataset. This is 1/12th part of an hour
# The amount of kWh will be 1/12th of the average Watts of the 5 records.
num_records = df.shape[0]
Poll_Frequency = (60 / num_records)
print ("There are", num_records, "records in the set. Poll frequency has been set to", Poll_Frequency, ".")
# Calculate the totals for 'pvEnergy', 'energy', and 'GridEnergy' and round to 2 decimal places
Current_PV_Energy = round(df['pvEnergy'].mean(), 2)
Current_PV_Energy_in_kWh = Current_PV_Energy / Poll_Frequency
Current_PV_Energy = str(Current_PV_Energy)
Current_Grid_Energy = round(df['GridEnergy'].mean(), 2)
Current_Grid_Energy_in_kWh = Current_Grid_Energy / Poll_Frequency
Current_Grid_Energy = str(Current_Grid_Energy)
Current_Combined_Energy = round(df['energy'].mean(), 2)
Current_Combined_Energy_in_kWh = Current_Combined_Energy / Poll_Frequency
Current_Combined_Energy = str(Current_Combined_Energy)
# Get the last value of 'charge'
Last_Charge_Value = df['charge'].iloc[-1]
# Get the new latest timstamp
New_Latest_Timestamp = df['recordedTimeEpoch'].iloc[-1]
print ("New latest timestamp : ", New_Latest_Timestamp)
except ValueError as e:
print("Error parsing JSON:", e)
print("Response content:", response.text[:500]) # Print the first 500 characters of the response
else:
print(f"Failed to retrieve data: {response.status_code}")
if not df.empty:
# Display the DataFrame with the new 'GridEnergy' column
print(df[['recordedTimeEpoch','recordedTime', 'receivedTime', 'pvEnergy', 'energy', 'GridEnergy']])
# Display the totals
print("\nCurrent avg pvEnergy (Watt):", Current_PV_Energy)
print("Current avg GridEnergy (Watt):", Current_Grid_Energy)
print("Current avg energy (Watt):", Current_Combined_Energy)
# Display the last 'charge' value
print("\nLast value of charge:", Last_Charge_Value)
# Get previous values for PV usage:
url = urllib.request.urlopen(urldomo + "type=command¶m=getdevices&rid=" + str(DomIndexPV))
data = json.load(url)
Total_PV_Energy = data["result"][0]["Data"]
Total_PV_Energy = float(Total_PV_Energy.replace (" kWh", ""))
if Total_PV_Energy == 0.0 :
print ("The value is 0.0, assuming this device is newly created. Setting value to 1kWh")
Total_PV_Energy = 1
Total_PV_Energy = (Total_PV_Energy*1000) + Current_PV_Energy_in_kWh
Total_PV_Energy = str(Total_PV_Energy)
print("Total PV from Domoticz : ", Total_PV_Energy)
# Get previous values for Grid usage:
url = urllib.request.urlopen(urldomo + "type=command¶m=getdevices&rid=" + str(DomIndexGrid))
data = json.load(url)
Total_Grid_Energy = data["result"][0]["Data"]
Total_Grid_Energy = float(Total_Grid_Energy.replace (" kWh", ""))
if Total_Grid_Energy == 0.0 :
print ("The value is 0.0, assuming this device is newly created. Setting value to 1kWh")
Total_Grid_Energy = 1
Total_Grid_Energy = (Total_Grid_Energy*1000) + Current_Grid_Energy_in_kWh
Total_Grid_Energy = str(Total_Grid_Energy)
print("Total Grid from Domoticz : ",Total_Grid_Energy)
# Get previous values for total usage:
url = urllib.request.urlopen(urldomo + "type=command¶m=getdevices&rid=" + str(DomIndexTotal))
data = json.load(url)
Total_Energy = data["result"][0]["Data"]
Total_Energy = float(Total_Energy.replace (" kWh", ""))
if Total_Energy == 0.0 :
print ("The value is 0.0, assuming this device is newly created. Setting value to 1kWh")
Total_Energy = 1
Total_Energy = (Total_Energy*1000) + Current_Combined_Energy_in_kWh
Total_Energy = str(Total_Energy)
print("Total power from Domoticz : ",Total_Energy)
# Insert new PV usage into Domoticz
New_Total_PV_Energy = (Current_PV_Energy + semicolumn + Total_PV_Energy)
getVars1 = {'type' : 'command', 'param' : 'udevice', 'nvalue' : 0, 'idx': DomIndexPV, 'svalue': (New_Total_PV_Energy)}
# Insert new Grid usage into Domoticz
New_Total_Grid_Energy = (Current_Grid_Energy + semicolumn + Total_Grid_Energy)
getVars2 = {'type' : 'command', 'param' : 'udevice', 'nvalue' : 0, 'idx': DomIndexGrid, 'svalue': (New_Total_Grid_Energy)}
# Insert new total usage into Domoticz:
New_Total_Energy = (Current_Combined_Energy + semicolumn + Total_Energy)
getVars3 = {'type' : 'command', 'param' : 'udevice', 'nvalue' : 0, 'idx': DomIndexTotal, 'svalue': (New_Total_Energy)}
# insert current Charge into Domoticz:
getVars4 = {'type' : 'command', 'param' : 'udevice', 'nvalue' : 0, 'idx': (DomIndexCharge), 'svalue': (Last_Charge_Value)}
# Store latest timestamp in Domoticz
getVars5 = {'type' : 'command', 'param' : 'udevice', 'nvalue' : 0, 'idx': DomIndexCount, 'svalue': (New_Latest_Timestamp)}
# insert everything into Domoticz
webUrl = urllib.request.urlopen(urldomo + urllib.parse.urlencode(getVars1))
webUrl = urllib.request.urlopen(urldomo + urllib.parse.urlencode(getVars2))
webUrl = urllib.request.urlopen(urldomo + urllib.parse.urlencode(getVars3))
webUrl = urllib.request.urlopen(urldomo + urllib.parse.urlencode(getVars4))
webUrl = urllib.request.urlopen(urldomo + urllib.parse.urlencode(getVars5))
