Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 15 additions & 0 deletions .github/workflows/mypy.yml
Original file line number Diff line number Diff line change
@@ -0,0 +1,15 @@
name: mypy
on:
- push
- pull_request

jobs:
mypy:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: "3.12"
- run: pip install mypy types-requests
- run: mypy --ignore-missing-imports growattServer/
63 changes: 35 additions & 28 deletions growattServer/base_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -76,7 +76,7 @@ def _raise_for_status(response, *args: object, **kwargs: object) -> None:
_ = kwargs
response.raise_for_status()

self.session.hooks = {"response": _raise_for_status}
self.session.hooks = {"response": [_raise_for_status]}

headers = {"User-Agent": self.agent_identifier}
self.session.headers.update(headers)
Expand Down Expand Up @@ -266,12 +266,13 @@ def inverter_data(self, inverter_id: str, date: datetime.datetime | None = None)

"""
date_str = self.__get_date_string(date=date)
response = self.session.get(self.get_url("newInverterAPI.do"), params={
params: dict[str, str | int] = {
"op": "getInverterData",
"id": inverter_id,
"type": 1,
"date": date_str
})
"date": date_str,
}
response = self.session.get(self.get_url("newInverterAPI.do"), params=params)

return response.json()

Expand Down Expand Up @@ -412,12 +413,13 @@ def tlx_data(self, tlx_id: str, date: datetime.datetime | None = None) -> dict[s

"""
date_str = self.__get_date_string(date=date)
response = self.session.get(self.get_url("newTlxApi.do"), params={
params: dict[str, str | int] = {
"op": "getTlxData",
"id": tlx_id,
"type": 1,
"date": date_str
})
"date": date_str,
}
response = self.session.get(self.get_url("newTlxApi.do"), params=params)

return response.json()

Expand Down Expand Up @@ -753,10 +755,10 @@ def get_mix_inverter_settings(self, serial_number: str) -> dict[str, Any]:
dict: A dictionary of settings.

"""
default_params = {
default_params: dict[str, str | int] = {
"op": "getMixSetParams",
"serialNum": serial_number,
"kind": 0
"kind": 0,
}
response = self.session.get(self.get_url("newMixApi.do"), params=default_params)
return response.json()
Expand Down Expand Up @@ -870,10 +872,12 @@ def inverter_list(self, plant_id: str) -> list[dict[str, Any]]:

def __get_all_devices(self, plant_id: str) -> dict[str, Any]:
"""Get basic plant information with device list."""
response = self.session.get(self.get_url("newTwoPlantAPI.do"),
params={"op": "getAllDeviceList",
"plantId": plant_id,
"language": 1})
params: dict[str, str | int] = {
"op": "getAllDeviceList",
"plantId": plant_id,
"language": 1,
}
response = self.session.get(self.get_url("newTwoPlantAPI.do"), params=params)

return response.json().get("deviceList", {})

Expand All @@ -889,12 +893,13 @@ def device_list(self, plant_id: str) -> list[dict[str, Any]]:

def plant_info(self, plant_id: str) -> dict[str, Any]:
"""Get basic plant information with device list."""
response = self.session.get(self.get_url("newTwoPlantAPI.do"), params={
params: dict[str, str | int] = {
"op": "getAllDeviceListTwo",
"plantId": plant_id,
"pageNum": 1,
"pageSize": 1
})
"pageSize": 1,
}
response = self.session.get(self.get_url("newTwoPlantAPI.do"), params=params)

return response.json()

Expand Down Expand Up @@ -1093,18 +1098,19 @@ def update_inverter_setting(self, serial_number: str, setting_type: str,
# Ensure declared but unused args are referenced to satisfy linters
_ = serial_number
_ = setting_type
settings_parameters = parameters

# If we've been passed an array then convert it into a dictionary
if isinstance(parameters, list):
settings_parameters = {}
settings_parameters: dict[str, Any] = {}
for index, param in enumerate(parameters, start=1):
settings_parameters["param" + str(index)] = param
else:
settings_parameters = parameters

settings_parameters = {**default_parameters, **settings_parameters}
merged = {**default_parameters, **settings_parameters}

response = self.session.post(self.get_url("newTcpsetAPI.do"),
params=settings_parameters)
params=merged)

return response.json()

Expand Down Expand Up @@ -1234,20 +1240,21 @@ def update_noah_settings(self, serial_number: str, setting_type: str, parameters
"""
default_parameters = {
"serialNum": serial_number,
"type": setting_type
"type": setting_type,
}
settings_parameters = parameters

# If we've been passed an array then convert it into a dictionary
if isinstance(parameters, list):
settings_parameters = {}
settings_parameters: dict[str, Any] = {}
for index, param in enumerate(parameters, start=1):
settings_parameters["param" + str(index)] = param
else:
settings_parameters = parameters

settings_parameters = {**default_parameters, **settings_parameters}
merged = {**default_parameters, **settings_parameters}

response = self.session.post(self.get_url("noahDeviceApi/noah/set"),
data=settings_parameters)
data=merged)

