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

ims_envista.meteo_data.meteo_data_from_json() C
last analyzed 2025-08-11 17:39 UTC

↳ Parent: ims_envista.meteo_data

Complexity

Conditions

Size

Total Lines	68
Code Lines	60

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	8
eloc	60
nop	2
dl	0
loc	68
rs	6.4424
c	0
b	0
f	0

How to fix Long Method

"""Data Class for IMS Meteorological Readings."""

from __future__ import annotations

import datetime
import logging
import textwrap
import time
from dataclasses import dataclass, field

import pytz

from .const import (
    API_BP,
    API_CHANNELS,
    API_DATA,
    API_DATETIME,
    API_DIFF,
    API_GRAD,
    API_NAME,
    API_NIP,
    API_RAIN,
    API_RAIN_1_MIN,
    API_RH,
    API_STATION_ID,
    API_STATUS,
    API_STD_WD,
    API_TD,
    API_TD_MAX,
    API_TD_MIN,
    API_TG,
    API_TIME,
    API_TW,
    API_VALID,
    API_VALUE,
    API_WD,
    API_WD_MAX,
    API_WS,
    API_WS_1MM,
    API_WS_10MM,
    API_WS_MAX,
    VARIABLES,
)

_LOGGER = logging.getLogger(__name__)
MAX_HOUR_INT = 60

@dataclass
class MeteorologicalData:
    """Meteorological Data."""

    station_id: int
    """Station ID"""
    datetime: datetime.datetime

    """Date and time of the data"""
    rain: float
    """Rainfall in mm"""
    ws: float
    """Wind speed in m/s"""
    ws_max: float
    """Gust wind speed in m/s"""
    wd: float
    """Wind direction in deg"""
    wd_max: float
    """Gust wind direction in deg"""
    std_wd: float
    """Standard deviation wind direction in deg"""
    td: float
    """Temperature in °C"""
    td_max: float
    """Maximum Temperature in °C"""
    td_min: float
    """Minimum Temperature in °C"""
    tg: float
    """Ground Temperature in °C"""
    tw: float
    """TW Temperature (?) in °C"""
    rh: float
    """Relative humidity in %"""
    ws_1mm: float
    """Maximum 1 minute wind speed in m/s"""
    ws_10mm: float
    """Maximum 10 minute wind speed in m/s"""
    time: datetime.time
    """Time"""
    bp: float
    """Maximum barometric pressure in mb"""
    diff_r: float
    """Distributed radiation in w/m^2"""
    grad: float
    """Global radiation in w/m^2"""
    nip: float
    """Direct radiation in w/m^2"""
    rain_1_min: float
    """Rainfall per minute in mm"""

    def _prety_print_field(self, value: float | str | None, unit: str | None) -> str:
        """Pretty Print a specific field."""
        if value:
            return f"{value!s}{unit if unit else ''}"
        return "None"

    def _pretty_print(self) -> str:
        """Pretty Print."""
        return (
                f"StationID: {self._prety_print_field(self.station_id, None)}, "
                f"Date: {self._prety_print_field(self.datetime, None)}, "
                f"Readings: ["
                f"(TD: {self._prety_print_field(self.td, VARIABLES[API_TD].unit)}), "
                f"(TDmax: {self._prety_print_field(self.td_max, VARIABLES[API_TD_MAX].unit)}), "
                f"(TDmin: {self._prety_print_field(self.td_min, VARIABLES[API_TD_MIN].unit)}), "
                f"(TG: {self._prety_print_field(self.tg, VARIABLES[API_TG].unit)}), "
                f"(RH: {self._prety_print_field(self.rh, VARIABLES[API_RH].unit)}), "
                f"(Rain: {self._prety_print_field(self.rain, VARIABLES[API_RAIN].unit)}), "
                f"(WS: {self._prety_print_field(self.ws, VARIABLES[API_WS].unit)}), "
                f"(WSmax: {self._prety_print_field(self.ws_max, VARIABLES[API_WS_MAX].unit)}), "
                f"(WD: {self._prety_print_field(self.wd, VARIABLES[API_WD].unit)}), "
                f"(WDmax: {self._prety_print_field(self.wd_max, VARIABLES[API_WD_MAX].unit)}), "
                f"(STDwd: {self._prety_print_field(self.std_wd, VARIABLES[API_STD_WD].unit)}), "
                f"(WS1mm: {self._prety_print_field(self.ws_1mm, VARIABLES[API_WS_1MM].unit)}), "
                f"(WS10mm: {self._prety_print_field(self.ws_10mm, VARIABLES[API_WS_10MM].unit)}), "
                f"(Time: {self._prety_print_field(self.time.strftime('%H:%M') if self.time else None, VARIABLES[API_TIME].unit)})]"
            )

    def __str__(self) -> str:
        return self._pretty_print()

    def __repr__(self) -> str:
        return self._pretty_print().replace("\n", " ")


