ims_envista.ims_envista.IMSEnvista._get_ims_url() - Code Metrics - Inspection of "Add setuptools to reqs" - GuyKh/py-ims-envista - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — main (#48)

by Guy

created 2024-09-17 10:38 UTC

ims_envista.ims_envista.IMSEnvista._get_ims_url() A

↳ Parent: ims_envista.ims_envista

Complexity

Conditions

Size

Total Lines	29
Code Lines	17

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	17
nop	2
dl	0
loc	29
rs	9.55
c	0
b	0
f	0

"""Module IMSEnvista getting IMS meteorological readings."""

from __future__ import annotations

import asyncio
import atexit
from typing import TYPE_CHECKING

from aiohttp import ClientSession, TraceConfig

from .commons import (
    get,
    on_request_chunk_sent_debug,
    on_request_end_debug,
    on_request_start_debug,
)
from .const import (
    API_NAME,
    API_REGION_ID,
    API_STATIONS,
    GET_ALL_REGIONS_DATA_URL,
    GET_ALL_STATIONS_DATA_URL,
    GET_DAILY_STATION_DATA_URL,
    GET_EARLIEST_STATION_DATA_URL,
    GET_LATEST_STATION_DATA_URL,
    GET_MONTHLY_STATION_DATA_BY_MONTH_URL,
    GET_MONTHLY_STATION_DATA_URL,
    GET_SPECIFIC_REGION_DATA_URL,
    GET_SPECIFIC_STATION_DATA_URL,
    GET_STATION_DATA_BY_DATE_URL,
    GET_STATION_DATA_BY_RANGE_URL,
    VARIABLES,
)
from .meteo_data import (
    StationMeteorologicalReadings,
    station_meteo_data_from_json,
)
from .station_data import RegionInfo, StationInfo, region_from_json, station_from_json

if TYPE_CHECKING:
    from datetime import date
    from uuid import UUID

    from .ims_variable import IMSVariable


class IMSEnvista:
    """API Wrapper to IMS Envista."""

    def __init__(self, token: UUID | str, session: ClientSession | None = None) -> None:
        if not token:
            raise ValueError

        # Custom Logger to the session
        trace_config = TraceConfig()
        trace_config.on_request_start.append(on_request_start_debug)
        trace_config.on_request_chunk_sent.append(on_request_chunk_sent_debug)
        trace_config.on_request_end.append(on_request_end_debug)
        trace_config.freeze()

        if not session:
            session = ClientSession(trace_configs=[trace_config])
            atexit.register(self._shutdown)
        else:
            session.trace_configs.append(trace_config)

        self._session = session
        self._token = token

    def _shutdown(self) -> None:
        if not self._session.closed:
            asyncio.run(self._session.close())

    @staticmethod
    def _get_channel_id_url_part(channel_id: int) -> str:
        """Get specific Channel Id url param."""
        if channel_id:
            return "/" + str(channel_id)
        return ""

    async def get_latest_station_data(
            self, station_id: int, channel_id: int | None = None
        ) -> StationMeteorologicalReadings:
        """
        Fetch the latest station data from IMS Envista API.

        Args:
        ----
            station_id (int): IMS Station Id
            channel_id (int): [Optional] Specific Channel Id

        Returns:
        -------
            data: Current station meteorological data

        """
        get_url = GET_LATEST_STATION_DATA_URL.format(
            str(station_id), self._get_channel_id_url_part(channel_id)
        )
        return station_meteo_data_from_json(await get(session=self._session, url=get_url, token=self._token))

    async def get_earliest_station_data(
            self, station_id: int, channel_id: int | None = None
        ) -> StationMeteorologicalReadings:
        """
        Fetch the earliest station data from IMS Envista API.

        Args:
        ----
            station_id (int): IMS Station ID
            channel_id (int): [Optional] Specific Channel ID

        Returns:
        -------
            data: Current station meteorological data

        """
        get_url = GET_EARLIEST_STATION_DATA_URL.format(
            str(station_id), self._get_channel_id_url_part(channel_id)
        )
        return station_meteo_data_from_json(await get(session=self._session, url=get_url, token=self._token))

    async def get_station_data_from_date(
            self, station_id: int, date_to_query: date, channel_id: int | None = None
        ) -> StationMeteorologicalReadings:
        """
        Fetch latest station data from IMS Envista API by date.

        Args:
        ----
            station_id (int): IMS Station ID
            date_to_query (date): Selected date to query
            channel_id (int): [Optional] Specific Channel Id

        Returns:
        -------
            data: Current station meteorological data

        """
        get_url = GET_STATION_DATA_BY_DATE_URL.format(
            str(station_id),
            self._get_channel_id_url_part(channel_id),
            str(date_to_query.year),
            str(date_to_query.month),
            str(date_to_query.day),
        )
        return station_meteo_data_from_json(await get(session=self._session, url=get_url, token=self._token))

    async def get_station_data_by_date_range(
            self,
            station_id: int,
            from_date: date,
            to_date: date,
            channel_id: int | None = None,
        ) -> StationMeteorologicalReadings:
        """
        Fetch latest station data from IMS Envista API by date range.

        Args:
        ----
            station_id (int): IMS Station ID
            from_date (date): From date to query
            to_date (date): to date to query
            channel_id (int): [Optional] Specific Channel Id

        Returns:
        -------
            data: Current station meteorological data

        """
        get_url = GET_STATION_DATA_BY_RANGE_URL.format(
            str(station_id),
            self._get_channel_id_url_part(channel_id),
            str(from_date.strftime("%Y")),
            str(from_date.strftime("%m")),
            str(from_date.strftime("%d")),
            str(to_date.strftime("%Y")),
            str(to_date.strftime("%m")),
            str(to_date.strftime("%d")),
        )
        return station_meteo_data_from_json(await get(session=self._session, url=get_url, token=self._token))

    async def get_daily_station_data(
            self, station_id: int, channel_id: int | None = None
        ) -> StationMeteorologicalReadings:
        """
        Fetch the daily station data from IMS Envista API.

        Args:
        ----
            station_id (int): IMS Station ID
            channel_id (int): [Optional] Specific Channel Id

        Returns:
        -------
            data: Current station meteorological data

        """
        get_url = GET_DAILY_STATION_DATA_URL.format(
            str(station_id),
            self._get_channel_id_url_part(channel_id),
        )
        return station_meteo_data_from_json(await get(session=self._session, url=get_url, token=self._token))

    async def get_monthly_station_data(
            self,
            station_id: int,
            channel_id: int | None = None,
            month: str | None = None,
            year: str | None = None,
        ) -> StationMeteorologicalReadings:
        """
        Fetch monthly station data from IMS Envista API.

        Args:
        ----
            station_id (int): IMS Station ID
            channel_id (int): [Optional] Specific Channel Id
            month (str): [Optional] Specific Month in MM format (07)
            year (str):  [Optional] Specific Year in YYYY format (2020)

        Returns:
        -------
            data: Current station meteorological data

        """
        if not month or not year:
            get_url = GET_MONTHLY_STATION_DATA_URL.format(
                str(station_id), self._get_channel_id_url_part(channel_id)
            )
        else:
            get_url = GET_MONTHLY_STATION_DATA_BY_MONTH_URL.format(
                str(station_id), self._get_channel_id_url_part(channel_id), year, month
            )
        return station_meteo_data_from_json(await get(session=self._session, url=get_url, token=self._token))

    async def get_all_stations_info(self) -> list[StationInfo]:
        """
        Fetch all stations data from IMS Envista API.

        Returns
        -------
            data: All stations data

        """
        get_url = GET_ALL_STATIONS_DATA_URL
        response = await get(session=self._session, url=get_url, token=self._token)
        return [station_from_json(station) for station in response]

    async def get_station_info(self, station_id: int) -> StationInfo:
        """
        Fetch station data from IMS Envista API.

        Args:
        ----
            station_id (int): IMS Station ID

        Returns:
        -------
            data: Current station data

        """
        get_url = GET_SPECIFIC_STATION_DATA_URL.format(str(station_id))
        return station_from_json(await get(session=self._session, url=get_url, token=self._token))

    async def get_all_regions_info(self) -> list[RegionInfo]:
        """
        Fetch all regions data from IMS Envista API.

        Returns
        -------
            data: All stations data

        """
        get_url = GET_ALL_REGIONS_DATA_URL
        response = await get(session=self._session, url=get_url, token=self._token)
        regions = []
        for region in response:
            stations = [station_from_json(station) for station in region[API_STATIONS]]
            regions.append(
                RegionInfo(region[API_REGION_ID], region[API_NAME], stations)
            )
        return regions

    async def get_region_info(self, region_id: int) -> RegionInfo:
        """
        Fetch region data from IMS Envista API.

        Args:
        ----
            region_id (int): IMS Region ID

        Returns:
        -------
            data: region data

        """
        get_url = GET_SPECIFIC_REGION_DATA_URL.format(str(region_id))
        response = await get(session=self._session, url=get_url, token=self._token)
        return region_from_json(response)

    def get_metrics_descriptions(self) -> list[IMSVariable]:
        """
        Return the descriptions of Meteorological Metrics collected by the stations.

        Returns
        -------
            list of IMSVariable, containing description and measuring unit

        """
        return list(VARIABLES.values())


1			"""Module IMSEnvista getting IMS meteorological readings."""
2
3			from __future__ import annotations
4
5			import asyncio
6			import atexit
7			from typing import TYPE_CHECKING
8
9			from aiohttp import ClientSession, TraceConfig
10
11			from .commons import (
12			get,
13			on_request_chunk_sent_debug,
14			on_request_end_debug,
15			on_request_start_debug,
16			)
17			from .const import (
18			API_NAME,
19			API_REGION_ID,
20			API_STATIONS,
21			GET_ALL_REGIONS_DATA_URL,
22			GET_ALL_STATIONS_DATA_URL,
23			GET_DAILY_STATION_DATA_URL,
24			GET_EARLIEST_STATION_DATA_URL,
25			GET_LATEST_STATION_DATA_URL,
26			GET_MONTHLY_STATION_DATA_BY_MONTH_URL,
27			GET_MONTHLY_STATION_DATA_URL,
28			GET_SPECIFIC_REGION_DATA_URL,
29			GET_SPECIFIC_STATION_DATA_URL,
30			GET_STATION_DATA_BY_DATE_URL,
31			GET_STATION_DATA_BY_RANGE_URL,
32			VARIABLES,
33			)
34			from .meteo_data import (
35			StationMeteorologicalReadings,
36			station_meteo_data_from_json,
37			)
38			from .station_data import RegionInfo, StationInfo, region_from_json, station_from_json
39
40			if TYPE_CHECKING:
41			from datetime import date
42			from uuid import UUID
43
44			from .ims_variable import IMSVariable
45
46
47			class IMSEnvista:
48			"""API Wrapper to IMS Envista."""
49
50			def __init__(self, token: UUID \| str, session: ClientSession \| None = None) -> None:
51			if not token:
52			raise ValueError
53
54			# Custom Logger to the session
55			trace_config = TraceConfig()
56			trace_config.on_request_start.append(on_request_start_debug)
57			trace_config.on_request_chunk_sent.append(on_request_chunk_sent_debug)
58			trace_config.on_request_end.append(on_request_end_debug)
59			trace_config.freeze()
60
61			if not session:
62			session = ClientSession(trace_configs=[trace_config])
63			atexit.register(self._shutdown)
64			else:
65			session.trace_configs.append(trace_config)
66
67			self._session = session
68			self._token = token
69
70			def _shutdown(self) -> None:
71			if not self._session.closed:
72			asyncio.run(self._session.close())
73
74			@staticmethod
75			def _get_channel_id_url_part(channel_id: int) -> str:
76			"""Get specific Channel Id url param."""
77			if channel_id:
78			return "/" + str(channel_id)
79			return ""
80
81			async def get_latest_station_data(
82			self, station_id: int, channel_id: int \| None = None
83			) -> StationMeteorologicalReadings:
84			"""
85			Fetch the latest station data from IMS Envista API.
86
87			Args:
88			----
89			station_id (int): IMS Station Id
90			channel_id (int): [Optional] Specific Channel Id
91
92			Returns:
93			-------
94			data: Current station meteorological data
95
96			"""
97			get_url = GET_LATEST_STATION_DATA_URL.format(
98			str(station_id), self._get_channel_id_url_part(channel_id)
99			)
100			return station_meteo_data_from_json(await get(session=self._session, url=get_url, token=self._token))
101
102			async def get_earliest_station_data(
103			self, station_id: int, channel_id: int \| None = None
104			) -> StationMeteorologicalReadings:
105			"""
106			Fetch the earliest station data from IMS Envista API.
107
108			Args:
109			----
110			station_id (int): IMS Station ID
111			channel_id (int): [Optional] Specific Channel ID
112
113			Returns:
114			-------
115			data: Current station meteorological data
116
117			"""
118			get_url = GET_EARLIEST_STATION_DATA_URL.format(
119			str(station_id), self._get_channel_id_url_part(channel_id)
120			)
121			return station_meteo_data_from_json(await get(session=self._session, url=get_url, token=self._token))
122
123			async def get_station_data_from_date(
124			self, station_id: int, date_to_query: date, channel_id: int \| None = None
125			) -> StationMeteorologicalReadings:
126			"""
127			Fetch latest station data from IMS Envista API by date.
128
129			Args:
130			----
131			station_id (int): IMS Station ID
132			date_to_query (date): Selected date to query
133			channel_id (int): [Optional] Specific Channel Id
134
135			Returns:
136			-------
137			data: Current station meteorological data
138
139			"""
140			get_url = GET_STATION_DATA_BY_DATE_URL.format(
141			str(station_id),
142			self._get_channel_id_url_part(channel_id),
143			str(date_to_query.year),
144			str(date_to_query.month),
145			str(date_to_query.day),
146			)
147			return station_meteo_data_from_json(await get(session=self._session, url=get_url, token=self._token))
148
149			async def get_station_data_by_date_range(
150			self,
151			station_id: int,
152			from_date: date,
153			to_date: date,
154			channel_id: int \| None = None,
155			) -> StationMeteorologicalReadings:
156			"""
157			Fetch latest station data from IMS Envista API by date range.
158
159			Args:
160			----
161			station_id (int): IMS Station ID
162			from_date (date): From date to query
163			to_date (date): to date to query
164			channel_id (int): [Optional] Specific Channel Id
165
166			Returns:
167			-------
168			data: Current station meteorological data
169
170			"""
171			get_url = GET_STATION_DATA_BY_RANGE_URL.format(
172			str(station_id),
173			self._get_channel_id_url_part(channel_id),
174			str(from_date.strftime("%Y")),
175			str(from_date.strftime("%m")),
176			str(from_date.strftime("%d")),
177			str(to_date.strftime("%Y")),
178			str(to_date.strftime("%m")),
179			str(to_date.strftime("%d")),
180			)
181			return station_meteo_data_from_json(await get(session=self._session, url=get_url, token=self._token))
182
183			async def get_daily_station_data(
184			self, station_id: int, channel_id: int \| None = None
185			) -> StationMeteorologicalReadings:
186			"""
187			Fetch the daily station data from IMS Envista API.
188
189			Args:
190			----
191			station_id (int): IMS Station ID
192			channel_id (int): [Optional] Specific Channel Id
193
194			Returns:
195			-------
196			data: Current station meteorological data
197
198			"""
199			get_url = GET_DAILY_STATION_DATA_URL.format(
200			str(station_id),
201			self._get_channel_id_url_part(channel_id),
202			)
203			return station_meteo_data_from_json(await get(session=self._session, url=get_url, token=self._token))
204
205			async def get_monthly_station_data(
206			self,
207			station_id: int,
208			channel_id: int \| None = None,
209			month: str \| None = None,
210			year: str \| None = None,
211			) -> StationMeteorologicalReadings:
212			"""
213			Fetch monthly station data from IMS Envista API.
214
215			Args:
216			----
217			station_id (int): IMS Station ID
218			channel_id (int): [Optional] Specific Channel Id
219			month (str): [Optional] Specific Month in MM format (07)
220			year (str): [Optional] Specific Year in YYYY format (2020)
221
222			Returns:
223			-------
224			data: Current station meteorological data
225
226			"""
227			if not month or not year:
228			get_url = GET_MONTHLY_STATION_DATA_URL.format(
229			str(station_id), self._get_channel_id_url_part(channel_id)
230			)
231			else:
232			get_url = GET_MONTHLY_STATION_DATA_BY_MONTH_URL.format(
233			str(station_id), self._get_channel_id_url_part(channel_id), year, month
234			)
235			return station_meteo_data_from_json(await get(session=self._session, url=get_url, token=self._token))
236
237			async def get_all_stations_info(self) -> list[StationInfo]:
238			"""
239			Fetch all stations data from IMS Envista API.
240
241			Returns
242			-------
243			data: All stations data
244
245			"""
246			get_url = GET_ALL_STATIONS_DATA_URL
247			response = await get(session=self._session, url=get_url, token=self._token)
248			return [station_from_json(station) for station in response]
249
250			async def get_station_info(self, station_id: int) -> StationInfo:
251			"""
252			Fetch station data from IMS Envista API.
253
254			Args:
255			----
256			station_id (int): IMS Station ID
257
258			Returns:
259			-------
260			data: Current station data
261
262			"""
263			get_url = GET_SPECIFIC_STATION_DATA_URL.format(str(station_id))
264			return station_from_json(await get(session=self._session, url=get_url, token=self._token))
265
266			async def get_all_regions_info(self) -> list[RegionInfo]:
267			"""
268			Fetch all regions data from IMS Envista API.
269
270			Returns
271			-------
272			data: All stations data
273
274			"""
275			get_url = GET_ALL_REGIONS_DATA_URL
276			response = await get(session=self._session, url=get_url, token=self._token)
277			regions = []
278			for region in response:
279			stations = [station_from_json(station) for station in region[API_STATIONS]]
280			regions.append(
281			RegionInfo(region[API_REGION_ID], region[API_NAME], stations)
282			)
283			return regions
284
285			async def get_region_info(self, region_id: int) -> RegionInfo:
286			"""
287			Fetch region data from IMS Envista API.
288
289			Args:
290			----
291			region_id (int): IMS Region ID
292
293			Returns:
294			-------
295			data: region data
296
297			"""
298			get_url = GET_SPECIFIC_REGION_DATA_URL.format(str(region_id))
299			response = await get(session=self._session, url=get_url, token=self._token)
300			return region_from_json(response)
301
302			def get_metrics_descriptions(self) -> list[IMSVariable]:
303			"""
304			Return the descriptions of Meteorological Metrics collected by the stations.
305
306			Returns
307			-------
308			list of IMSVariable, containing description and measuring unit
309
310			"""
311			return list(VARIABLES.values())
312

GuyKh / py-ims-envista

Pull Request — main (#48)

ims_envista.ims_envista.IMSEnvista._get_ims_url() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like