diff --git a/README.md b/README.md index ad25183..29516ce 100644 --- a/README.md +++ b/README.md @@ -53,3 +53,42 @@ Period: 2025-01 to 2025-02, Value: 148.985 kWh, Calculated: False Period: 2025-02 to 2025-03, Value: 44.619 kWh, Calculated: False Period: 2025-03 to 2025-04, Value: 29.662 kWh, Calculated: False ``` + +### Listing metering points for a sharing group + +You can retrieve the metering points belonging to a sharing group for a given date. + +The Leneda API expects the sharing group contract number, for example +`CR00007479`. + +If no date is provided, the client uses today's date. + +```python +import asyncio +import os +from datetime import date + +from leneda import LenedaClient + + +async def main() -> None: + client = LenedaClient( + api_key=os.environ["LENEDA_API_KEY"], + energy_id=os.environ["LENEDA_ENERGY_ID"], + ) + + groups = await client.list_sharing_groups() + + for group in groups: + print(f"{group.contract_number} | {group.type}") + + metering_points = await client.get_sharing_group_metering_points( + group.contract_number, + on_date=date.today(), + ) + + for metering_point in metering_points: + print(f" {metering_point}") + + +asyncio.run(main()) diff --git a/examples/sharing_groups.py b/examples/sharing_groups.py new file mode 100644 index 0000000..14dd073 --- /dev/null +++ b/examples/sharing_groups.py @@ -0,0 +1,195 @@ +""" +Example: List sharing groups from Leneda. + +Environment variables: + LENEDA_API_KEY: Your Leneda API key + LENEDA_ENERGY_ID: Your Energy ID + +Usage: + python examples/sharing_groups.py + python examples/sharing_groups.py --type CEL + python examples/sharing_groups.py --all + python examples/sharing_groups.py --page 1 --size 20 +""" + +import argparse +import asyncio +import logging +import os +import sys + +from leneda import LenedaClient + +SHARING_GROUP_TYPES = ["AIR", "AIN", "ACR", "AC1", "CEL", "APS", "CER", "CEN"] + + +def parse_arguments() -> argparse.Namespace: + """Parse command-line arguments.""" + parser = argparse.ArgumentParser(description="List Leneda sharing groups") + + parser.add_argument( + "--api-key", + help="Your Leneda API key, or set LENEDA_API_KEY", + ) + parser.add_argument( + "--energy-id", + help="Your Energy ID, or set LENEDA_ENERGY_ID", + ) + parser.add_argument( + "--type", + choices=SHARING_GROUP_TYPES, + help="Filter by sharing group type", + ) + parser.add_argument( + "--page", + type=int, + default=1, + help="Page number to retrieve. Default: 1", + ) + parser.add_argument( + "--size", + type=int, + default=10, + help="Number of items per page. Default: 10", + ) + parser.add_argument( + "--all", + action="store_true", + help="Fetch all pages instead of only one page", + ) + parser.add_argument( + "--debug", + action="store_true", + help="Enable debug logging", + ) + parser.add_argument( + "--metering-points", + action="store_true", + help="Also list metering points for each sharing group", + ) + parser.add_argument( + "--date", + help="Date for sharing group metering points, in YYYY-MM-DD format. Defaults to today.", + ) + + return parser.parse_args() + + +def get_credentials(args: argparse.Namespace) -> tuple[str, str]: + """Get API credentials from arguments or environment variables.""" + api_key = args.api_key or os.environ.get("LENEDA_API_KEY") + energy_id = args.energy_id or os.environ.get("LENEDA_ENERGY_ID") + + if not api_key: + print("Error: API key not provided.") + print("Use --api-key or set LENEDA_API_KEY.") + sys.exit(1) + + if not energy_id: + print("Error: Energy ID not provided.") + print("Use --energy-id or set LENEDA_ENERGY_ID.") + sys.exit(1) + + return api_key, energy_id + + +def print_group(group) -> None: + """Print one sharing group.""" + print( + f"{group.contract_number} | " + f"{group.type} | " + f"Owner={group.owner_energy_id} | " + f"Manager={group.manager_energy_id} | " + f"Start={group.start_date} | " + f"End={group.end_date}" + ) + + +async def main() -> None: + """Run the sharing groups example.""" + args = parse_arguments() + + if args.debug: + logging.basicConfig(level=logging.DEBUG) + + api_key, energy_id = get_credentials(args) + + client = LenedaClient( + api_key=api_key, + energy_id=energy_id, + debug=args.debug, + ) + + try: + if args.all: + groups = await client.list_sharing_groups( + type=args.type, + size=args.size, + ) + + print(f"Retrieved {len(groups)} sharing group(s).") + + if not groups: + print("No sharing groups found for this Energy ID.") + return + + print() + for group in groups: + print_group(group) + + if args.metering_points: + metering_points = await client.get_sharing_group_metering_points( + group.contract_number, + on_date=args.date, + ) + + if not metering_points: + print(" No metering points found.") + else: + print(" Metering points:") + for metering_point in metering_points: + print(f" {metering_point}") + + else: + page = await client.get_sharing_groups( + page=args.page, + size=args.size, + type=args.type, + ) + + print( + f"Page {page.number}/{page.total_pages} - " + f"{len(page.content)} item(s) on this page, " + f"{page.total_elements} total item(s)." + ) + + if not page.content: + print("No sharing groups found on this page.") + return + + print() + for group in page.content: + print_group(group) + + if args.metering_points: + metering_points = await client.get_sharing_group_metering_points( + group.contract_number, + on_date=args.date, + ) + + if not metering_points: + print(" No metering points found.") + else: + print(" Metering points:") + for metering_point in metering_points: + print(f" {metering_point}") + + except Exception as exc: + print(f"Error while retrieving sharing groups: {exc}") + if args.debug: + raise + sys.exit(1) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/src/leneda/__init__.py b/src/leneda/__init__.py index 6ac7091..ad391b7 100644 --- a/src/leneda/__init__.py +++ b/src/leneda/__init__.py @@ -14,6 +14,8 @@ AggregatedMeteringValue, MeteringData, MeteringValue, + SharingGroup, + SharingGroupsPage, ) # Import the OBIS code constants @@ -30,5 +32,7 @@ "MeteringData", "AggregatedMeteringValue", "AggregatedMeteringData", + "SharingGroup", + "SharingGroupsPage", "__version__", ] diff --git a/src/leneda/client.py b/src/leneda/client.py index bc7b27b..bf21529 100644 --- a/src/leneda/client.py +++ b/src/leneda/client.py @@ -7,23 +7,30 @@ import json import logging -from datetime import datetime, timedelta -from typing import Any, Awaitable, Callable, Dict, List, Optional, Union +from datetime import date, datetime, timedelta +from typing import Any, Awaitable, Callable, Dict, List, Literal, Optional, Union import aiohttp from aiohttp import ClientResponseError, ClientTimeout -from .exceptions import ForbiddenException, MeteringPointNotFoundException, UnauthorizedException +from .exceptions import ( + ForbiddenException, + MeteringPointNotFoundException, + UnauthorizedException, +) from .models import ( AggregatedMeteringData, AuthenticationProbeResult, MeteringData, + SharingGroupsPage, ) from .obis_codes import ObisCode # Set up logging logger = logging.getLogger("leneda.client") +SharingGroupType = Literal["AIR", "AIN", "ACR", "AC1", "CEL", "APS", "CER", "CEN"] + class LenedaClient: """Client for the Leneda API.""" @@ -396,3 +403,136 @@ async def probe_credentials(self) -> AuthenticationProbeResult: return AuthenticationProbeResult.UNKNOWN return AuthenticationProbeResult.UNKNOWN + + async def get_sharing_groups( + self, + page: int = 1, + size: int = 10, + type: Optional[SharingGroupType] = None, + ) -> SharingGroupsPage: + """ + Get a paginated list of sharing groups available to the authenticated Energy ID. + + Args: + page: Page number to retrieve. Leneda uses 1-based pagination. + size: Number of items per page. + type: Optional sharing group type filter. + Supported values: AIR, AIN, ACR, AC1, CEL, APS, CER, CEN. + + Returns: + SharingGroupsPage containing the paginated sharing groups response. + + Raises: + UnauthorizedException: If the API returns a 401 status code + ForbiddenException: If the API returns a 403 status code + aiohttp.ClientError: For other request errors + json.JSONDecodeError: If the response cannot be parsed as JSON + """ + params: Dict[str, Any] = { + "page": page, + "size": size, + } + + if type is not None: + params["type"] = type + + response_data = await self._make_request( + method="GET", + endpoint="sharing-groups-small", + params=params, + ) + + return SharingGroupsPage.from_dict(response_data) + + async def list_sharing_groups( + self, + type: Optional[SharingGroupType] = None, + size: int = 100, + ) -> List: + """ + Get all sharing groups available to the authenticated Energy ID. + + This method follows Leneda pagination automatically until there are no + more pages. + + Args: + type: Optional sharing group type filter. + Supported values: AIR, AIN, ACR, AC1, CEL, APS, CER, CEN. + size: Number of items per page. + + Returns: + A list of all SharingGroup objects. + """ + page = 1 + groups = [] + + while True: + result = await self.get_sharing_groups( + page=page, + size=size, + type=type, + ) + + groups.extend(result.content) + + if not result.has_next_page: + break + + page += 1 + + return groups + + async def get_sharing_group_metering_points( + self, + contract_number: str, + on_date: Optional[Union[str, date, datetime]] = None, + ) -> List[Dict[str, Any]]: + """ + Get all metering points belonging to a sharing group for a given date. + + Args: + contract_number: The contract number of the sharing group, + for example "CR00007479". + on_date: Date for which to retrieve the metering points. + If omitted, today's date is used. + Can be a date, datetime, or ISO date string. + + Returns: + A list of metering point dictionaries as returned by the Leneda API. + + Raises: + UnauthorizedException: If the API returns a 401 status code + ForbiddenException: If the API returns a 403 status code + aiohttp.ClientError: For other request errors + json.JSONDecodeError: If the response cannot be parsed as JSON + """ + if on_date is None: + date_value = date.today().isoformat() + elif isinstance(on_date, datetime): + date_value = on_date.date().isoformat() + elif isinstance(on_date, date): + date_value = on_date.isoformat() + else: + date_value = on_date + + endpoint = f"sharing-groups/{contract_number}/metering-points" + + response_data = await self._make_request( + method="GET", + endpoint=endpoint, + params={"date": date_value}, + ) + + if isinstance(response_data, list): + return response_data + + if isinstance(response_data, dict): + if "content" in response_data and isinstance(response_data["content"], list): + return response_data["content"] + + if "meteringPoints" in response_data and isinstance( + response_data["meteringPoints"], list + ): + return response_data["meteringPoints"] + + return [] diff --git a/src/leneda/models.py b/src/leneda/models.py index 6c144af..37cc06c 100644 --- a/src/leneda/models.py +++ b/src/leneda/models.py @@ -7,9 +7,9 @@ import logging from dataclasses import dataclass, field -from datetime import datetime +from datetime import date, datetime from enum import Enum -from typing import Any, Dict, List +from typing import Any, Dict, List, Optional from dateutil import parser @@ -246,3 +246,98 @@ class AuthenticationProbeResult(Enum): SUCCESS = "SUCCESS" FAILURE = "FAILURE" UNKNOWN = "UNKNOWN" + + +@dataclass +class SharingGroup: + """A sharing group returned by the Leneda API.""" + + id: str + contract_number: str + type: str + owner_energy_id: str + manager_energy_id: str + start_date: date + end_date: Optional[date] = None + + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> "SharingGroup": + """Create a SharingGroup from a dictionary.""" + try: + return cls( + id=data["id"], + contract_number=data["contractNumber"], + type=data["type"], + owner_energy_id=data["ownerEnergyId"], + manager_energy_id=data["managerEnergyId"], + start_date=date.fromisoformat(data["startDate"]), + end_date=(date.fromisoformat(data["endDate"]) if data.get("endDate") else None), + ) + except KeyError as e: + logger.error(f"Missing key in sharing group API response: {e}") + logger.debug(f"API response data: {data}") + raise + except Exception as e: + logger.error(f"Error parsing sharing group: {e}") + logger.debug(f"API response data: {data}") + raise + + def to_dict(self) -> Dict[str, Any]: + """Convert the SharingGroup to a dictionary.""" + return { + "id": self.id, + "contractNumber": self.contract_number, + "type": self.type, + "ownerEnergyId": self.owner_energy_id, + "managerEnergyId": self.manager_energy_id, + "startDate": self.start_date.isoformat(), + "endDate": self.end_date.isoformat() if self.end_date else None, + } + + +@dataclass +class SharingGroupsPage: + """Paginated sharing groups response.""" + + content: List[SharingGroup] = field(default_factory=list) + has_next_page: bool = False + number: int = 1 + size: int = 10 + total_elements: int = 0 + total_pages: int = 0 + + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> "SharingGroupsPage": + """Create a SharingGroupsPage from a dictionary.""" + try: + groups = [] + for item_data in data.get("content", []): + try: + groups.append(SharingGroup.from_dict(item_data)) + except Exception as e: + logger.warning(f"Skipping invalid sharing group: {e}") + logger.debug(f"Invalid sharing group data: {item_data}") + + return cls( + content=groups, + has_next_page=bool(data.get("hasNextPage", False)), + number=int(data.get("number", 1)), + size=int(data.get("size", 10)), + total_elements=int(data.get("totalElements", len(groups))), + total_pages=int(data.get("totalPages", 1)), + ) + except Exception as e: + logger.error(f"Error creating SharingGroupsPage: {e}") + logger.debug(f"API response data: {data}") + raise + + def to_dict(self) -> Dict[str, Any]: + """Convert the SharingGroupsPage to a dictionary.""" + return { + "content": [group.to_dict() for group in self.content], + "hasNextPage": self.has_next_page, + "number": self.number, + "size": self.size, + "totalElements": self.total_elements, + "totalPages": self.total_pages, + } diff --git a/tests/test_client.py b/tests/test_client.py index 226a139..c18dc47 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -4,6 +4,7 @@ import json import unittest +from datetime import date from typing import Any from unittest.mock import AsyncMock, MagicMock, patch @@ -23,6 +24,8 @@ AuthenticationProbeResult, MeteringData, MeteringValue, + SharingGroup, + SharingGroupsPage, ) from src.leneda.obis_codes import ObisCode @@ -80,6 +83,34 @@ def setup(self) -> None: ], } + self.sample_sharing_groups = { + "content": [ + { + "id": "531e0bce-1fbd-4ba7-947f-212a5e62b438", + "contractNumber": "CR00000001", + "type": "AIR", + "ownerEnergyId": "LUXE-MARIE-CAT42", + "managerEnergyId": "LUXE-LE-ET-ECHO1", + "startDate": "2021-11-01", + "endDate": None, + }, + { + "id": "6def0402-cf29-42c2-a83a-7e7f6e8eb7d0", + "contractNumber": "CR00000002", + "type": "CEL", + "ownerEnergyId": "LUXE-MARIE-CAT42", + "managerEnergyId": "LUXE-LE-ET-ECHO1", + "startDate": "2025-11-01", + "endDate": None, + }, + ], + "hasNextPage": False, + "number": 1, + "size": 10, + "totalElements": 2, + "totalPages": 1, + } + @patch("aiohttp.ClientSession.request") async def test_get_time_series(self, mock_request: Any) -> None: """Test getting time series data.""" @@ -450,6 +481,218 @@ async def test_probe_credentials_network_error(self, mock_request: Any) -> None: # Check that the request was attempted mock_request.assert_called_once() + @patch("aiohttp.ClientSession.request") + async def test_get_sharing_groups(self, mock_request: Any) -> None: + """Test getting paginated sharing groups.""" + mock_response = AsyncMock() + mock_response.status = 200 + mock_response.json = AsyncMock(return_value=self.sample_sharing_groups) + mock_response.content = json.dumps(self.sample_sharing_groups).encode() + mock_response.raise_for_status = lambda: None + mock_request.return_value.__aenter__.return_value = mock_response + + result = await self.client.get_sharing_groups(page=1, size=10) + + assert isinstance(result, SharingGroupsPage) + assert result.number == 1 + assert result.size == 10 + assert result.total_elements == 2 + assert result.total_pages == 1 + assert result.has_next_page is False + assert len(result.content) == 2 + + first_group = result.content[1] + assert isinstance(first_group, SharingGroup) + assert first_group.id == "6def0402-cf29-42c2-a83a-7e7f6e8eb7d0" + assert first_group.contract_number == "CR00000002" + assert first_group.type == "CEL" + assert first_group.owner_energy_id == "LUXE-MARIE-CAT42" + assert first_group.manager_energy_id == "LUXE-LE-ET-ECHO1" + assert first_group.start_date.isoformat() == "2025-11-01" + assert first_group.end_date is None + + mock_request.assert_called_once() + call_kwargs = mock_request.call_args[1] + + assert call_kwargs["method"] == "GET" + assert call_kwargs["url"] == "https://api.leneda.lu/api/sharing-groups-small" + assert call_kwargs["headers"] == { + "X-API-KEY": "test_api_key", + "X-ENERGY-ID": "test_energy_id", + "Content-Type": "application/json", + } + assert call_kwargs["params"] == { + "page": 1, + "size": 10, + } + + @patch("aiohttp.ClientSession.request") + async def test_get_sharing_groups_with_type_filter(self, mock_request: Any) -> None: + """Test getting sharing groups with a type filter.""" + mock_response = AsyncMock() + mock_response.status = 200 + mock_response.json = AsyncMock(return_value=self.sample_sharing_groups) + mock_response.content = json.dumps(self.sample_sharing_groups).encode() + mock_response.raise_for_status = lambda: None + mock_request.return_value.__aenter__.return_value = mock_response + + result = await self.client.get_sharing_groups(page=1, size=10, type="CEL") + + assert isinstance(result, SharingGroupsPage) + + mock_request.assert_called_once() + call_kwargs = mock_request.call_args[1] + + assert call_kwargs["method"] == "GET" + assert call_kwargs["url"] == "https://api.leneda.lu/api/sharing-groups-small" + assert call_kwargs["params"] == { + "page": 1, + "size": 10, + "type": "CEL", + } + + @patch("aiohttp.ClientSession.request") + async def test_list_sharing_groups(self, mock_request: Any) -> None: + """Test listing all sharing groups across pages.""" + first_page = { + "content": [ + { + "id": "group-1", + "contractNumber": "CR00000001", + "type": "AIR", + "ownerEnergyId": "owner-1", + "managerEnergyId": "manager-1", + "startDate": "2024-01-01", + "endDate": None, + } + ], + "hasNextPage": True, + "number": 1, + "size": 1, + "totalElements": 2, + "totalPages": 2, + } + + second_page = { + "content": [ + { + "id": "group-2", + "contractNumber": "CR00000002", + "type": "CEL", + "ownerEnergyId": "owner-2", + "managerEnergyId": "manager-2", + "startDate": "2024-02-01", + "endDate": None, + } + ], + "hasNextPage": False, + "number": 2, + "size": 1, + "totalElements": 2, + "totalPages": 2, + } + + first_response = AsyncMock() + first_response.status = 200 + first_response.json = AsyncMock(return_value=first_page) + first_response.content = json.dumps(first_page).encode() + first_response.raise_for_status = lambda: None + + second_response = AsyncMock() + second_response.status = 200 + second_response.json = AsyncMock(return_value=second_page) + second_response.content = json.dumps(second_page).encode() + second_response.raise_for_status = lambda: None + + mock_request.return_value.__aenter__.side_effect = [ + first_response, + second_response, + ] + + result = await self.client.list_sharing_groups(size=1) + + assert len(result) == 2 + assert result[0].id == "group-1" + assert result[1].id == "group-2" + + assert mock_request.call_count == 2 + + @patch("aiohttp.ClientSession.request") + async def test_get_sharing_group_metering_points(self, mock_request: Any) -> None: + """Test getting metering points for a sharing group.""" + contract_number = "CR00007479" + + sample_metering_points = [ + { + "code": "LU0000012345678901234000000000000", + "type": "CONSUMPTION", + }, + { + "code": "LU0000098765432109876000000000000", + "type": "PRODUCTION", + }, + ] + + mock_response = AsyncMock() + mock_response.status = 200 + mock_response.json = AsyncMock(return_value=sample_metering_points) + mock_response.content = json.dumps(sample_metering_points).encode() + mock_response.raise_for_status = lambda: None + mock_request.return_value.__aenter__.return_value = mock_response + + result = await self.client.get_sharing_group_metering_points( + contract_number, + on_date=date(2026, 5, 22), + ) + + assert isinstance(result, list) + assert len(result) == 2 + assert result[0]["code"] == "LU0000012345678901234000000000000" + assert result[1]["code"] == "LU0000098765432109876000000000000" + + mock_request.assert_called_once() + call_kwargs = mock_request.call_args[1] + + assert call_kwargs["method"] == "GET" + assert ( + call_kwargs["url"] == "https://api.leneda.lu/api/" + "sharing-groups/CR00007479/metering-points" + ) + assert call_kwargs["headers"] == { + "X-API-KEY": "test_api_key", + "X-ENERGY-ID": "test_energy_id", + "Content-Type": "application/json", + } + assert call_kwargs["params"] == { + "date": "2026-05-22", + } + + @patch("aiohttp.ClientSession.request") + async def test_get_sharing_group_metering_points_defaults_to_today( + self, + mock_request: Any, + ) -> None: + """Test that sharing group metering points defaults to today's date.""" + contract_number = "CR00007479" + + mock_response = AsyncMock() + mock_response.status = 200 + mock_response.json = AsyncMock(return_value=[]) + mock_response.content = json.dumps([]).encode() + mock_response.raise_for_status = lambda: None + mock_request.return_value.__aenter__.return_value = mock_response + + await self.client.get_sharing_group_metering_points(contract_number) + + call_kwargs = mock_request.call_args[1] + + assert call_kwargs["url"] == ( + "https://api.leneda.lu/api/sharing-groups/" "CR00007479/metering-points" + ) + assert call_kwargs["params"] == { + "date": date.today().isoformat(), + } + def build_error_mock_response(status: int, message: str) -> AsyncMock: """Build a mock response for an error case."""