@dataclass
class StationMeteorologicalReadings:
    """Station Meteorological Readings."""

    station_id: int
    """ Station Id"""
    data: list[MeteorologicalData] = field(default_factory=list)
    """ List of Meteorological Data """

    def __repr__(self) -> str:
        return textwrap.dedent("""Station ({}), Data: {}""").format(
            self.station_id, self.data
        )

tz = pytz.timezone("Asia/Jerusalem")


def _fix_datetime_offset(dt: datetime.datetime) -> tuple[datetime.datetime, bool]:
    dt = dt.replace(tzinfo=None)
    dt = tz.localize(dt)

    # Get the UTC offset in seconds
    offset_seconds = dt.utcoffset().total_seconds()

    # Create a fixed timezone with the same offset and name
    fixed_timezone = datetime.timezone(datetime.timedelta(seconds=offset_seconds), dt.tzname())

    # Replace the pytz tzinfo with the fixed timezone
    dt = dt.replace(tzinfo=fixed_timezone)

    is_dst = dt.dst() and dt.dst() != datetime.timedelta(0)
    if is_dst:
        dt = dt + datetime.timedelta(hours=1)

    return dt,is_dst


def meteo_data_from_json(station_id: int, data: dict) -> MeteorologicalData:
    """Create a MeteorologicalData object from a JSON object."""
    dt = datetime.datetime.fromisoformat(data[API_DATETIME])
    dt, is_dst = _fix_datetime_offset(dt)

    channel_value_dict = {}
    for channel_value in data[API_CHANNELS]:
        if channel_value[API_VALID] is True and channel_value[API_STATUS] == 1:
            channel_value_dict[channel_value[API_NAME]] = float(
                channel_value[API_VALUE]
            )

    rain = channel_value_dict.get(API_RAIN)
    ws_max = channel_value_dict.get(API_WS_MAX)
    wd_max = channel_value_dict.get(API_WD_MAX)
    ws = channel_value_dict.get(API_WS)
    wd = channel_value_dict.get(API_WD)
    std_wd = channel_value_dict.get(API_STD_WD)
    td = channel_value_dict.get(API_TD)
    rh = channel_value_dict.get(API_RH)
    td_max = channel_value_dict.get(API_TD_MAX)
    td_min = channel_value_dict.get(API_TD_MIN)
    ws_1mm = channel_value_dict.get(API_WS_1MM)
    ws_10mm = channel_value_dict.get(API_WS_10MM)
    tg = channel_value_dict.get(API_TG)
    tw = channel_value_dict.get(API_TW)
    time_val = channel_value_dict.get(API_TIME)
    if time_val:
        time_int = int(time_val)
        if time_int <= MAX_HOUR_INT:
            t = time.strptime(str(time_int), "%M")
        else :
            t = time.strptime(str(time_int), "%H%M")
        time_val = datetime.time(t.tm_hour, t.tm_min, tzinfo=tz)
    bp = channel_value_dict.get(API_BP)
    diff_r = channel_value_dict.get(API_DIFF)
    grad = channel_value_dict.get(API_GRAD)
    nip = channel_value_dict.get(API_NIP)
    rain_1_min = channel_value_dict.get(API_RAIN_1_MIN)

    if is_dst and time_val:
        # Strange IMS logic :o
        dt = dt + datetime.timedelta(hours=1)
        time_val = time_val.replace(hour=(time_val.hour+1)%24)

    return MeteorologicalData(
        station_id=station_id,
        datetime=dt,
        rain=rain,
        ws=ws,
        ws_max=ws_max,
        wd=wd,
        wd_max=wd_max,
        std_wd=std_wd,
        td=td,
        td_max=td_max,
        td_min=td_min,
        tg=tg,
        tw=tw,
        rh=rh,
        ws_1mm=ws_1mm,
        ws_10mm=ws_10mm,
        time=time_val,
        bp=bp,
        diff_r=diff_r,
        grad=grad,
        nip=nip,
        rain_1_min=rain_1_min
    )


