Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Expose storage API #13

Open
wants to merge 11 commits into
base: master
Choose a base branch
from
Prev Previous commit
Next Next commit
Add Spa calls. Move Storage and Spa to their own classes.
  • Loading branch information
Sjord committed Sep 21, 2020
commit 834e00003c9330782219f1ba0087fc9bce5c47da
174 changes: 129 additions & 45 deletions growatt/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,28 @@ def hash_password(password):
return password_md5


def _back_success_response(response):
"""
Check and return the response, where we expect a "back" key with a
"success" item.
"""
if response.status_code != 200:
raise GrowattApiError("Request failed: %s" % response)
data = response.json()
result = data["back"]
if not "success" in result or not result["success"]:
raise GrowattApiError(result)
return result

def _obj_success_response(response):
if response.status_code != 200:
raise GrowattApiError("Request failed: %s" % response)
data = response.json()
if data["result"] != 1 or "obj" not in data:
raise GrowattApiError(data)
return data["obj"]


class Timespan(IntEnum):
day = 1
month = 2
Expand Down Expand Up @@ -68,7 +90,7 @@ def login(self, username, password):
data={"userName": username, "password": password_md5},
)
try:
result = self._back_success_response(response)
result = _back_success_response(response)
self.logged_in = True
return result
except GrowattApiError:
Expand All @@ -81,7 +103,7 @@ def plant_list(self):
response = self.session.get(
self.get_url("PlantListAPI.do"), allow_redirects=False
)
return self._back_success_response(response)
return _back_success_response(response)

def plant_detail(self, plant_id, timespan, date):
"""
Expand All @@ -98,7 +120,7 @@ def plant_detail(self, plant_id, timespan, date):
self.get_url("PlantDetailAPI.do"),
params={"plantId": plant_id, "type": timespan.value, "date": date_str},
)
return self._back_success_response(response)
return _back_success_response(response)

def new_plant_detail(self, plant_id, timespan, date):
"""
Expand All @@ -115,7 +137,7 @@ def new_plant_detail(self, plant_id, timespan, date):
self.get_url("newPlantDetailAPI.do"),
params={"plantId": plant_id, "type": timespan.value, "date": date_str},
)
return self._back_success_response(response)
return _back_success_response(response)

def get_user_center_energy_data(self):
"""
Expand Down Expand Up @@ -143,84 +165,146 @@ def get_all_device_list(self, plant_id):
)
return response.json()

def get_energy_prod_and_cons_data(self, plant_id, storage_sn, date, type=0):
def logout(self):
self.session.get(self.get_url("logout.do"))
self.logged_in = False

def storage(self, plant_id, device_sn):
return Storage(self, plant_id, device_sn)

def spa(self, plant_id, device_sn):
return Spa(self, plant_id, device_sn)


class Storage:
def __init__(self, api, plant_id, device_sn):
self.plant_id = plant_id
self.device_sn = device_sn
self.api = api
self.session = api.session

def get_url(self, page):
return self.api.get_url(page)

def get_energy_prod_and_cons_data(self, date, type=0):
response = self.session.post(
self.get_url("newStorageAPI.do"),
params={"op": "getEnergyProdAndConsData"},
data={
"plantId": plant_id,
"storageSn": storage_sn,
"plantId": self.plant_id,
"storageSn": self.device_sn,
"date": date,
"type": type,
}
)
return self._obj_success_response(response)
return _obj_success_response(response)

def get_storage_energy_data(self, plant_id, storage_sn, date):
def get_storage_energy_data(self, date):
response = self.session.post(
self.get_url("newStorageAPI.do"),
params={"op": "getStorageEnergyData"},
data={
"plantId": plant_id,
"storageSn": storage_sn,
"plantId": self.plant_id,
"storageSn": self.device_sn,
"date": date,
}
)
return self._obj_success_response(response)
return _obj_success_response(response)

def get_system_status_data(self, plant_id, storage_sn):
def get_system_status_data(self):
response = self.session.post(
self.get_url("newStorageAPI.do"),
params={"op": "getSystemStatusData"},
data={
"plantId": plant_id,
"storageSn": storage_sn,
"plantId": self.plant_id,
"storageSn": self.device_sn,
}
)
return self._obj_success_response(response)
return _obj_success_response(response)

def get_energy_overview_data(self, plant_id, storage_sn):
def get_energy_overview_data(self):
response = self.session.post(
self.get_url("newStorageAPI.do"),
params={"op": "getEnergyOverviewData"},
data={
"plantId": plant_id,
"storageSn": storage_sn,
"plantId": self.plant_id,
"storageSn": self.device_sn,
}
)
return self._obj_success_response(response)
return _obj_success_response(response)

def get_storage_params(self, storage_sn):
def get_storage_params(self):
response = self.session.get(
self.get_url("newStorageAPI.do"),
params={
"op": "getStorageParams",
"storageId": storage_sn,
"storageId": self.device_sn,
}
)
return response.json()["storageDetailBean"]

def logout(self):
self.session.get(self.get_url("logout.do"))
self.logged_in = False

def _back_success_response(self, response):
"""
Check and return the response, where we expect a "back" key with a
"success" item.
"""
if response.status_code != 200:
raise GrowattApiError("Request failed: %s" % response)
data = response.json()
result = data["back"]
if not "success" in result or not result["success"]:
raise GrowattApiError(result)
return result

def _obj_success_response(self, response):
if response.status_code != 200:
raise GrowattApiError("Request failed: %s" % response)
data = response.json()
if data["result"] != 1 or "obj" not in data:
raise GrowattApiError(data)
return data["obj"]
class Spa:
def __init__(self, api, plant_id, device_sn):
self.plant_id = plant_id
self.device_sn = device_sn
self.api = api
self.session = api.session

def get_url(self, page):
return self.api.get_url(page)

def get_system_status(self):
response = self.session.post(
self.get_url("newSpaApi.do"),
params={
"op": "getSystemStatus",
},
data={
"plantId": self.plant_id,
"spaId": self.device_sn,
}
)
return _obj_success_response(response)

def get_spa_energy(self, date):
response = self.session.post(
self.get_url("newSpaApi.do"),
params={
"op": "getSpaEnergy",
},
data={
"plantId": self.plant_id,
"spaId": self.device_sn,
"date": date,
}
)
return _obj_success_response(response)

def get_energy_overview(self):
response = self.session.post(
self.get_url("newSpaApi.do"),
params={
"op": "getEnergyOverview",
},
data={
"plantId": self.plant_id,
"spaId": self.device_sn,
}
)
return _obj_success_response(response)

def get_energy_prod_and_cons_data(self, date, timespan):
response = self.session.post(
self.get_url("newSpaApi.do"),
params={
"op": "getEnergyProdAndConsData",
},
data={
"plantId": self.plant_id,
"spaId": self.device_sn,
"date": date,
"type": timespan.value
}
)
return _obj_success_response(response)