SHARE
TWEET

remote.py

bbrendon Apr 16th, 2019 6 Never
Not a member of Pastebin yet? Sign Up, it unlocks many cool features!
  1.  
  2. from datetime import datetime
  3. import enum
  4. import json
  5. import logging
  6. import urllib.parse
  7.  
  8. from typing import Optional, Dict, Any, List
  9.  
  10. from aiohttp.hdrs import METH_GET, METH_POST, METH_DELETE, CONTENT_TYPE
  11. import requests
  12.  
  13. from homeassistant import core as ha
  14. from homeassistant.const import (
  15.     URL_API, SERVER_PORT, URL_API_CONFIG, URL_API_EVENTS, URL_API_STATES,
  16.     URL_API_SERVICES, CONTENT_TYPE_JSON, HTTP_HEADER_HA_AUTH,
  17.     URL_API_EVENTS_EVENT, URL_API_STATES_ENTITY, URL_API_SERVICES_SERVICE)
  18. from homeassistant.exceptions import HomeAssistantError
  19.  
  20. _LOGGER = logging.getLogger(__name__)
  21.  
  22.  
  23. class APIStatus(enum.Enum):
  24.     """Representation of an API status."""
  25.  
  26.     OK = "ok"
  27.     INVALID_PASSWORD = "invalid_password"
  28.     CANNOT_CONNECT = "cannot_connect"
  29.     UNKNOWN = "unknown"
  30.  
  31.     def __str__(self) -> str:
  32.         """Return the state."""
  33.         return self.value  # type: ignore
  34.  
  35.  
  36. class API:
  37.     """Object to pass around Home Assistant API location and credentials."""
  38.  
  39.     def __init__(self, host: str, api_password: Optional[str] = None,
  40.                  port: Optional[int] = SERVER_PORT,
  41.                  use_ssl: bool = False) -> None:
  42.         """Init the API."""
  43.         _LOGGER.warning('This class is deprecated and will be removed in 0.77')
  44.         self.host = host
  45.         self.port = port
  46.         self.api_password = api_password
  47.  
  48.         if host.startswith(("http://", "https://")):
  49.             self.base_url = host
  50.         elif use_ssl:
  51.             self.base_url = "https://{}".format(host)
  52.         else:
  53.             self.base_url = "http://{}".format(host)
  54.  
  55.         if port is not None:
  56.             self.base_url += ':{}'.format(port)
  57.  
  58.         self.status = None  # type: Optional[APIStatus]
  59.         self._headers = {CONTENT_TYPE: CONTENT_TYPE_JSON}
  60.  
  61.         if api_password is not None:
  62.             self._headers[HTTP_HEADER_HA_AUTH] = api_password
  63.  
  64.     def validate_api(self, force_validate: bool = False) -> bool:
  65.         """Test if we can communicate with the API."""
  66.         if self.status is None or force_validate:
  67.             self.status = validate_api(self)
  68.  
  69.         return self.status == APIStatus.OK
  70.  
  71.     def __call__(self, method: str, path: str, data: Dict = None,
  72.                  timeout: int = 5) -> requests.Response:
  73.         """Make a call to the Home Assistant API."""
  74.         if data is None:
  75.             data_str = None
  76.         else:
  77.             data_str = json.dumps(data, cls=JSONEncoder)
  78.  
  79.         url = urllib.parse.urljoin(self.base_url, path)
  80.  
  81.         try:
  82.             if method == METH_GET:
  83.                 return requests.get(
  84.                     url, params=data_str, timeout=timeout,
  85.                     headers=self._headers)
  86.  
  87.             return requests.request(
  88.                 method, url, data=data_str, timeout=timeout,
  89.                 headers=self._headers)
  90.  
  91.         except requests.exceptions.ConnectionError:
  92.             _LOGGER.exception("Error connecting to server")
  93.             raise HomeAssistantError("Error connecting to server")
  94.  
  95.         except requests.exceptions.Timeout:
  96.             error = "Timeout when talking to {}".format(self.host)
  97.             _LOGGER.exception(error)
  98.             raise HomeAssistantError(error)
  99.  
  100.     def __repr__(self) -> str:
  101.         """Return the representation of the API."""
  102.         return "<API({}, password: {})>".format(
  103.             self.base_url, 'yes' if self.api_password is not None else 'no')
  104.  
  105.  
  106. class JSONEncoder(json.JSONEncoder):
  107.     """JSONEncoder that supports Home Assistant objects."""
  108.  
  109.     # pylint: disable=method-hidden
  110.     def default(self, o: Any) -> Any:
  111.         """Convert Home Assistant objects.
  112.  
  113.         Hand other objects to the original method.
  114.         """
  115.         if isinstance(o, datetime):
  116.             return o.isoformat()
  117.         if isinstance(o, set):
  118.             return list(o)
  119.         if hasattr(o, 'as_dict'):
  120.             return o.as_dict()
  121.  
  122.         return json.JSONEncoder.default(self, o)
  123.  
  124.  
  125. def validate_api(api: API) -> APIStatus:
  126.     """Make a call to validate API."""
  127.     try:
  128.         req = api(METH_GET, URL_API)
  129.  
  130.         if req.status_code == 200:
  131.             return APIStatus.OK
  132.  
  133.         if req.status_code == 401:
  134.             return APIStatus.INVALID_PASSWORD
  135.  
  136.         return APIStatus.UNKNOWN
  137.  
  138.     except HomeAssistantError:
  139.         return APIStatus.CANNOT_CONNECT
  140.  
  141.  
  142. def get_event_listeners(api: API) -> Dict:
  143.     """List of events that is being listened for."""
  144.     try:
  145.         req = api(METH_GET, URL_API_EVENTS)
  146.  
  147.         return req.json() if req.status_code == 200 else {}  # type: ignore
  148.  
  149.     except (HomeAssistantError, ValueError):
  150.         # ValueError if req.json() can't parse the json
  151.         _LOGGER.exception("Unexpected result retrieving event listeners")
  152.  
  153.         return {}
  154.  
  155.  
  156. def fire_event(api: API, event_type: str, data: Dict = None) -> None:
  157.     """Fire an event at remote API."""
  158.     try:
  159.         req = api(METH_POST, URL_API_EVENTS_EVENT.format(event_type), data)
  160.  
  161.         if req.status_code != 200:
  162.             _LOGGER.error("Error firing event: %d - %s",
  163.                           req.status_code, req.text)
  164.  
  165.     except HomeAssistantError:
  166.         _LOGGER.exception("Error firing event")
  167.  
  168.  
  169. def get_state(api: API, entity_id: str) -> Optional[ha.State]:
  170.     """Query given API for state of entity_id."""
  171.     try:
  172.         req = api(METH_GET, URL_API_STATES_ENTITY.format(entity_id))
  173.  
  174.         # req.status_code == 422 if entity does not exist
  175.  
  176.         return ha.State.from_dict(req.json()) \
  177.             if req.status_code == 200 else None
  178.  
  179.     except (HomeAssistantError, ValueError):
  180.         # ValueError if req.json() can't parse the json
  181.         _LOGGER.exception("Error fetching state")
  182.  
  183.         return None
  184.  
  185.  
  186. def get_states(api: API) -> List[ha.State]:
  187.     """Query given API for all states."""
  188.     try:
  189.         req = api(METH_GET,
  190.                   URL_API_STATES)
  191.  
  192.         return [ha.State.from_dict(item) for
  193.                 item in req.json()]
  194.  
  195.     except (HomeAssistantError, ValueError, AttributeError):
  196.         # ValueError if req.json() can't parse the json
  197.         _LOGGER.exception("Error fetching states")
  198.  
  199.         return []
  200.  
  201.  
  202. def remove_state(api: API, entity_id: str) -> bool:
  203.     """Call API to remove state for entity_id.
  204.  
  205.     Return True if entity is gone (removed/never existed).
  206.     """
  207.     try:
  208.         req = api(METH_DELETE, URL_API_STATES_ENTITY.format(entity_id))
  209.  
  210.         if req.status_code in (200, 404):
  211.             return True
  212.  
  213.         _LOGGER.error("Error removing state: %d - %s",
  214.                       req.status_code, req.text)
  215.         return False
  216.     except HomeAssistantError:
  217.         _LOGGER.exception("Error removing state")
  218.  
  219.         return False
  220.  
  221.  
  222. def set_state(api: API, entity_id: str, new_state: str,
  223.               attributes: Dict = None, force_update: bool = False) -> bool:
  224.     """Tell API to update state for entity_id.
  225.  
  226.     Return True if success.
  227.     """
  228.     attributes = attributes or {}
  229.  
  230.     data = {'state': new_state,
  231.             'attributes': attributes,
  232.             'force_update': force_update}
  233.  
  234.     try:
  235.         req = api(METH_POST, URL_API_STATES_ENTITY.format(entity_id), data)
  236.  
  237.         if req.status_code not in (200, 201):
  238.             _LOGGER.error("Error changing state: %d - %s",
  239.                           req.status_code, req.text)
  240.             return False
  241.  
  242.         return True
  243.  
  244.     except HomeAssistantError:
  245.         _LOGGER.exception("Error setting state")
  246.  
  247.         return False
  248.  
  249.  
  250. def is_state(api: API, entity_id: str, state: str) -> bool:
  251.     """Query API to see if entity_id is specified state."""
  252.     cur_state = get_state(api, entity_id)
  253.  
  254.     return bool(cur_state and cur_state.state == state)
  255.  
  256.  
  257. def get_services(api: API) -> Dict:
  258.     """Return a list of dicts.
  259.  
  260.     Each dict has a string "domain" and a list of strings "services".
  261.     """
  262.     try:
  263.         req = api(METH_GET, URL_API_SERVICES)
  264.  
  265.         return req.json() if req.status_code == 200 else {}  # type: ignore
  266.  
  267.     except (HomeAssistantError, ValueError):
  268.         # ValueError if req.json() can't parse the json
  269.         _LOGGER.exception("Got unexpected services result")
  270.  
  271.         return {}
  272.  
  273.  
  274. def call_service(api: API, domain: str, service: str,
  275.                  service_data: Dict = None,
  276.                  timeout: int = 5) -> None:
  277.     """Call a service at the remote API."""
  278.     try:
  279.         req = api(METH_POST,
  280.                   URL_API_SERVICES_SERVICE.format(domain, service),
  281.                   service_data, timeout=timeout)
  282.  
  283.         if req.status_code != 200:
  284.             _LOGGER.error("Error calling service: %d - %s",
  285.                           req.status_code, req.text)
  286.  
  287.     except HomeAssistantError:
  288.         _LOGGER.exception("Error calling service")
  289.  
  290.  
  291. def get_config(api: API) -> Dict:
  292.     """Return configuration."""
  293.     try:
  294.         req = api(METH_GET, URL_API_CONFIG)
  295.  
  296.         if req.status_code != 200:
  297.             return {}
  298.  
  299.         result = req.json()
  300.         if 'components' in result:
  301.             result['components'] = set(result['components'])
  302.         return result  # type: ignore
  303.  
  304.     except (HomeAssistantError, ValueError):
  305.         # ValueError if req.json() can't parse the JSON
  306.         _LOGGER.exception("Got unexpected configuration results")
  307.  
  308.         return {}
RAW Paste Data
We use cookies for various purposes including analytics. By continuing to use Pastebin, you agree to our use of cookies as described in the Cookies Policy. OK, I Understand
Not a member of Pastebin yet?
Sign Up, it unlocks many cool features!
 
Top