14911143141

Committed 08 May 2025 04:17PM UTC coverage: 95.0%. First build

Build # 14911143141

Build Type

Pull #47

github

Committed by

web-flow

Commit Message

Merge f555c0299 into bb2d3873f

Pull Request Pull Request #47: [repo-helper] Configuration Update

Run Details

342 of 360 relevant lines covered (95.0%)

0.95 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

94.2

/octo_api/api.py

#!/usr/bin/env python3
#
#  api.py
"""
The primary interface to the Octopus Energy API.

.. note::

        The Octopus Energy API uses the term "Grid Supply Point" (GSP) to refer to what are actually
        the 14 former Public Electricity Suppliers. The GSP terminology has been used here to better
        reflect the REST API.

"""
#
#  Copyright © 2020 Dominic Davis-Foster <dominic@davis-foster.co.uk>
#
#  Permission is hereby granted, free of charge, to any person obtaining a copy
#  of this software and associated documentation files (the "Software"), to deal
#  in the Software without restriction, including without limitation the rights
#  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
#  copies of the Software, and to permit persons to whom the Software is
#  furnished to do so, subject to the following conditions:
#
#  The above copyright notice and this permission notice shall be included in all
#  copies or substantial portions of the Software.
#
#  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
#  EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
#  MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
#  IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
#  DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
#  OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE
#  OR OTHER DEALINGS IN THE SOFTWARE.
#

# stdlib
from datetime import datetime
from typing import Any, Dict, MutableMapping, Optional, Union

# 3rd party
from apeye.slumber_url import SlumberURL
from domdf_python_tools.secrets import Secret
from typing_extensions import Literal

# this package
from octo_api.consumption import Consumption
from octo_api.pagination import PaginatedResponse
from octo_api.products import DetailedProduct, Product, RateInfo
from octo_api.utils import MeterPointDetails, RateType, Region

__all__ = ["OctoAPI"]


