From 6d086a1152f278ea1c2fa80fe529f5e1dacd01d8 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 16 Jul 2026 20:09:53 +0000 Subject: [PATCH] feat: add site-data API endpoints Adds the SolarEdge Monitoring API "Site Data" endpoints that were not yet wrapped by the client: - get_sites -> /sites/list - get_data_period -> /site/{id}/dataPeriod - get_data_period_bulk -> /sites/{ids}/dataPeriod - get_energy -> /site/{id}/energy - get_energy_bulk -> /sites/{ids}/energy - get_time_frame_energy[_bulk] -> /site|sites/.../timeFrameEnergy - get_power -> /site/{id}/power - get_power_bulk -> /sites/{ids}/power - get_overview_bulk -> /sites/{ids}/overview - get_power_details -> /site/{id}/powerDetails - get_environmental_benefits -> /site/{id}/envBenefits Introduces shared infrastructure reused by later endpoint PRs: the Meter/TimeUnit/SystemUnits/SortOrder Literal aliases, the _format_date/_format_datetime/_join_ids helpers (so methods accept date, datetime or pre-formatted str), and the _get_sites_url builder for bulk calls. Tests use aiointercept (matching main's test stack) with lookahead- based regex so query-parameter ordering is not assumed. One PR in a series splitting the full endpoint coverage work into reviewable pieces. Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- src/aiosolaredge/solaredge.py | 288 +++++++++++++++++++++++++++++++++- tests/test_init.py | 200 +++++++++++++++++++++++ 2 files changed, 487 insertions(+), 1 deletion(-) diff --git a/src/aiosolaredge/solaredge.py b/src/aiosolaredge/solaredge.py index 2d8d9bd..9162055 100644 --- a/src/aiosolaredge/solaredge.py +++ b/src/aiosolaredge/solaredge.py @@ -1,7 +1,7 @@ from __future__ import annotations import logging -from datetime import datetime +from datetime import date, datetime from typing import Any, Iterable, Literal import aiohttp @@ -9,9 +9,34 @@ _BASE_URL = yarl.URL("https://monitoringapi.solaredge.com") _DATETIME_FORMAT = "%Y-%m-%d %H:%M:%S" +_DATE_FORMAT = "%Y-%m-%d" _LOGGER = logging.getLogger(__name__) +Meter = Literal["PRODUCTION", "CONSUMPTION", "SELFCONSUMPTION", "FEEDIN", "PURCHASED"] +TimeUnit = Literal["QUARTER_OF_AN_HOUR", "HOUR", "DAY", "WEEK", "MONTH", "YEAR"] +SystemUnits = Literal["Metrics", "Imperial"] +SortOrder = Literal["ASC", "DESC"] + + +def _format_date(value: date | datetime | str) -> str: + """Format a date value for the SolarEdge API (YYYY-MM-DD).""" + if isinstance(value, str): + return value + return value.strftime(_DATE_FORMAT) + + +def _format_datetime(value: datetime | str) -> str: + """Format a datetime value for the SolarEdge API (YYYY-MM-DD hh:mm:ss).""" + if isinstance(value, str): + return value + return value.strftime(_DATETIME_FORMAT) + + +def _join_ids(site_ids: Iterable[int | str]) -> str: + """Join site IDs with a comma for bulk API calls.""" + return ",".join(str(site_id) for site_id in site_ids) + class SolarEdge: """SolarEdge API client.""" @@ -37,6 +62,10 @@ def _get_site_url(self, site_id: int | str) -> yarl.URL: """Get the site URL.""" return _BASE_URL.joinpath("site", str(site_id)) + def _get_sites_url(self, site_ids: Iterable[int | str]) -> yarl.URL: + """Get the bulk sites URL.""" + return _BASE_URL.joinpath("sites", _join_ids(site_ids)) + async def get_details(self, site_id: int | str) -> dict[str, Any]: """ Get details of the SolarEdge system. @@ -137,6 +166,263 @@ async def get_current_power_flow(self, site_id: int | str) -> dict[str, Any]: self._get_site_url(site_id).joinpath("currentPowerFlow") ) + async def get_sites( + self, + size: int | None = None, + start_index: int | None = None, + search_text: str | None = None, + sort_property: str | None = None, + sort_order: SortOrder | None = None, + status: str | None = None, + ) -> dict[str, Any]: + """ + Get the list of sites for the given account API key. + + :param size: Maximum number of sites to return (default 100, max 100). + :param start_index: First site index to be returned in the results. + :param search_text: Search text for sites + (Name, Notes, Address, City, Zip, Full address, Country). + :param sort_property: Sorting option for the site list (e.g. "Name", + "Country", "Status", "PeakPower", "InstallationDate", etc.). + :param sort_order: Sort order: "ASC" or "DESC" (default "ASC"). + :param status: Filter sites by status. A comma-separated combination + of "Active", "Pending", "Disabled" or "All". + :return: The list of sites. + """ + params: dict[str, Any] = {} + if size is not None: + params["size"] = size + if start_index is not None: + params["startIndex"] = start_index + if search_text is not None: + params["searchText"] = search_text + if sort_property is not None: + params["sortProperty"] = sort_property + if sort_order is not None: + params["sortOrder"] = sort_order + if status is not None: + params["status"] = status + return await self._get_json(_BASE_URL.joinpath("sites", "list"), params=params) + + async def get_data_period(self, site_id: int | str) -> dict[str, Any]: + """ + Get the energy production start and end dates for the site. + + :param site_id: The site ID. + :return: The data period. + """ + return await self._get_json(self._get_site_url(site_id).joinpath("dataPeriod")) + + async def get_data_period_bulk( + self, site_ids: Iterable[int | str] + ) -> dict[str, Any]: + """ + Get the energy production start and end dates for multiple sites. + + :param site_ids: An iterable of site IDs (up to 100). + :return: The data period for each site. + """ + return await self._get_json( + self._get_sites_url(site_ids).joinpath("dataPeriod") + ) + + async def get_energy( + self, + site_id: int | str, + start_date: date | datetime | str, + end_date: date | datetime | str, + time_unit: TimeUnit = "DAY", + ) -> dict[str, Any]: + """ + Get site energy measurements. + + :param site_id: The site ID. + :param start_date: The start date. + :param end_date: The end date. + :param time_unit: Aggregation granularity. Default "DAY". + Allowed values: "QUARTER_OF_AN_HOUR", "HOUR", "DAY", + "WEEK", "MONTH", "YEAR". + :return: Site energy measurements. + """ + params = { + "startDate": _format_date(start_date), + "endDate": _format_date(end_date), + "timeUnit": time_unit, + } + return await self._get_json( + self._get_site_url(site_id).joinpath("energy"), params=params + ) + + async def get_energy_bulk( + self, + site_ids: Iterable[int | str], + start_date: date | datetime | str, + end_date: date | datetime | str, + time_unit: TimeUnit = "DAY", + ) -> dict[str, Any]: + """ + Get site energy measurements for multiple sites. + + :param site_ids: An iterable of site IDs (up to 100). + :param start_date: The start date. + :param end_date: The end date. + :param time_unit: Aggregation granularity. Default "DAY". + :return: Site energy measurements per site. + """ + params = { + "startDate": _format_date(start_date), + "endDate": _format_date(end_date), + "timeUnit": time_unit, + } + return await self._get_json( + self._get_sites_url(site_ids).joinpath("energy"), params=params + ) + + async def get_time_frame_energy( + self, + site_id: int | str, + start_date: date | datetime | str, + end_date: date | datetime | str, + ) -> dict[str, Any]: + """ + Get the site total energy produced for a given time period. + + :param site_id: The site ID. + :param start_date: The start date. + :param end_date: The end date. + :return: The total energy for the period. + """ + params = { + "startDate": _format_date(start_date), + "endDate": _format_date(end_date), + } + return await self._get_json( + self._get_site_url(site_id).joinpath("timeFrameEnergy"), + params=params, + ) + + async def get_time_frame_energy_bulk( + self, + site_ids: Iterable[int | str], + start_date: date | datetime | str, + end_date: date | datetime | str, + ) -> dict[str, Any]: + """ + Get the total energy produced for a given time period for multiple sites. + + :param site_ids: An iterable of site IDs (up to 100). + :param start_date: The start date. + :param end_date: The end date. + :return: The total energy per site. + """ + params = { + "startDate": _format_date(start_date), + "endDate": _format_date(end_date), + } + return await self._get_json( + self._get_sites_url(site_ids).joinpath("timeFrameEnergy"), + params=params, + ) + + async def get_power( + self, + site_id: int | str, + start_time: datetime | str, + end_time: datetime | str, + ) -> dict[str, Any]: + """ + Get site power measurements at 15-minute resolution. + + :param site_id: The site ID. + :param start_time: The start time. + :param end_time: The end time. + :return: Site power measurements. + """ + params = { + "startTime": _format_datetime(start_time), + "endTime": _format_datetime(end_time), + } + return await self._get_json( + self._get_site_url(site_id).joinpath("power"), params=params + ) + + async def get_power_bulk( + self, + site_ids: Iterable[int | str], + start_time: datetime | str, + end_time: datetime | str, + ) -> dict[str, Any]: + """ + Get power measurements at 15-minute resolution for multiple sites. + + :param site_ids: An iterable of site IDs (up to 100). + :param start_time: The start time. + :param end_time: The end time. + :return: Site power measurements per site. + """ + params = { + "startTime": _format_datetime(start_time), + "endTime": _format_datetime(end_time), + } + return await self._get_json( + self._get_sites_url(site_ids).joinpath("power"), params=params + ) + + async def get_overview_bulk(self, site_ids: Iterable[int | str]) -> dict[str, Any]: + """ + Get overview data for multiple sites. + + :param site_ids: An iterable of site IDs (up to 100). + :return: The overview per site. + """ + return await self._get_json(self._get_sites_url(site_ids).joinpath("overview")) + + async def get_power_details( + self, + site_id: int | str, + start_time: datetime | str, + end_time: datetime | str, + meters: Iterable[Meter] = (), + ) -> dict[str, Any]: + """ + Get detailed site power measurements from meters. + + :param site_id: The site ID. + :param start_time: The start time. + :param end_time: The end time. + :param meters: Optional iterable of meter types + (PRODUCTION, CONSUMPTION, SELFCONSUMPTION, FEEDIN, PURCHASED). + :return: Detailed site power measurements. + """ + params: dict[str, Any] = { + "startTime": _format_datetime(start_time), + "endTime": _format_datetime(end_time), + } + if meters: + params["meters"] = ",".join(meters) + return await self._get_json( + self._get_site_url(site_id).joinpath("powerDetails"), params=params + ) + + async def get_environmental_benefits( + self, + site_id: int | str, + system_units: SystemUnits | None = None, + ) -> dict[str, Any]: + """ + Get environmental benefits based on site energy production. + + :param site_id: The site ID. + :param system_units: Optional unit system: "Metrics" or "Imperial". + :return: Environmental benefits (CO2 saved, trees planted, etc.). + """ + params: dict[str, Any] = {} + if system_units is not None: + params["systemUnits"] = system_units + return await self._get_json( + self._get_site_url(site_id).joinpath("envBenefits"), params=params + ) + async def _get_json( self, url: yarl.URL, params: dict[str, Any] | None = None ) -> dict[str, Any]: diff --git a/tests/test_init.py b/tests/test_init.py index e523d46..13e0845 100644 --- a/tests/test_init.py +++ b/tests/test_init.py @@ -110,3 +110,203 @@ async def test_simple_requests() -> None: serials=["SN1", "SN2"], ) == {"storageData": {"batteryCount": 1, "batteries": []}} await solar_edge.close() + + +@pytest.mark.asyncio +async def test_get_sites() -> None: + """Test getting the list of sites with all options.""" + async with aiointercept(mock_external_urls=True) as mocked: + solar_edge = SolarEdge("API_KEY") + mocked.get( + "https://monitoringapi.solaredge.com/sites/list?api_key=API_KEY", + payload={"sites": "sites"}, + ) + assert await solar_edge.get_sites() == {"sites": "sites"} + + pattern = re.compile( + r"^https://monitoringapi\.solaredge\.com/sites/list\?" + r"(?=.*api_key=API_KEY)(?=.*size=5)(?=.*startIndex=10)(?=.*searchText=Lyon)" + r"(?=.*sortProperty=Name)(?=.*sortOrder=DESC)(?=.*status=Active).*$" + ) + mocked.get(pattern, payload={"sites": "filtered"}) + assert await solar_edge.get_sites( + size=5, + start_index=10, + search_text="Lyon", + sort_property="Name", + sort_order="DESC", + status="Active", + ) == {"sites": "filtered"} + await solar_edge.close() + + +@pytest.mark.asyncio +async def test_get_data_period() -> None: + """Test getting data period for a single site and bulk.""" + async with aiointercept(mock_external_urls=True) as mocked: + solar_edge = SolarEdge("API_KEY") + mocked.get( + "https://monitoringapi.solaredge.com/site/123/dataPeriod?api_key=API_KEY", + payload={"dataPeriod": "dataPeriod"}, + ) + assert await solar_edge.get_data_period(123) == {"dataPeriod": "dataPeriod"} + + mocked.get( + "https://monitoringapi.solaredge.com/sites/1,4/dataPeriod?api_key=API_KEY", + payload={"dataPeriod": "bulk"}, + ) + assert await solar_edge.get_data_period_bulk([1, 4]) == {"dataPeriod": "bulk"} + await solar_edge.close() + + +@pytest.mark.asyncio +async def test_get_energy() -> None: + """Test getting energy and bulk energy.""" + async with aiointercept(mock_external_urls=True) as mocked: + solar_edge = SolarEdge("API_KEY") + pattern = re.compile( + r"^https://monitoringapi\.solaredge\.com/site/123/energy\?" + r"(?=.*startDate=2013-05-01)(?=.*endDate=2013-05-30)(?=.*timeUnit=DAY).*$" + ) + mocked.get(pattern, payload={"energy": "energy"}) + assert await solar_edge.get_energy( + 123, + datetime.date(2013, 5, 1), + datetime.date(2013, 5, 30), + ) == {"energy": "energy"} + + pattern = re.compile( + r"^https://monitoringapi\.solaredge\.com/site/123/energy\?" + r"(?=.*startDate=2013-05-01)(?=.*endDate=2013-05-30)(?=.*timeUnit=HOUR).*$" + ) + mocked.get(pattern, payload={"energy": "energy_str"}) + assert await solar_edge.get_energy( + 123, "2013-05-01", "2013-05-30", time_unit="HOUR" + ) == {"energy": "energy_str"} + + pattern = re.compile( + r"^https://monitoringapi\.solaredge\.com/sites/1,4/energy\?" + r"(?=.*startDate=2013-05-01)(?=.*endDate=2013-05-30).*$" + ) + mocked.get(pattern, payload={"energy": "bulk"}) + assert await solar_edge.get_energy_bulk( + [1, 4], + datetime.date(2013, 5, 1), + datetime.date(2013, 5, 30), + ) == {"energy": "bulk"} + await solar_edge.close() + + +@pytest.mark.asyncio +async def test_get_time_frame_energy() -> None: + """Test getting time frame energy and bulk.""" + async with aiointercept(mock_external_urls=True) as mocked: + solar_edge = SolarEdge("API_KEY") + pattern = re.compile( + r"^https://monitoringapi\.solaredge\.com/site/123/timeFrameEnergy\?" + r"(?=.*startDate=2013-05-01)(?=.*endDate=2013-05-06).*$" + ) + mocked.get(pattern, payload={"timeFrameEnergy": "tfe"}) + assert await solar_edge.get_time_frame_energy( + 123, + datetime.date(2013, 5, 1), + datetime.date(2013, 5, 6), + ) == {"timeFrameEnergy": "tfe"} + + pattern = re.compile( + r"^https://monitoringapi\.solaredge\.com/sites/1,4/timeFrameEnergy\?" + r"(?=.*startDate=2013-05-01)(?=.*endDate=2013-05-06).*$" + ) + mocked.get(pattern, payload={"timeFrameEnergy": "bulk"}) + assert await solar_edge.get_time_frame_energy_bulk( + [1, 4], "2013-05-01", "2013-05-06" + ) == {"timeFrameEnergy": "bulk"} + await solar_edge.close() + + +@pytest.mark.asyncio +async def test_get_power() -> None: + """Test getting power and bulk power.""" + async with aiointercept(mock_external_urls=True) as mocked: + solar_edge = SolarEdge("API_KEY") + start = datetime.datetime(2013, 6, 4, 11, 0, 0) + end = datetime.datetime(2013, 6, 4, 14, 0, 0) + pattern = re.compile( + r"^https://monitoringapi\.solaredge\.com/site/123/power\?" + r"(?=.*startTime=2013-06-04)(?=.*endTime=2013-06-04).*$" + ) + mocked.get(pattern, payload={"power": "power"}) + assert await solar_edge.get_power(123, start, end) == {"power": "power"} + + mocked.get(pattern, payload={"power": "power_str"}) + assert await solar_edge.get_power( + 123, "2013-06-04 11:00:00", "2013-06-04 14:00:00" + ) == {"power": "power_str"} + + pattern = re.compile( + r"^https://monitoringapi\.solaredge\.com/sites/1,4/power\?" + ) + mocked.get(pattern, payload={"power": "bulk"}) + assert await solar_edge.get_power_bulk([1, 4], start, end) == {"power": "bulk"} + await solar_edge.close() + + +@pytest.mark.asyncio +async def test_get_overview_bulk() -> None: + """Test getting bulk overview.""" + async with aiointercept(mock_external_urls=True) as mocked: + solar_edge = SolarEdge("API_KEY") + mocked.get( + "https://monitoringapi.solaredge.com/sites/1,4/overview?api_key=API_KEY", + payload={"overview": "bulk"}, + ) + assert await solar_edge.get_overview_bulk([1, 4]) == {"overview": "bulk"} + await solar_edge.close() + + +@pytest.mark.asyncio +async def test_get_power_details() -> None: + """Test getting power details.""" + async with aiointercept(mock_external_urls=True) as mocked: + solar_edge = SolarEdge("API_KEY") + start = datetime.datetime(2015, 11, 21, 11, 0, 0) + end = datetime.datetime(2015, 11, 21, 11, 30, 0) + pattern = re.compile( + r"^https://monitoringapi\.solaredge\.com/site/123/powerDetails\?" + ) + mocked.get(pattern, payload={"powerDetails": "pd"}) + assert await solar_edge.get_power_details(123, start, end) == { + "powerDetails": "pd" + } + + pattern = re.compile( + r"^https://monitoringapi\.solaredge\.com/site/123/powerDetails\?.*" + r"meters=PRODUCTION.*CONSUMPTION" + ) + mocked.get(pattern, payload={"powerDetails": "pd_meters"}) + assert await solar_edge.get_power_details( + 123, start, end, meters=["PRODUCTION", "CONSUMPTION"] + ) == {"powerDetails": "pd_meters"} + await solar_edge.close() + + +@pytest.mark.asyncio +async def test_get_environmental_benefits() -> None: + """Test getting environmental benefits.""" + async with aiointercept(mock_external_urls=True) as mocked: + solar_edge = SolarEdge("API_KEY") + mocked.get( + "https://monitoringapi.solaredge.com/site/123/envBenefits?api_key=API_KEY", + payload={"envBenefits": "eb"}, + ) + assert await solar_edge.get_environmental_benefits(123) == {"envBenefits": "eb"} + + pattern = re.compile( + r"^https://monitoringapi\.solaredge\.com/site/123/envBenefits\?.*" + r"systemUnits=Imperial" + ) + mocked.get(pattern, payload={"envBenefits": "imperial"}) + assert await solar_edge.get_environmental_benefits( + 123, system_units="Imperial" + ) == {"envBenefits": "imperial"} + await solar_edge.close()