hass

#!/usr/bin/env python3
import argparse
import asyncio
import json
import os

import httpx

TOKEN = os.environ['HASS_TOKEN']
URL = os.environ['HASS_URL']

TIMEOUT = 30


class HASS:
    def __init__(self, url, token):
        self.url = url
        self.client = httpx.AsyncClient()
        self.client.headers['Authorization'] = f'Bearer {token}'

    async def get_states(self):
        response = await self.client.get(f'{self.url}/api/states')
        response.raise_for_status()
        return response.json()

    async def get_services(self):
        response = await self.client.get(f'{self.url}/api/services')
        response.raise_for_status()
        return response.json()

    async def post_services(self, domain, service, entity_id, **kwargs):
        response = await self.client.post(
            f'{self.url}/api/services/{domain}/{service}',
            json=dict(entity_id=entity_id, **kwargs),
        )
        response.raise_for_status()
        return response.json()


async def office_fan_off(_):
    hass = HASS(URL, TOKEN)
    await hass.post_services(
        'scene', 'turn_on', entity_id='scene.office_fan_off')


async def office_fan_silent(_):
    hass = HASS(URL, TOKEN)
    await hass.post_services(
        'scene', 'turn_on', entity_id='scene.office_fan_silent')


async def office_fan_max(_):
    hass = HASS(URL, TOKEN)
    await hass.post_services(
        'scene', 'turn_on', entity_id='scene.office_fan_max')


async def office_at_monitor_desk(_):
    hass = HASS(URL, TOKEN)
    await hass.post_services(
        'scene', 'turn_on', entity_id='scene.office_at_monitor_desk')


async def office_at_tv_desk(_):
    hass = HASS(URL, TOKEN)
    await hass.post_services(
        'scene', 'turn_on', entity_id='scene.office_at_tv_desk')


async def office_on(_):
    hass = HASS(URL, TOKEN)
    await hass.post_services(
        'script', 'turn_on', entity_id='script.office_on')


async def office_dark(_):
    hass = HASS(URL, TOKEN)
    await hass.post_services(
        'scene', 'turn_on', entity_id='scene.office_dark')


async def office_on_cool(_):
    hass = HASS(URL, TOKEN)
    await hass.post_services(
        'scene', 'turn_on', entity_id='scene.office_on_cyan')


async def office_closet_lights_toggle(_):
    hass = HASS(URL, TOKEN)
    await hass.post_services(
        'light', 'toggle', entity_id='light.office_closet_lights')


async def office_fan_light_toggle(_):
    hass = HASS(URL, TOKEN)
    await hass.post_services(
        'light', 'toggle', entity_id='light.fan_office_light')


async def office_desk_lamp_toggle(_):
    hass = HASS(URL, TOKEN)
    await hass.post_services(
        'light', 'toggle', entity_id='light.office_desk_lamp')


async def office_studio_monitors_on(_):
    hass = HASS(URL, TOKEN)
    await hass.post_services(
        'switch', 'turn_on', entity_id='switch.office_studio_monitors')


async def office_studio_monitors_off(_):
    hass = HASS(URL, TOKEN)
    await hass.post_services(
        'switch', 'turn_off', entity_id='switch.office_studio_monitors')


async def get_states(args):
    hass = HASS(URL, TOKEN)
    print(json.dumps(await hass.get_states(), indent=2, sort_keys=True))


async def get_services(args):
    hass = HASS(URL, TOKEN)
    print(json.dumps(await hass.get_services(), indent=2, sort_keys=True))


async def post_services(args):
    hass = HASS(URL, TOKEN)
    await hass.post_services(args.domain, args.service, args.entity_id)


async def main():
    parser = argparse.ArgumentParser()
    subparsers = parser.add_subparsers(
        dest='command', metavar='COMMAND', required=True)

    # generic
    get_states_parser = subparsers.add_parser('get-states')
    get_states_parser.set_defaults(func=get_states)

    get_services_parser = subparsers.add_parser('get-services')
    get_services_parser.set_defaults(func=get_services)

    post_services_parser = subparsers.add_parser('post-services')
    post_services_parser.set_defaults(func=post_services)
    post_services_parser.add_argument('domain')
    post_services_parser.add_argument('service')
    post_services_parser.add_argument('entity_id')

    # specific
    office_studio_monitors_on_parser = subparsers.add_parser(
        'office-studio-monitors-on')
    office_studio_monitors_on_parser.set_defaults(
        func=office_studio_monitors_on)

    office_studio_monitors_off_parser = subparsers.add_parser(
        'office-studio-monitors-off')
    office_studio_monitors_off_parser.set_defaults(
        func=office_studio_monitors_off)

    office_closet_lights_on_parser = subparsers.add_parser(
        'office-closet-lights')
    office_closet_lights_on_parser.set_defaults(
        func=office_closet_lights_toggle)

    office_fan_light_toggle_parser = subparsers.add_parser(
        'office-fan-light')
    office_fan_light_toggle_parser.set_defaults(
        func=office_fan_light_toggle)

    office_desk_lamp_toggle_parser = subparsers.add_parser(
        'office-desk-lamp')
    office_desk_lamp_toggle_parser.set_defaults(
        func=office_desk_lamp_toggle)

    office_at_monitor_desk_parser = subparsers.add_parser(
        'office-at-monitor-desk')
    office_at_monitor_desk_parser.set_defaults(func=office_at_monitor_desk)

    office_at_tv_desk_parser = subparsers.add_parser('office-at-tv-desk')
    office_at_tv_desk_parser.set_defaults(func=office_at_tv_desk)

    office_on_parser = subparsers.add_parser('office-on')
    office_on_parser.set_defaults(func=office_on)

    office_on_cool_parser = subparsers.add_parser('office-on-cool')
    office_on_cool_parser.set_defaults(func=office_on_cool)

    office_fan_silent_parser = subparsers.add_parser('office-fan-silent')
    office_fan_silent_parser.set_defaults(func=office_fan_silent)

    office_fan_off_parser = subparsers.add_parser('office-fan-off')
    office_fan_off_parser.set_defaults(func=office_fan_off)

    office_fan_max_parser = subparsers.add_parser('office-fan-max')
    office_fan_max_parser.set_defaults(func=office_fan_max)

    office_dark_parser = subparsers.add_parser('office-dark')
    office_dark_parser.set_defaults(func=office_dark)

    args = parser.parse_args()

    await args.func(args)


if __name__ == '__main__':
    asyncio.run(main())