return response.json()

Expand Down Expand Up @@ -1391,13 +1398,13 @@ def update_classic_inverter_setting(self, default_parameters: dict[str, Any], pa
dict: Server JSON response.

"""
settings_parameters = parameters

# If we've been passed an array then convert it into a dictionary
if isinstance(parameters, list):
settings_parameters = {}
settings_parameters: dict[str, Any] = {}
for index, param in enumerate(parameters, start=1):
settings_parameters["param" + str(index)] = param
else:
settings_parameters = parameters

settings_parameters = {**default_parameters, **settings_parameters}

Expand Down
36 changes: 18 additions & 18 deletions growattServer/open_api_v1/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,7 @@ def __init__(self, token: str) -> None:
# Set up authentication for V1 API using the provided token
self.session.headers.update({"token": token})

def process_response(self, response: dict[str, Any], operation_name: str = "API operation") -> dict[str, Any]:
def process_response(self, response: dict[str, Any], operation_name: str = "API operation") -> Any:
"""
Process API response and handle errors.

Expand Down Expand Up @@ -90,7 +90,7 @@ def get_url(self, page: str) -> str:
"""Return the page URL for the v1 API."""
return self.api_url + page

def plant_list(self) -> dict[str, Any]:
def plant_list(self) -> dict[str, Any]: # type: ignore[override]
"""
Get a list of all plants with detailed information.

Expand Down Expand Up @@ -211,12 +211,13 @@ def plant_power_overview(
if day is None:
day = datetime.now(tz=UTC).astimezone().date()

params: dict[str, str | int] = {
"plant_id": plant_id,
"date": str(day),
}
response = self.session.get(
self.get_url("plant/power"),
params={
"plant_id": plant_id,
"date": day,
},
params=params,
)

return self.process_response(response.json(), "getting plant power overview")
Expand Down Expand Up @@ -265,12 +266,10 @@ def plant_energy_history(
max_day_interval = 7
max_year_interval = 20

if start_date is None and end_date is None:
start_date = datetime.now(tz=UTC).astimezone().date()
end_date = datetime.now(tz=UTC).astimezone().date()
elif start_date is None:
start_date = end_date
elif end_date is None:
today = datetime.now(tz=UTC).astimezone().date()
if start_date is None:
start_date = end_date if end_date is not None else today
if end_date is None:
end_date = start_date

# Validate date ranges based on time_unit
Expand Down Expand Up @@ -309,7 +308,7 @@ def plant_energy_history(

return self.process_response(response.json(), "getting plant energy history")

def device_list(self, plant_id: int) -> dict[str, Any]:
def device_list(self, plant_id: int) -> dict[str, Any]: # type: ignore[override]
"""
Get devices associated with plant.

Expand Down Expand Up @@ -354,13 +353,14 @@ def device_list(self, plant_id: int) -> dict[str, Any]:
}

"""
params: dict[str, str | int] = {
"plant_id": plant_id,
"page": "",
"perpage": "",
}
response = self.session.get(
url=self.get_url("device/list"),
params={
"plant_id": plant_id,
"page": "",
"perpage": "",
},
params=params,
)
return self.process_response(response.json(), "getting device list")

Expand Down
12 changes: 5 additions & 7 deletions growattServer/open_api_v1/devices/min.py
Original file line number Diff line number Diff line change
Expand Up @@ -101,12 +101,10 @@ def energy_history(
https://www.showdoc.com.cn/262556420217021/6129764475556048

"""
if start_date is None and end_date is None:
start_date = datetime.now(tz=UTC).astimezone().date()
end_date = datetime.now(tz=UTC).astimezone().date()
elif start_date is None:
start_date = end_date
elif end_date is None:
today = datetime.now(tz=UTC).astimezone().date()
if start_date is None:
start_date = end_date if end_date is not None else today
if end_date is None:
end_date = start_date

# check interval validity
Expand Down Expand Up @@ -451,7 +449,7 @@ def read_time_segments(self, settings_data: dict[str, Any] | None = None) -> lis
segment = {
"segment_id": i,
"batt_mode": batt_mode,
"mode_name": mode_names.get(batt_mode, "Unknown"),
"mode_name": mode_names.get(batt_mode, "Unknown") if batt_mode is not None else "Unknown",
"start_time": start_time,
"end_time": end_time,
"enabled": enabled,
Expand Down
10 changes: 4 additions & 6 deletions growattServer/open_api_v1/devices/sph.py
Original file line number Diff line number Diff line change
Expand Up @@ -100,12 +100,10 @@ def energy_history(
https://www.showdoc.com.cn/262556420217021/6129765461123058

"""
if start_date is None and end_date is None:
start_date = datetime.now(tz=UTC).astimezone().date()
end_date = datetime.now(tz=UTC).astimezone().date()
elif start_date is None:
start_date = end_date
elif end_date is None:
today = datetime.now(tz=UTC).astimezone().date()
if start_date is None:
start_date = end_date if end_date is not None else today
if end_date is None:
end_date = start_date

# check interval validity
Expand Down
Loading