-
-
Save cgtobi/c27d4e3bb0e3b812f75582b41f7ff41d to your computer and use it in GitHub Desktop.
Gather pyatmo test data
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import logging | |
import pyatmo | |
logging.basicConfig(level=logging.INFO) | |
LOG = logging.getLogger(__name__) | |
# monkey patch the postRequest function to gather data | |
def new_postRequest(url, params=None, timeout=10): | |
import requests | |
import json | |
def scrmbl(data): | |
keywords = { | |
"password": "my-secret-password", | |
"client_id": "1234a56789b01c2d345ef67a", | |
"client_secret": "TH1S1SMYV3RYPR1V4T3CL13NTS33CR3T", | |
"username": "user.name@mail.com", | |
"access_token": "a1b2c3d4e5f6a7b8c9d0e1f2a30|a09b87c65d43e21f09a87b65c43d2", | |
} | |
output = data | |
# def search(x): | |
# for y in x.keys(): | |
# if isinstance(x[y], str): | |
# for keyword in keywords: | |
# if keyword in data: | |
# output[keyword] = keywords[keyword] | |
for keyword in keywords: | |
if keyword in data: | |
output[keyword] = keywords[keyword] | |
return output | |
resp = requests.post(url, data=params, timeout=timeout) | |
LOG.info("url: %s", url) | |
LOG.info("params: %s", json.dumps(scrmbl(params))) | |
LOG.info("Response:") | |
( | |
LOG.info(json.dumps(scrmbl(resp.json()))) | |
if "application/json" in resp.headers.get("content-type") | |
else LOG.info(scrmbl(resp.content)) | |
) | |
return ( | |
resp.json() | |
if "application/json" in resp.headers.get("content-type") | |
else resp.content | |
) | |
pyatmo.postRequest.__code__ = new_postRequest.__code__ | |
if __name__ == "__main__": | |
import argparse | |
parser = argparse.ArgumentParser() | |
parser.add_argument("--CLIENT_ID") | |
parser.add_argument("--CLIENT_SECRET") | |
parser.add_argument("--USERNAME") | |
parser.add_argument("--PASSWORD") | |
parser.add_argument( | |
"--modules", | |
default="Thermostat,Public,Camera,WeatherStation,HomeCoach", | |
help="Comma separated list of Modules, e.g. " | |
"'Thermostat,Public,Camera,WeatherStation,HomeCoach'", | |
) | |
parser.parse_args() | |
args = parser.parse_args() | |
if args.modules is None: | |
LOG.error("No modules specified") | |
exit(1) | |
modules = args.modules.split(",") | |
if args.CLIENT_ID and args.CLIENT_SECRET and args.USERNAME and args.PASSWORD: | |
pass | |
else: | |
LOG.error( | |
"No credentials passed to pyatmo.py " | |
"(client_id, client_secret, username, password)" | |
) | |
exit(1) | |
scope = ( | |
"read_station read_camera access_camera read_thermostat " | |
"write_thermostat read_presence access_presence read_homecoach" | |
) | |
authorization = pyatmo.ClientAuth( | |
clientId=args.CLIENT_ID, | |
clientSecret=args.CLIENT_SECRET, | |
username=args.USERNAME, | |
password=args.PASSWORD, | |
scope=scope, | |
) | |
if "webhook" in modules: | |
webhook_url = "http://foo.bar" | |
authorization.addwebhook(webhook_url) | |
authorization.dropwebhook() | |
if "WeatherStation" in modules: | |
try: | |
LOG.info("Checking for available weather stations...") | |
LOG.info("<pyatmo.WeatherStation>") | |
LOG.info("<pyatmo.WeatherStationData>") | |
weather_data = pyatmo.WeatherStationData(authorization) | |
sid = weather_data.stationByName()["_id"] | |
mod = weather_data.modulesNamesList()[0] | |
mid = weather_data.moduleByName(mod)["_id"] | |
conds = weather_data.monitoredConditions(mod) | |
if 'temperature' in conds: | |
typ = 'temperature' | |
elif 'windstrength' in conds: | |
typ = 'windstrength' | |
elif 'sum_rain_1' in conds: | |
typ = 'sum_rain_1' | |
else: | |
typ = conds[0] | |
weather_data.getMeasure(sid, 'max', typ, mid) | |
except pyatmo.NoDevice: | |
LOG.error("warning, no weather station available for testing") | |
if "HomeCoach" in modules: | |
try: | |
LOG.info("Checking for available home coach stations...") | |
LOG.info("<pyatmo.HomeCoach>") | |
LOG.info("<pyatmo.HomeCoachData>") | |
home_coach_data = pyatmo.HomeCoachData(authorization) | |
except pyatmo.NoDevice: | |
LOG.error("warning, no home coach station available for testing") | |
if "Camera" in modules: | |
try: | |
LOG.info("Checking for available cameras...") | |
LOG.info("<pyatmo.Camera>") | |
LOG.info("<pyatmo.CameraData>") | |
camera_data = pyatmo.CameraData(authorization) | |
except pyatmo.NoDevice: | |
LOG.error("warning, no camera available for testing") | |
if "Thermostat" in modules: | |
try: | |
LOG.info("Checking for available thermostats/valves...") | |
LOG.info("<pyatmo.Thermostat>") | |
LOG.info("<pyatmo.HomeData>") | |
home_data = pyatmo.HomeData(authorization) | |
LOG.info("<pyatmo.HomeStatus>") | |
home_status = pyatmo.HomeStatus(authorization) | |
except pyatmo.NoDevice: | |
LOG.error("warning, no thermostat available for testing") | |
if "Public" in modules: | |
try: | |
LOG.info("Checking for public data...") | |
LOG.info("<pyatmo.PublicData>") | |
pyatmo.PublicData( | |
authorization, | |
LAT_NE=50.23, | |
LON_NE=8.80, | |
LAT_SW=50.20, | |
LON_SW=8.79, | |
) | |
except pyatmo.NoDevice: | |
LOG.error("no public weather data available for testing") |
With pyatmo > 3.0.0 there should be no pyatmo.py
any more as the module has been restructured. Please check with pip show pyatmo
what version you've installed. My guess is that you're on 2.3.2, so you can either update HA to 0.102 or use an older version of this script, e.g. the revision before.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
When executing the script inside the hassio container (
docker exec -ti homeassistant bash
) it returns:I've tried to set
export PYTHONPATH=/usr/local/lib/python3.7/site-packages
, where is the pyatmo.py file, but it doesn't solve the problem.