ims_envista.ims_envista.IMSEnvista._get_channel_id_url_part() - Code Metrics - GuyKh/py-ims-envista - Measure and Improve Code Quality continuously with Scrutinizer

IMSEnvista._get_channel_id_url_part() A
last analyzed 2025-11-10 17:14 UTC

↳ Parent: ims_envista.ims_envista

Complexity

Conditions

Size

Total Lines	6
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	5
nop	1
dl	0
loc	6
rs	10
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

from .commons import get
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:
            err_msg = "Missing IMS Token"
            raise ValueError(err_msg)

        if not session:
            session = ClientSession()
            atexit.register(self._shutdown)

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

GuyKh / py-ims-envista

IMSEnvista._get_channel_id_url_part() A last analyzed 2025-11-10 17:14 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

IMSEnvista._get_channel_id_url_part() A
last analyzed 2025-11-10 17:14 UTC