104 lines
3.3 KiB
Python
104 lines
3.3 KiB
Python
|
|
import os
|
|
import json
|
|
import requests
|
|
from jinja2 import Environment, FileSystemLoader
|
|
from datetime import datetime
|
|
|
|
# --- Configuration ---
|
|
HA_URL = os.environ.get("HA_URL")
|
|
HA_TOKEN = os.environ.get("HA_TOKEN")
|
|
DATA_FILE = "sensor_data.json"
|
|
HTML_FILE = "index.html"
|
|
TEMPLATE_DIR = "templates"
|
|
|
|
# --- Error Handling ---
|
|
if not HA_URL or not HA_TOKEN:
|
|
raise ValueError("HA_URL and HA_TOKEN environment variables must be set.")
|
|
|
|
# --- Functions ---
|
|
|
|
def get_ha_data():
|
|
"""Fetches all states from Home Assistant and filters for temp/humidity sensors."""
|
|
headers = {
|
|
"Authorization": f"Bearer {HA_TOKEN}",
|
|
"content-type": "application/json",
|
|
}
|
|
url = f"{HA_URL}/api/states"
|
|
try:
|
|
response = requests.get(url, headers=headers)
|
|
response.raise_for_status() # Raise an exception for bad status codes
|
|
states = response.json()
|
|
|
|
sensors = [
|
|
s for s in states
|
|
if "unit_of_measurement" in s["attributes"] and (
|
|
s["attributes"]["unit_of_measurement"] == "°C" or
|
|
(s["attributes"]["unit_of_measurement"] == "%" and
|
|
s["attributes"].get("device_class") != "battery")
|
|
)
|
|
]
|
|
return sensors
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"Error connecting to Home Assistant: {e}")
|
|
return None
|
|
|
|
def read_historical_data():
|
|
"""Reads historical sensor data from the JSON file."""
|
|
if not os.path.exists(DATA_FILE):
|
|
return []
|
|
with open(DATA_FILE, "r") as f:
|
|
return json.load(f)
|
|
|
|
def write_historical_data(new_data):
|
|
"""Appends new sensor data to the historical record."""
|
|
historical_data = read_historical_data()
|
|
timestamp = datetime.now().isoformat()
|
|
|
|
for sensor in new_data:
|
|
historical_data.append({
|
|
"entity_id": sensor["entity_id"],
|
|
"friendly_name": sensor["attributes"].get("friendly_name", sensor["entity_id"]),
|
|
"state": sensor["state"],
|
|
"unit": sensor["attributes"].get("unit_of_measurement"),
|
|
"timestamp": timestamp,
|
|
})
|
|
|
|
with open(DATA_FILE, "w") as f:
|
|
json.dump(historical_data, f, indent=4)
|
|
|
|
def generate_html_report(recent_data, historical_data, last_updated_time):
|
|
"""Generates an HTML report from the sensor data."""
|
|
env = Environment(loader=FileSystemLoader(TEMPLATE_DIR))
|
|
template = env.get_template("index.html.j2")
|
|
|
|
html_content = template.render(
|
|
recent_data=recent_data,
|
|
historical_data=historical_data,
|
|
last_updated_time=last_updated_time
|
|
)
|
|
|
|
with open(HTML_FILE, "w") as f:
|
|
f.write(html_content)
|
|
print(f"Successfully generated HTML report: {HTML_FILE}")
|
|
|
|
# --- Main Execution ---
|
|
|
|
def main():
|
|
"""Main function to run the script."""
|
|
print("Fetching data from Home Assistant...")
|
|
recent_sensor_data = get_ha_data()
|
|
|
|
if recent_sensor_data:
|
|
print(f"Found {len(recent_sensor_data)} temperature/humidity sensors.")
|
|
write_historical_data(recent_sensor_data)
|
|
historical_data = read_historical_data()
|
|
last_updated_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
generate_html_report(recent_sensor_data, historical_data, last_updated_time)
|
|
else:
|
|
print("Could not fetch new data. Report generation skipped.")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|