124 lines
3.7 KiB
Python
Executable File
124 lines
3.7 KiB
Python
Executable File
# -*- coding: utf-8 -*-
|
|
"""
|
|
Ecowatt Data
|
|
|
|
Configuration parameters:
|
|
name: human-readable name to display
|
|
id: client ID
|
|
token: secret associated with this ID
|
|
"""
|
|
|
|
import datetime
|
|
import requests
|
|
import base64
|
|
import typing as ty
|
|
|
|
|
|
class Py3status:
|
|
API_AUTH: str = "https://digital.iservices.rte-france.com/token/oauth/"
|
|
API_BASE: str = "https://digital.iservices.rte-france.com/open_api/ecowatt/v4"
|
|
API_SIGNALS: str = "/signals"
|
|
UPDATE_EVERY = datetime.timedelta(seconds=3600)
|
|
|
|
TEXT = "\uf0e7\uf21e" # FontAwesome
|
|
|
|
api_id: str
|
|
api_secret: str
|
|
|
|
auth_expires: datetime.datetime
|
|
next_update: datetime.datetime
|
|
bearer_token: str
|
|
|
|
api_data = None
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self.auth_expires = datetime.datetime.fromtimestamp(0) # force re-auth
|
|
self.next_update = self.auth_expires
|
|
self.bearer_token = ""
|
|
|
|
def _authenticate(self):
|
|
"""Authenticate to the OAuth API"""
|
|
auth_str: str = base64.b64encode(
|
|
f"{self.api_id}:{self.api_secret}".encode("utf8")
|
|
).decode("utf8")
|
|
response = requests.post(
|
|
self.API_AUTH, headers={"Authorization": f"Basic {auth_str}"}
|
|
)
|
|
if response.status_code != 200:
|
|
raise Exception(f"Failed to authenticate: {response.body()}")
|
|
json_response = response.json()
|
|
self.bearer_token = (
|
|
f"{json_response['token_type']} {json_response['access_token']}"
|
|
)
|
|
self.auth_expires = datetime.datetime.now() + datetime.timedelta(
|
|
seconds=int(json_response["expires_in"] - 60)
|
|
)
|
|
|
|
def _update(self):
|
|
if datetime.datetime.now() < self.next_update:
|
|
return
|
|
|
|
if datetime.datetime.now() > self.auth_expires:
|
|
self._authenticate()
|
|
|
|
response = requests.get(
|
|
self.API_BASE + self.API_SIGNALS,
|
|
headers={
|
|
"Authorization": self.bearer_token,
|
|
},
|
|
)
|
|
now = datetime.datetime.now()
|
|
if response.status_code == 429:
|
|
print(response.headers)
|
|
self.next_update = now + datetime.timedelta(
|
|
seconds=int(response.headers["Retry-After"]) + 10
|
|
)
|
|
self.api_data = None
|
|
return
|
|
if response.status_code != 200:
|
|
raise Exception(f"Failed to get data: {response.body()}")
|
|
self.api_data = response.json()
|
|
self.next_update = now + self.UPDATE_EVERY
|
|
|
|
def ecowatt(self):
|
|
if None in [self.api_id, self.api_secret]:
|
|
return {"full_text": "Must be configured: api_id, api_secret"}
|
|
self._update()
|
|
|
|
if self.api_data is None:
|
|
return {
|
|
"full_text": f"{self.TEXT} [?]",
|
|
"color": self.py3.COLOR_DEGRADED,
|
|
"urgent": False,
|
|
}
|
|
|
|
day_data = None
|
|
for day in self.api_data["signals"]:
|
|
if (
|
|
datetime.datetime.fromisoformat(day["jour"]).date()
|
|
== datetime.date.today()
|
|
):
|
|
day_data = day
|
|
break
|
|
else:
|
|
raise Exception("Current date not found in data")
|
|
|
|
cur_hour = datetime.datetime.now().hour
|
|
hour_data = day["values"][cur_hour]
|
|
assert int(hour_data["pas"]) == cur_hour
|
|
|
|
alert_val = max(int(hour_data["hvalue"]), int(day_data["dvalue"]))
|
|
|
|
color = self.py3.COLOR_GOOD
|
|
if alert_val == 2:
|
|
color = self.py3.COLOR_DEGRADED
|
|
if alert_val > 2:
|
|
color = self.py3.COLOR_BAD
|
|
|
|
return {
|
|
"full_text": self.TEXT,
|
|
"color": color,
|
|
"urgent": False,
|
|
}
|