diff --git a/.github/workflows/mypy.yml b/.github/workflows/mypy.yml new file mode 100644 index 0000000..70b5a93 --- /dev/null +++ b/.github/workflows/mypy.yml @@ -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/ diff --git a/growattServer/base_api.py b/growattServer/base_api.py index cc0c7c1..cdf1a57 100644 --- a/growattServer/base_api.py +++ b/growattServer/base_api.py @@ -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) @@ -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() @@ -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() @@ -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() @@ -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", {}) @@ -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() @@ -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() @@ -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() @@ -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} diff --git a/growattServer/open_api_v1/__init__.py b/growattServer/open_api_v1/__init__.py index a3b7633..5980980 100644 --- a/growattServer/open_api_v1/__init__.py +++ b/growattServer/open_api_v1/__init__.py @@ -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. @@ -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. @@ -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") @@ -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 @@ -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. @@ -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") diff --git a/growattServer/open_api_v1/devices/min.py b/growattServer/open_api_v1/devices/min.py index b766fac..2c9c4b1 100644 --- a/growattServer/open_api_v1/devices/min.py +++ b/growattServer/open_api_v1/devices/min.py @@ -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 @@ -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, diff --git a/growattServer/open_api_v1/devices/sph.py b/growattServer/open_api_v1/devices/sph.py index 14459aa..4790079 100644 --- a/growattServer/open_api_v1/devices/sph.py +++ b/growattServer/open_api_v1/devices/sph.py @@ -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