class FairwaterAPIClient:
def __init__(self, base_url, api_key):
self.base_url = base_url.rstrip('/')
self.headers = {"X-API-Key": api_key}
def _get(self, endpoint, params):
url = f"{self.base_url}{endpoint}"
response = requests.get(url, headers=self.headers, params=params)
if response.status_code == 403:
raise PermissionError("Invalid API Key or unauthorized access.")
elif response.status_code == 404:
raise ValueError("Endpoint not found.")
elif response.status_code != 200:
response.raise_for_status()
return response.json()
def _format_datetime(self, dt):
if isinstance(dt, datetime):
return dt.isoformat()
return dt
def _fetch_paginated(self, endpoint, starttime, endtime, limit=5000, house_name=None, extra_params=None):
offset = 0
all_results = []
is_sensor_map = False
while True:
params = {
"starttime": self._format_datetime(starttime),
"endtime": self._format_datetime(endtime),
"limit": limit,
"offset": offset
}
if extra_params:
params.update(extra_params)
path = endpoint.format(house_name=house_name) if house_name else endpoint
response = self._get(path, params)
if "sensors" in response:
is_sensor_map = True
all_results.extend(response["sensors"])
if len(response["sensors"]) < limit:
break
elif "data" in response:
all_results.extend(response["data"])
if len(response["data"]) < limit:
break
else:
break
offset += limit
return all_results if not is_sensor_map else {"sensors": all_results}
def get_houses(self):
return self._get("/houses", {})["houses"]
def get_house(self, house_name):
return self._get(f"/houses/{house_name}", {})
def get_house_sensors(self, house_name):
return self._get(f"/houses/{house_name}/sensors", {})
def get_smartmeter_data(self, starttime, endtime, limit=5000):
return self._fetch_paginated("/smartmeter-data", starttime, endtime, limit)
def get_shower_data(self, starttime, endtime, limit=5000):
return self._fetch_paginated("/shower-data", starttime, endtime, limit)
def get_droople_data(self, starttime, endtime, limit=5000, include_heartbeat=False):
return self._fetch_paginated(
"/droople-data", starttime, endtime, limit,
extra_params={"include_heartbeat": str(include_heartbeat).lower()}
)
def get_droople_data_for_house(self, house_name, starttime, endtime, limit=5000, include_heartbeat=False):
return self._fetch_paginated(
"/houses/{house_name}/droople-data", starttime, endtime, limit, house_name,
extra_params={"include_heartbeat": str(include_heartbeat).lower()}
)
def get_smartmeter_data_for_house(self, house_name, starttime, endtime, limit=5000):
return self._fetch_paginated("/houses/{house_name}/smartmeter-data", starttime, endtime, limit, house_name)
def get_shower_data_for_house(self, house_name, starttime, endtime, limit=5000):
return self._fetch_paginated("/houses/{house_name}/shower-data", starttime, endtime, limit, house_name)