Working example implementation for this library

[Lock128]

Hey

do we somewhere have a "working" piece of code for this repository - so what are the 5 lines of code that we need to use the library?

e.g. how to I use async_update_weather_stations?

In my current code I had been using

authorization = pyatmo.NetatmoOAuth2(
    client_id=CLIENT_ID,
    client_secret=CLIENT_SECRET,
   # scope="read_station",
)

authorization.extra["refresh_token"]="xxx"
authorization.refresh_tokens()

# 2 : Get devices list
weatherData = pyatmo.WeatherStationData(authorization)

but somehow pyatmo.WeatherStationData does not show any stations anymore...

What's the adviced implementation way right now? Do I need to switch to async_update_weather_stations?

[cgtobi]

Yes, you have to use async_account.async_update_weather_stations

Here is a rough sketch:

account = pyatmo.AsyncAccount(auth)
await account.async_update_topology()
await account.async_update_weather_stations()
async_account.homes[home_id].modules

Check

pyatmo/tests/test_weather.py

Line 21 in 4296c98

async def test_async_weather_update(async_account):

for more details

I hope this helps a bit.

[stefano-garzarella]

@cgtobi any suggestions on what to use to handle refresh tokens with the new async API?

[stefano-garzarella]

@cgtobi any suggestions on what to use to handle refresh tokens with the new async API?

I just used OAuth2Session from requests_oauthlib. I'm pasting here, so it may be useful for someone else.

import os
import time
import json
from requests_oauthlib import OAuth2Session

class NetatmoAuth(AbstractAsyncAuth):
    client_id = 'YOUR_CLIENT_ID'
    client_secret = 'YOUR_CLIENT_SECRET'
    # used only the first time when the token file is not there
    # if you generate a new token with new grants, remove the file and put your first refresh token here
    initial_refresh_token = 'YOUR_FIRST_REFRESH_TOKEN_RELEASED__BY_UI'
    token_url = 'https://api.netatmo.com/oauth2/token'
    refresh_token_file = '/data/refresh_token.json'
    token_expiry_threshold = 300  # seconds

    def __init__(self, websession: ClientSession):
        super().__init__(websession)
        self.oauth = OAuth2Session(self.client_id)

    def save_token(self):
        token = self.oauth.token
        if 'expires_at' not in token:
            token['expires_at'] = int(time.time()) + token['expires_in']
        with open(self.refresh_token_file, 'w') as f:
            json.dump(token, f)

    def load_token(self):
        if os.path.exists(self.refresh_token_file):
            with open(self.refresh_token_file, 'r') as f:
                token = json.load(f)
                self.oauth.token = token
        else:
            print("Token not found, asking it...")
            token = self.oauth.refresh_token(
                self.token_url,
                refresh_token=self.initial_refresh_token,
                client_id=self.client_id,
                client_secret=self.client_secret
            )
            self.save_token()

    def is_token_expired(self):
        return self.oauth.token['expires_at'] - time.time() < self.token_expiry_threshold

    def get_token(self):
        if self.oauth.access_token is None:
            self.load_token()

        if self.is_token_expired():
            print("Token expiring, refreshing it...")
            token = self.oauth.refresh_token(
                self.token_url,
                client_id=self.client_id,
                client_secret=self.client_secret
            )
            self.save_token()

        return self.oauth.access_token

    async def async_get_access_token(self):
        return self.get_token()

[cgtobi]

Thanks for sharing @stefano-garzarella