pymeteobridgesql/async_test.py at main · briis/pymeteobridgesql · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
"""This module is only used to run some realtime data tests using the async functions,
while developing the module.

Create a .env file and add STATION_ID with the id of your station and API_TOKEN with the personal Token.
"""

from __future__ import annotations

import asyncio
import logging
import os
import time
from pathlib import Path

from dotenv import load_dotenv

from pymeteobridgesql import MeteobridgeSQL

_LOGGER = logging.getLogger(__name__)


async def main() -> None:
    """Async test module."""

    logging.basicConfig(level=logging.DEBUG)
    start = time.time()

    load_dotenv(Path(__file__).parent / ".env")
    _host = os.getenv("HOST") or ""
    _user = os.getenv("USER") or ""
    _password = os.getenv("PASSWORD") or ""
    _database = os.getenv("DATABASE") or ""
    _id = os.getenv("ID") or ""

    missing = [
        k
        for k, v in {
            "HOST": _host,
            "USER": _user,
            "PASSWORD": _password,
            "DATABASE": _database,
            "ID": _id,
        }.items()
        if not v
    ]
    if missing:
        raise ValueError(f"Missing required .env variables: {', '.join(missing)}")

    weather = MeteobridgeSQL(_host, _user, _password, _database)
    await weather.async_init()

    try:
        result = await weather.async_get_realtime_data(_id)
        print("")
        print("========================================================")
        print("ID: ", result.ID)
        print("ABSOLUTE HUMIDITY: ", result.absolute_humidity)
        print("FEELS LIKE: ", result.feels_like_temperature)
        print("FREEZING ALTITUDE: ", result.freezing_altitude)
        print("PM2.5: ", result.pm25)
        print("PM10: ", result.pm10)
        print("AIR QUALITY: ", result.aqi)
        print("TEMPERATURE: ", result.temperature)
        print("RAIN TODAY: ", result.raintoday)
        print("VISIBILITY: ", result.visibility)
        print("WIND BEARING: ", result.windbearing)
        print("WIND BEARING 10 MIN AVG: ", result.windbearingavg10)
        print("WIND BEARING DAY AVG: ", result.windbearingdavg)
        print("WIND DIRECTION: ", result.wind_direction)
        print("WIND GUST: ", result.windgust)
        print("")

    except Exception as err:
        print(err)

    try:
        result = await weather.async_get_station_data(_id)
        print("")
        print("========================================================")
        print("ID: ", result.ID)
        print("IP: ", result.mb_ip)
        print("SW Version: ", result.mb_swversion)
        print("BUILD NUM: ", result.mb_buildnum)
        print("PLATFORM: ", result.mb_platform)
        print("STATION: ", result.mb_station)
        print("STATION NAME: ", result.mb_stationname)
        print("")

    except Exception as err:
        print(err)

    # try:
    #     result = await weather.async_get_forecast(True)
    #     for row in result:
    #         print("")
    #         print("========================================================")
    #         print("DATE: ", row.datetime)
    #         print("TEMP: ", row.temperature)
    #         print("ICON: ", row.icon)
    #         print("")

    # except Exception as err:
    #     print(err)

    # try:
    #     result = await weather.async_get_forecast(False)
    #     for row in result:
    #         print("")
    #         print("========================================================")
    #         print("DATE: ", row.datetime)
    #         print("TEMP: ", row.temperature)
    #         print("ICON: ", row.icon)
    #         print("CONDITION: ", row.description)
    #         print("")

    # except Exception as err:
    #     print(err)

    # try:
    #     result = await weather.async_get_daily_data('6')
    #     for row in result:
    #         print("")
    #         print("========================================================")
    #         print("DATE: ", row.logdate)
    #         print("TEMP: ", row.temperature_high)
    #         print("WIND MAX: ", row.wind_speed_max)
    #         print("")

    # except Exception as err:
    #     print(err)

    await weather.async_disconnect()

    end = time.time()

    _LOGGER.info("Execution time: %s seconds", round(end - start, 3))


asyncio.run(main())