def station_meteo_data_from_json(json: dict) -> StationMeteorologicalReadings | None:
    station_id = int(json[API_STATION_ID])
    data = json.get(API_DATA)
    if not data:
        return None
    meteo_data = [meteo_data_from_json(station_id, single_meteo_data) for single_meteo_data in data]
    return StationMeteorologicalReadings(station_id, meteo_data)


1			"""Data Class for IMS Meteorological Readings."""
2
3			from __future__ import annotations
4
5			import datetime
6			import logging
7			import textwrap
8			import time
9			from dataclasses import dataclass, field
10
11			import pytz
12
13			from .const import (
14			API_BP,
15			API_CHANNELS,
16			API_DATA,
17			API_DATETIME,
18			API_DIFF,
19			API_GRAD,
20			API_NAME,
21			API_NIP,
22			API_RAIN,
23			API_RAIN_1_MIN,
24			API_RH,
25			API_STATION_ID,
26			API_STATUS,
27			API_STD_WD,
28			API_TD,
29			API_TD_MAX,
30			API_TD_MIN,
31			API_TG,
32			API_TIME,
33			API_TW,
34			API_VALID,
35			API_VALUE,
36			API_WD,
37			API_WD_MAX,
38			API_WS,
39			API_WS_1MM,
40			API_WS_10MM,
41			API_WS_MAX,
42			VARIABLES,
43			)
44
45			_LOGGER = logging.getLogger(__name__)
46			MAX_HOUR_INT = 60
47
48			@dataclass
49			class MeteorologicalData:
50			"""Meteorological Data."""
51
52			station_id: int
53			"""Station ID"""
54			datetime: datetime.datetime
			0 ignored issues – show Comprehensibility Best Practice introduced 2023-03-01 09:08 UTC by Report Bug Copy Issue Report The variable `datetime` does not seem to be defined. Loading history...