class OctoAPI:
        """
        The primary interface to the Octopus Energy API.

        :param api_key: API key to access the Octopus Energy API.

        If you are an Octopus Energy customer, you can generate an API key from your
        `online dashboard <https://octopus.energy/dashboard/developer/>`_.
        """

        def __init__(self, api_key: str):

                #: The API key to access the Octopus Energy API.
                self.API_KEY: Secret = Secret(api_key)

                #: The base URL of the Octopus Energy API.
                self.API_BASE: SlumberURL = SlumberURL("https://api.octopus.energy/v1", auth=(self.API_KEY.value, ''))

        def get_products(
                        self,
                        is_variable: Optional[bool] = None,
                        is_green: Optional[bool] = None,
                        is_tracker: Optional[bool] = None,
                        is_prepay: Optional[bool] = None,
                        is_business: bool = False,
                        available_at: Optional[datetime] = None,
                        ) -> PaginatedResponse[Product]:
                """
                Returns a list of energy products.

                By default, the results only include public energy products.
                Authenticated organisations will also see products available to their organisation.

                :param is_variable: Show only variable products.
                :param is_green: Show only green products.
                :param is_tracker: Show only tracker products.
                :param is_prepay: Show only pre-pay products.
                :param is_business: Show only business products.
                :param available_at: Show products available for new agreements on the given datetime.
                        Defaults to the current datetime, effectively showing products that are currently available.
                :no-default available_at:

                .. https://developer.octopus.energy/docs/api/#list-products

                **Example**

                .. code-block:: python

                        >>> api.get_products()[0]
                        octo_api.products.Product(
                                available_from='2016-01-01T:00:00:00+00:00',
                                available_to=None,
                                brand='AFFECT_ENERGY',
                                code='1201',
                                description='Affect Standard Tariff',
                                display_name='Affect Standard Tariff',
                                full_name='Affect Standard Tariff',
                                is_business=False,
                                is_green=False,
                                is_prepay=False,
                                is_restricted=False,
                                is_tracker=False,
                                is_variable=True,
                                links=[
                                        {
                                                'href': 'https://api.octopus.energy/v1/products/1201/',
                                                'method': 'GET',
                                                'rel': 'self'
                                        }
                                ],
                                term=None,
                                direction='IMPORT',
                        )

                """

                parameters: Dict[str, Any] = {}

                if is_variable is not None:
                        parameters["is_variable"] = is_variable
                if is_green is not None:
                        parameters["is_green"] = is_green
                if is_tracker is not None:
                        parameters["is_tracker"] = is_tracker
                if is_prepay is not None:
                        parameters["is_prepay"] = is_prepay
                parameters["is_business"] = is_business
                if available_at is not None:
                        parameters["available_at"] = available_at.isoformat()

                query_url = self.API_BASE / "products"
                return PaginatedResponse(query_url, parameters, obj_type=Product)

        def get_product_info(
                        self,
                        product_code: str,
                        tariffs_active_at: Optional[datetime] = None,
                        ) -> DetailedProduct:
                """
                Retrieve the details of a product (including all its tariffs) for a particular point in time.

                :param product_code: The code of the product to be retrieved, for example ``VAR-17-01-11``.
                :param tariffs_active_at: The point in time in which to show the active charges. Defaults to current datetime.
                :no-default available_at:

                .. https://developer.octopus.energy/docs/api/#retrieve-a-product


                **Example**

                .. code-block:: python

                        >>> api.get_product_info(product_code='VAR-17-01-11')
                        octo_api.products.DetailedProduct(
                                available_from='2017-01-11T10:00:00+00:00',
                                available_to='2018-02-15T00:00:00+00:00',
                                brand='S_ENERGY',
                                code='7-01-11',
                                description='This variable tariff always offers great value - driven by our'
                                                        'belief that prices should be fair for the long term, not just a'
                                                        'fixed term. We aim for 50% renewable electricity on this tariff.',
                                display_name='pus',
                                full_name='ctopus January 2017 v1',
                                is_business=False,
                                is_green=False,
                                is_prepay=False,
                                is_restricted=False,
                                is_tracker=False,
                                is_variable=True,
                                links=[
                                        {
                                                'href': 'https://api.octopus.energy/v1/products/VAR-17-01-11/',
                                                'method': 'GET',
                                                'rel': 'self'
                                        }
                                ],
                                term=None,
                                tariffs_active_at='2020-10-26T11:15:17.208285+00:00',
                                single_register_electricity_tariffs=RegionalTariffs(['direct_debit_monthly']),
                                dual_register_electricity_tariffs=RegionalTariffs(['direct_debit_monthly']),
                                single_register_gas_tariffs=RegionalTariffs(['direct_debit_monthly']),
                                sample_quotes=RegionalQuotes([dual_fuel_dual_rate, dual_fuel_single_rate, electricity_dual_rate, electricity_single_rate]),
                                sample_consumption={
                                        'electricity_single_rate': {'electricity_standard': 2900},
                                        'electricity_dual_rate': {
                                                'electricity_day': 2436,
                                                'electricity_night': 1764
                                        },
                                        'dual_fuel_single_rate': {
                                                'electricity_standard': 2900,
                                                'gas_standard': 12000
                                        },
                                        'dual_fuel_dual_rate': {
                                                'electricity_day': 2436,
                                                'electricity_night': 1764,
                                                'gas_standard': 12000
                                        }
                                },
                        )
                """

                parameters = {}

                if tariffs_active_at is not None:
                        parameters["tariffs_active_at"] = tariffs_active_at.isoformat()

                query_url = self.API_BASE / "products" / product_code
                return DetailedProduct(**query_url.get(**parameters))

        def get_tariff_charges(
                        self,
                        product_code: str,
                        tariff_code: str,
                        fuel: Literal["electricity", "gas"],
                        rate_type: RateType,
                        period_from: Optional[datetime] = None,
                        period_to: Optional[datetime] = None,
                        page_size: int = 100,
                        ) -> PaginatedResponse[RateInfo]:
                """
                Returns a list of time periods and their associated unit rates charges.

                If the tariff has a fixed unit rate the list will only contain one element.

                :param product_code: The code of the product to be retrieved, for example ``VAR-17-01-11``.
                :param tariff_code: The code of the tariff to be retrieved, for example ``E-1R-VAR-17-01-11-A``.
                        From what I can tell the format is::

                                <E for electricity><optional hyphen><1R for single rate?><the product code>-<the grid supply point>

                :param fuel:
                :param rate_type:
                :param period_from: Show charges active from the given datetime (inclusive).
                        This parameter can be provided on its own.
                :param period_to: Show charges active up to the given datetime (exclusive).
                        You must also provide the ``period_from`` parameter in order to create a range.
                :param page_size: Page size of returned results.
                        Default is ``100``, maximum is ``1,500`` to give up to a month of half-hourly prices.
                :no-default page_size:

                .. https://developer.octopus.energy/docs/api/#list-tariff-charges

                .. note::

                        If you're using this API to query future unit-rates of the Agile Octopus product,
                        note that day-ahead prices are normally created by 4pm in the Europe/London timezone.
                        Further, the market index used to calculate unit rates is based in the CET timezone (UTC+1)
                        and so its "day" corresponds to 11pm to 11pm in UK time.
                        Hence, if you query today's unit rates before 4pm, you'll get 46 results back rather than 48.
                """

                parameters: MutableMapping[str, Union[str, int]] = {}

                if period_from is not None:
                        parameters["period_from"] = period_from.isoformat()
                if period_to is not None:
                        parameters["period_to"] = period_to.isoformat()

                if page_size > 1500:
                        raise ValueError("'page_size' may not be greater than 1,500")

                parameters["page_size"] = int(page_size)

                query_url = self.API_BASE / "products" / product_code / f"{fuel}-tariffs" / tariff_code / str(rate_type)
                return PaginatedResponse(query_url, query_params=parameters, obj_type=RateInfo)

        def get_meter_point_details(self, mpan: str) -> MeterPointDetails:
                """
                Retrieve the details of a meter-point.

                This can be used to get the GSP of a given meter-point.

                :param mpan: The electricity meter-point's MPAN.

                :return:
                """

                return MeterPointDetails._from_dict((self.API_BASE / "electricity-meter-points" / mpan).get())

        def get_grid_supply_point(self, postcode: str) -> Region:
                """
                Returns the grid supply point for the given postcode.

                :param postcode:

                :raises: :exc:`ValueError` if the postcode cannot be mapped to a GSP.
                """

                query_url = self.API_BASE / "industry" / "grid-supply-points"

                results = query_url.get(postcode=postcode)["results"]
                if results:
                        return Region(results[0]["group_id"])
                else:
                        raise ValueError(f"Cannot map the postcode {postcode!r} to a GSP.")

        def get_consumption(
                        self,
                        mpan: str,
                        serial_number: str,
                        fuel: Literal["electricity", "gas"],
                        period_from: Optional[datetime] = None,
                        period_to: Optional[datetime] = None,
                        page_size: int = 100,
                        reverse: bool = False,
                        group_by: Optional[str] = None,
                        ) -> PaginatedResponse[Consumption]:
                r"""
                Return a list of consumption values for half-hour periods for a given meter-point and meter.

                Unit of measurement:

                * Electricity meters: kWh
                * SMETS1 Secure gas meters: kWh
                * SMETS2 gas meters: m\ :superscript:`3`

                .. attention::

                        Half-hourly consumption data is only available for smart meters.
                        Requests for consumption data for non-smart meters will return an empty response payload.

                :param mpan: The electricity meter-point's MPAN or gas meter-point's MPRN.
                :param serial_number: The meter's serial number.
                :param fuel:
                :param period_from: Show consumption for periods which start at or after the given datetime.
                        This parameter can be provided on its own.
                :param period_to: Show consumption for periods which start at or before the given datetime.
                        This parameter also requires providing the ``period_from`` parameter to create a range.
                :param page_size: Page size of returned results.
                        Default is ``100``, maximum is ``25,000`` to give a full year of half-hourly consumption details.
                :no-default page_size:
                :param reverse: Returns the results ordered from most oldest to newest. By default the results are from most recent backwards.
                :no-default reverse:
                :param group_by: The grouping of the consumption data.
                        By default the consumption is returned in half-hour periods.

                        Possible alternatives are:

                        * ``'hour'``
                        * ``'day'``
                        * ``'week'``
                        * ``'month'``
                        * ``'quarter'``
                :no-default group_by:
                """

                parameters: MutableMapping[str, Union[str, int]] = {}

                if period_from is not None:
                        parameters["period_from"] = period_from.isoformat()
                if period_to is not None:
                        parameters["period_to"] = period_to.isoformat()

                if page_size > 25000:
                        raise ValueError("'page_size' may not be greater than 25,000")

                parameters["page_size"] = int(page_size)

                if reverse:
                        parameters["order_by"] = "period"
                if group_by is not None:
                        parameters["group_by"] = str(group_by)

                query_url = self.API_BASE / f"{fuel}-meter-points" / mpan / "meters" / serial_number / "consumption"
                return PaginatedResponse(query_url, query_params=parameters, obj_type=Consumption)

