-
Notifications
You must be signed in to change notification settings - Fork 2
Open
Description
I made a single script LLM genrated with yaml config to port home assistant sensors
home_assistant:
url: "https://home.ddnsgeek.com"
token: "1234567890"
sensors:
ha_upload: "sensor.fruzelbox_upload_throughput"
ha_download: "sensor.fruzelbox_download_throughput"
#!/usr/bin/env python3
import yaml
import requests
import os
import sys
# Get absolute path for Debian stability
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
CONFIG_PATH = os.path.join(BASE_DIR, "config.yaml")
def load_config():
with open(CONFIG_PATH, "r") as f:
return yaml.safe_load(f)
def get_ha_state(entity_id, url, token):
"""
Fetch the state from Home Assistant.
Returns the state and attributes for potential future use.
"""
api_url = f"{url.rstrip('/')}/api/states/{entity_id}"
headers = {"Authorization": f"Bearer {token}"}
try:
r = requests.get(api_url, headers=headers, timeout=2)
r.raise_for_status()
return r.json()
except Exception:
return None
def main():
config = load_config()
ha_url = config['home_assistant']['url']
ha_token = config['home_assistant']['token']
sensor_map = config.get('sensors', {})
while True:
try:
line = sys.stdin.readline()
if not line:
break
req = line.strip().split("\t")
command = req[0]
# 1. Metadata request
if command == "?":
# Print all keys from YAML separated by tabs
print("\t".join(sensor_map.keys()))
# 2. Sensor specific requests
elif command in sensor_map:
entity_id = sensor_map[command]
sub_req = req[1] if len(req) > 1 else "value"
if sub_req == "value":
data = get_ha_state(entity_id, ha_url, ha_token)
print(data['state'] if data else "")
elif sub_req == "unit":
data = get_ha_state(entity_id, ha_url, ha_token)
# Fetch unit from HA attributes if available
if data and 'unit_of_measurement' in data['attributes']:
print(data['attributes']['unit_of_measurement'])
else:
print("")
elif sub_req == "min":
print(0)
elif sub_req == "max":
# For throughput, there is often no fixed max
print(1000)
else:
print("")
else:
print("")
# Ensure output is sent immediately
sys.stdout.flush()
except EOFError:
break
except Exception as e:
# Errors to stderr to keep stdout clean for ksystemstats
sys.stderr.write(f"Error: {e}\n")
print("")
sys.stdout.flush()
if __name__ == "__main__":
main()
Reactions are currently unavailable
Metadata
Metadata
Assignees
Labels
No labels