55			"""Date and time of the data"""
56			rain: float
57			"""Rainfall in mm"""
58			ws: float
59			"""Wind speed in m/s"""
60			ws_max: float
61			"""Gust wind speed in m/s"""
62			wd: float
63			"""Wind direction in deg"""
64			wd_max: float
65			"""Gust wind direction in deg"""
66			std_wd: float
67			"""Standard deviation wind direction in deg"""
68			td: float
69			"""Temperature in °C"""
70			td_max: float
71			"""Maximum Temperature in °C"""
72			td_min: float
73			"""Minimum Temperature in °C"""
74			tg: float
75			"""Ground Temperature in °C"""
76			tw: float
77			"""TW Temperature (?) in °C"""
78			rh: float
79			"""Relative humidity in %"""
80			ws_1mm: float
81			"""Maximum 1 minute wind speed in m/s"""
82			ws_10mm: float
83			"""Maximum 10 minute wind speed in m/s"""
84			time: datetime.time
85			"""Time"""
86			bp: float
87			"""Maximum barometric pressure in mb"""
88			diff_r: float
89			"""Distributed radiation in w/m^2"""
90			grad: float
91			"""Global radiation in w/m^2"""
92			nip: float
93			"""Direct radiation in w/m^2"""
94			rain_1_min: float
95			"""Rainfall per minute in mm"""
96
97			def _prety_print_field(self, value: float \| str \| None, unit: str \| None) -> str:
98			"""Pretty Print a specific field."""
99			if value:
100			return f"{value!s}{unit if unit else ''}"
101			return "None"
102
103			def _pretty_print(self) -> str:
104			"""Pretty Print."""
105			return (
106			f"StationID: {self._prety_print_field(self.station_id, None)}, "
107			f"Date: {self._prety_print_field(self.datetime, None)}, "
108			f"Readings: ["
109			f"(TD: {self._prety_print_field(self.td, VARIABLES[API_TD].unit)}), "
110			f"(TDmax: {self._prety_print_field(self.td_max, VARIABLES[API_TD_MAX].unit)}), "
111			f"(TDmin: {self._prety_print_field(self.td_min, VARIABLES[API_TD_MIN].unit)}), "
112			f"(TG: {self._prety_print_field(self.tg, VARIABLES[API_TG].unit)}), "
113			f"(RH: {self._prety_print_field(self.rh, VARIABLES[API_RH].unit)}), "
114			f"(Rain: {self._prety_print_field(self.rain, VARIABLES[API_RAIN].unit)}), "
115			f"(WS: {self._prety_print_field(self.ws, VARIABLES[API_WS].unit)}), "
116			f"(WSmax: {self._prety_print_field(self.ws_max, VARIABLES[API_WS_MAX].unit)}), "
117			f"(WD: {self._prety_print_field(self.wd, VARIABLES[API_WD].unit)}), "
118			f"(WDmax: {self._prety_print_field(self.wd_max, VARIABLES[API_WD_MAX].unit)}), "
119			f"(STDwd: {self._prety_print_field(self.std_wd, VARIABLES[API_STD_WD].unit)}), "
120			f"(WS1mm: {self._prety_print_field(self.ws_1mm, VARIABLES[API_WS_1MM].unit)}), "
121			f"(WS10mm: {self._prety_print_field(self.ws_10mm, VARIABLES[API_WS_10MM].unit)}), "
122			f"(Time: {self._prety_print_field(self.time.strftime('%H:%M') if self.time else None, VARIABLES[API_TIME].unit)})]"
123			)
124
125			def __str__(self) -> str:
126			return self._pretty_print()
127
128			def __repr__(self) -> str:
129			return self._pretty_print().replace("\n", " ")
130
131
132			@dataclass
133			class StationMeteorologicalReadings:
134			"""Station Meteorological Readings."""
135
136			station_id: int
137			""" Station Id"""
138			data: list[MeteorologicalData] = field(default_factory=list)
139			""" List of Meteorological Data """
140
141			def __repr__(self) -> str:
142			return textwrap.dedent("""Station ({}), Data: {}""").format(
143			self.station_id, self.data
144			)
145
146			tz = pytz.timezone("Asia/Jerusalem")
147
148
149			def _fix_datetime_offset(dt: datetime.datetime) -> tuple[datetime.datetime, bool]:
150			dt = dt.replace(tzinfo=None)
151			dt = tz.localize(dt)
152
153			# Get the UTC offset in seconds
154			offset_seconds = dt.utcoffset().total_seconds()
155
156			# Create a fixed timezone with the same offset and name
157			fixed_timezone = datetime.timezone(datetime.timedelta(seconds=offset_seconds), dt.tzname())
158
159			# Replace the pytz tzinfo with the fixed timezone
160			dt = dt.replace(tzinfo=fixed_timezone)
161
162			is_dst = dt.dst() and dt.dst() != datetime.timedelta(0)
163			if is_dst:
164			dt = dt + datetime.timedelta(hours=1)
165
166			return dt,is_dst
167
168
169			def meteo_data_from_json(station_id: int, data: dict) -> MeteorologicalData:
170			"""Create a MeteorologicalData object from a JSON object."""
171			dt = datetime.datetime.fromisoformat(data[API_DATETIME])
172			dt, is_dst = _fix_datetime_offset(dt)
173
174			channel_value_dict = {}
175			for channel_value in data[API_CHANNELS]:
176			if channel_value[API_VALID] is True and channel_value[API_STATUS] == 1:
177			channel_value_dict[channel_value[API_NAME]] = float(
178			channel_value[API_VALUE]
179			)
180
181			rain = channel_value_dict.get(API_RAIN)
182			ws_max = channel_value_dict.get(API_WS_MAX)
183			wd_max = channel_value_dict.get(API_WD_MAX)
184			ws = channel_value_dict.get(API_WS)
185			wd = channel_value_dict.get(API_WD)
186			std_wd = channel_value_dict.get(API_STD_WD)
187			td = channel_value_dict.get(API_TD)
188			rh = channel_value_dict.get(API_RH)
189			td_max = channel_value_dict.get(API_TD_MAX)
190			td_min = channel_value_dict.get(API_TD_MIN)
191			ws_1mm = channel_value_dict.get(API_WS_1MM)
192			ws_10mm = channel_value_dict.get(API_WS_10MM)
193			tg = channel_value_dict.get(API_TG)
194			tw = channel_value_dict.get(API_TW)
195			time_val = channel_value_dict.get(API_TIME)
196			if time_val:
197			time_int = int(time_val)
198			if time_int <= MAX_HOUR_INT:
199			t = time.strptime(str(time_int), "%M")
200			else :
201			t = time.strptime(str(time_int), "%H%M")
202			time_val = datetime.time(t.tm_hour, t.tm_min, tzinfo=tz)
203			bp = channel_value_dict.get(API_BP)
204			diff_r = channel_value_dict.get(API_DIFF)
205			grad = channel_value_dict.get(API_GRAD)
206			nip = channel_value_dict.get(API_NIP)
207			rain_1_min = channel_value_dict.get(API_RAIN_1_MIN)
208
209			if is_dst and time_val:
210			# Strange IMS logic :o
211			dt = dt + datetime.timedelta(hours=1)
212			time_val = time_val.replace(hour=(time_val.hour+1)%24)
213
214			return MeteorologicalData(
215			station_id=station_id,
216			datetime=dt,
217			rain=rain,
218			ws=ws,
219			ws_max=ws_max,
220			wd=wd,
221			wd_max=wd_max,
222			std_wd=std_wd,
223			td=td,
224			td_max=td_max,
225			td_min=td_min,
226			tg=tg,
227			tw=tw,
228			rh=rh,
229			ws_1mm=ws_1mm,
230			ws_10mm=ws_10mm,
231			time=time_val,
232			bp=bp,
233			diff_r=diff_r,
234			grad=grad,
235			nip=nip,
236			rain_1_min=rain_1_min
237			)
238
239
240			def station_meteo_data_from_json(json: dict) -> StationMeteorologicalReadings \| None:
241			station_id = int(json[API_STATION_ID])
242			data = json.get(API_DATA)
243			if not data:
244			return None
245			meteo_data = [meteo_data_from_json(station_id, single_meteo_data) for single_meteo_data in data]
246			return StationMeteorologicalReadings(station_id, meteo_data)
247

GuyKh / py-ims-envista

ims_envista.meteo_data.meteo_data_from_json() C last analyzed 2025-08-11 17:39 UTC

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like

ims_envista.meteo_data.meteo_data_from_json() C
last analyzed 2025-08-11 17:39 UTC