1	#!/usr/bin/env python3
2	#
3	# api.py
4	"""
5	The primary interface to the Octopus Energy API.
6
7	.. note::
8
9	The Octopus Energy API uses the term "Grid Supply Point" (GSP) to refer to what are actually
10	the 14 former Public Electricity Suppliers. The GSP terminology has been used here to better
11	reflect the REST API.
12
13	"""
14	#
15	# Copyright © 2020 Dominic Davis-Foster <dominic@davis-foster.co.uk>
16	#
17	# Permission is hereby granted, free of charge, to any person obtaining a copy
18	# of this software and associated documentation files (the "Software"), to deal
19	# in the Software without restriction, including without limitation the rights
20	# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
21	# copies of the Software, and to permit persons to whom the Software is
22	# furnished to do so, subject to the following conditions:
23	#
24	# The above copyright notice and this permission notice shall be included in all
25	# copies or substantial portions of the Software.
26	#
27	# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
28	# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
29	# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
30	# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
31	# DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
32	# OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE
33	# OR OTHER DEALINGS IN THE SOFTWARE.
34	#
35
36	# stdlib
37	from datetime import datetime	1✔
38	from typing import Any, Dict, MutableMapping, Optional, Union	1✔
39
40	# 3rd party
41	from apeye.slumber_url import SlumberURL	1✔
42	from domdf_python_tools.secrets import Secret	1✔
43	from typing_extensions import Literal	1✔
44
45	# this package
46	from octo_api.consumption import Consumption	1✔
47	from octo_api.pagination import PaginatedResponse	1✔
48	from octo_api.products import DetailedProduct, Product, RateInfo	1✔
49	from octo_api.utils import MeterPointDetails, RateType, Region	1✔
50
51	__all__ = ["OctoAPI"]	1✔
52
53
54	class OctoAPI:	1✔
55	"""
56	The primary interface to the Octopus Energy API.
57
58	:param api_key: API key to access the Octopus Energy API.
59
60	If you are an Octopus Energy customer, you can generate an API key from your
61	`online dashboard <https://octopus.energy/dashboard/developer/>`_.
62	"""
63
64	def __init__(self, api_key: str):	1✔
65
66	#: The API key to access the Octopus Energy API.
67	self.API_KEY: Secret = Secret(api_key)	1✔
68
69	#: The base URL of the Octopus Energy API.
70	self.API_BASE: SlumberURL = SlumberURL("https://api.octopus.energy/v1", auth=(self.API_KEY.value, ''))	1✔
71
72	def get_products(	1✔
73	self,
74	is_variable: Optional[bool] = None,
75	is_green: Optional[bool] = None,
76	is_tracker: Optional[bool] = None,
77	is_prepay: Optional[bool] = None,
78	is_business: bool = False,
79	available_at: Optional[datetime] = None,
80	) -> PaginatedResponse[Product]:
81	"""
82	Returns a list of energy products.
83
84	By default, the results only include public energy products.
85	Authenticated organisations will also see products available to their organisation.
86
87	:param is_variable: Show only variable products.
88	:param is_green: Show only green products.
89	:param is_tracker: Show only tracker products.
90	:param is_prepay: Show only pre-pay products.
91	:param is_business: Show only business products.
92	:param available_at: Show products available for new agreements on the given datetime.
93	Defaults to the current datetime, effectively showing products that are currently available.
94	:no-default available_at:
95
96	.. https://developer.octopus.energy/docs/api/#list-products
97
98	Example
99
100	.. code-block:: python
101
102	>>> api.get_products()[0]
103	octo_api.products.Product(
104	available_from='2016-01-01T:00:00:00+00:00',
105	available_to=None,
106	brand='AFFECT_ENERGY',
107	code='1201',
108	description='Affect Standard Tariff',
109	display_name='Affect Standard Tariff',
110	full_name='Affect Standard Tariff',
111	is_business=False,
112	is_green=False,
113	is_prepay=False,
114	is_restricted=False,
115	is_tracker=False,
116	is_variable=True,
117	links=[
118	{
119	'href': 'https://api.octopus.energy/v1/products/1201/',
120	'method': 'GET',
121	'rel': 'self'
122	}
123	],
124	term=None,
125	direction='IMPORT',
126	)
127
128	"""
129
130	parameters: Dict[str, Any] = {}	1✔
131
132	if is_variable is not None:	1✔
133	parameters["is_variable"] = is_variable	1✔
134	if is_green is not None:	1✔
135	parameters["is_green"] = is_green	1✔
136	if is_tracker is not None:	1✔
137	parameters["is_tracker"] = is_tracker	1✔
138	if is_prepay is not None:	1✔
139	parameters["is_prepay"] = is_prepay	1✔
140	parameters["is_business"] = is_business	1✔
141	if available_at is not None:	1✔
142	parameters["available_at"] = available_at.isoformat()	×
143
144	query_url = self.API_BASE / "products"	1✔
145	return PaginatedResponse(query_url, parameters, obj_type=Product)	1✔
146
147	def get_product_info(	1✔
148	self,
149	product_code: str,
150	tariffs_active_at: Optional[datetime] = None,
151	) -> DetailedProduct:
152	"""
153	Retrieve the details of a product (including all its tariffs) for a particular point in time.
154
155	:param product_code: The code of the product to be retrieved, for example ``VAR-17-01-11``.
156	:param tariffs_active_at: The point in time in which to show the active charges. Defaults to current datetime.
157	:no-default available_at:
158
159	.. https://developer.octopus.energy/docs/api/#retrieve-a-product
160
161
162	Example
163
164	.. code-block:: python
165
166	>>> api.get_product_info(product_code='VAR-17-01-11')
167	octo_api.products.DetailedProduct(
168	available_from='2017-01-11T10:00:00+00:00',
169	available_to='2018-02-15T00:00:00+00:00',
170	brand='S_ENERGY',
171	code='7-01-11',
172	description='This variable tariff always offers great value - driven by our'
173	'belief that prices should be fair for the long term, not just a'
174	'fixed term. We aim for 50% renewable electricity on this tariff.',
175	display_name='pus',
176	full_name='ctopus January 2017 v1',
177	is_business=False,
178	is_green=False,
179	is_prepay=False,
180	is_restricted=False,
181	is_tracker=False,
182	is_variable=True,
183	links=[
184	{
185	'href': 'https://api.octopus.energy/v1/products/VAR-17-01-11/',
186	'method': 'GET',
187	'rel': 'self'
188	}
189	],
190	term=None,
191	tariffs_active_at='2020-10-26T11:15:17.208285+00:00',
192	single_register_electricity_tariffs=RegionalTariffs(['direct_debit_monthly']),
193	dual_register_electricity_tariffs=RegionalTariffs(['direct_debit_monthly']),
194	single_register_gas_tariffs=RegionalTariffs(['direct_debit_monthly']),
195	sample_quotes=RegionalQuotes([dual_fuel_dual_rate, dual_fuel_single_rate, electricity_dual_rate, electricity_single_rate]),
196	sample_consumption={
197	'electricity_single_rate': {'electricity_standard': 2900},
198	'electricity_dual_rate': {
199	'electricity_day': 2436,
200	'electricity_night': 1764
201	},
202	'dual_fuel_single_rate': {
203	'electricity_standard': 2900,
204	'gas_standard': 12000
205	},
206	'dual_fuel_dual_rate': {
207	'electricity_day': 2436,
208	'electricity_night': 1764,
209	'gas_standard': 12000
210	}
211	},
212	)
213	"""
214
215	parameters = {}	1✔
216
217	if tariffs_active_at is not None:	1✔
218	parameters["tariffs_active_at"] = tariffs_active_at.isoformat()	×
219
220	query_url = self.API_BASE / "products" / product_code	1✔
221	return DetailedProduct(query_url.get(parameters))	1✔
222
223	def get_tariff_charges(	1✔
224	self,
225	product_code: str,
226	tariff_code: str,
227	fuel: Literal["electricity", "gas"],
228	rate_type: RateType,
229	period_from: Optional[datetime] = None,
230	period_to: Optional[datetime] = None,
231	page_size: int = 100,
232	) -> PaginatedResponse[RateInfo]:
233	"""
234	Returns a list of time periods and their associated unit rates charges.
235
236	If the tariff has a fixed unit rate the list will only contain one element.
237
238	:param product_code: The code of the product to be retrieved, for example ``VAR-17-01-11``.
239	:param tariff_code: The code of the tariff to be retrieved, for example ``E-1R-VAR-17-01-11-A``.
240	From what I can tell the format is::
241
242	<E for electricity><optional hyphen><1R for single rate?><the product code>-<the grid supply point>
243
244	:param fuel:
245	:param rate_type:
246	:param period_from: Show charges active from the given datetime (inclusive).
247	This parameter can be provided on its own.
248	:param period_to: Show charges active up to the given datetime (exclusive).
249	You must also provide the ``period_from`` parameter in order to create a range.
250	:param page_size: Page size of returned results.
251	Default is ``100``, maximum is ``1,500`` to give up to a month of half-hourly prices.
252	:no-default page_size:
253
254	.. https://developer.octopus.energy/docs/api/#list-tariff-charges
255
256	.. note::
257
258	If you're using this API to query future unit-rates of the Agile Octopus product,
259	note that day-ahead prices are normally created by 4pm in the Europe/London timezone.
260	Further, the market index used to calculate unit rates is based in the CET timezone (UTC+1)
261	and so its "day" corresponds to 11pm to 11pm in UK time.
262	Hence, if you query today's unit rates before 4pm, you'll get 46 results back rather than 48.
263	"""
264
265	parameters: MutableMapping[str, Union[str, int]] = {}	1✔
266
267	if period_from is not None:	1✔
268	parameters["period_from"] = period_from.isoformat()	×
269	if period_to is not None:	1✔
270	parameters["period_to"] = period_to.isoformat()	×
271
272	if page_size > 1500:	1✔
273	raise ValueError("'page_size' may not be greater than 1,500")	1✔
274
275	parameters["page_size"] = int(page_size)	1✔
276
277	query_url = self.API_BASE / "products" / product_code / f"{fuel}-tariffs" / tariff_code / str(rate_type)	1✔
278	return PaginatedResponse(query_url, query_params=parameters, obj_type=RateInfo)	1✔
279
280	def get_meter_point_details(self, mpan: str) -> MeterPointDetails:	1✔
281	"""
282	Retrieve the details of a meter-point.
283
284	This can be used to get the GSP of a given meter-point.
285
286	:param mpan: The electricity meter-point's MPAN.
287
288	:return:
289	"""
290
291	return MeterPointDetails._from_dict((self.API_BASE / "electricity-meter-points" / mpan).get())	1✔
292
293	def get_grid_supply_point(self, postcode: str) -> Region:	1✔
294	"""
295	Returns the grid supply point for the given postcode.
296
297	:param postcode:
298
299	:raises: :exc:`ValueError` if the postcode cannot be mapped to a GSP.
300	"""
301
302	query_url = self.API_BASE / "industry" / "grid-supply-points"	1✔
303
304	results = query_url.get(postcode=postcode)["results"]	1✔
305	if results:	1✔
306	return Region(results[0]["group_id"])	1✔
307	else:
308	raise ValueError(f"Cannot map the postcode {postcode!r} to a GSP.")	1✔
309
310	def get_consumption(	1✔
311	self,
312	mpan: str,
313	serial_number: str,
314	fuel: Literal["electricity", "gas"],
315	period_from: Optional[datetime] = None,
316	period_to: Optional[datetime] = None,
317	page_size: int = 100,
318	reverse: bool = False,
319	group_by: Optional[str] = None,
320	) -> PaginatedResponse[Consumption]:
321	r"""
322	Return a list of consumption values for half-hour periods for a given meter-point and meter.
323
324	Unit of measurement:
325
326	* Electricity meters: kWh
327	* SMETS1 Secure gas meters: kWh
328	* SMETS2 gas meters: m\ :superscript:`3`
329
330	.. attention::
331
332	Half-hourly consumption data is only available for smart meters.
333	Requests for consumption data for non-smart meters will return an empty response payload.
334
335	:param mpan: The electricity meter-point's MPAN or gas meter-point's MPRN.
336	:param serial_number: The meter's serial number.
337	:param fuel:
338	:param period_from: Show consumption for periods which start at or after the given datetime.
339	This parameter can be provided on its own.
340	:param period_to: Show consumption for periods which start at or before the given datetime.
341	This parameter also requires providing the ``period_from`` parameter to create a range.
342	:param page_size: Page size of returned results.
343	Default is ``100``, maximum is ``25,000`` to give a full year of half-hourly consumption details.
344	:no-default page_size:
345	:param reverse: Returns the results ordered from most oldest to newest. By default the results are from most recent backwards.
346	:no-default reverse:
347	:param group_by: The grouping of the consumption data.
348	By default the consumption is returned in half-hour periods.
349
350	Possible alternatives are:
351
352	* ``'hour'``
353	* ``'day'``
354	* ``'week'``
355	* ``'month'``
356	* ``'quarter'``
357	:no-default group_by:
358	"""
359
360	parameters: MutableMapping[str, Union[str, int]] = {}	1✔
361
362	if period_from is not None:	1✔
363	parameters["period_from"] = period_from.isoformat()	1✔
364	if period_to is not None:	1✔
365	parameters["period_to"] = period_to.isoformat()	1✔
366
367	if page_size > 25000:	1✔
368	raise ValueError("'page_size' may not be greater than 25,000")	1✔
369
370	parameters["page_size"] = int(page_size)	1✔
371
372	if reverse:	1✔
373	parameters["order_by"] = "period"	1✔
374	if group_by is not None:	1✔
375	parameters["group_by"] = str(group_by)	1✔
376
377	query_url = self.API_BASE / f"{fuel}-meter-points" / mpan / "meters" / serial_number / "consumption"	1✔
378	return PaginatedResponse(query_url, query_params=parameters, obj_type=Consumption)	1✔

domdfcoding / octo-api / 14911143141

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous