"""Handle the frontend for Home Assistant.""" import json import logging import os import pathlib from aiohttp import web import voluptuous as vol import jinja2 import homeassistant.helpers.config_validation as cv from homeassistant.components.http.view import HomeAssistantView from homeassistant.components import websocket_api from homeassistant.config import find_config_file, load_yaml_config_file from homeassistant.const import CONF_NAME, EVENT_THEMES_UPDATED from homeassistant.core import callback from homeassistant.helpers.translation import async_get_translations from homeassistant.loader import bind_hass from .storage import async_setup_frontend_storage DOMAIN = 'frontend' CONF_THEMES = 'themes' CONF_EXTRA_HTML_URL = 'extra_html_url' CONF_EXTRA_HTML_URL_ES5 = 'extra_html_url_es5' CONF_FRONTEND_REPO = 'development_repo' CONF_JS_VERSION = 'javascript_version' EVENT_PANELS_UPDATED = 'panels_updated' DEFAULT_THEME_COLOR = '#03A9F4' MANIFEST_JSON = { 'background_color': '#FFFFFF', 'description': 'Home automation platform that puts local control and privacy first.', 'dir': 'ltr', 'display': 'standalone', 'icons': [], 'lang': 'en-US', 'name': 'Home Assistant', 'short_name': 'Assistant', 'start_url': '/?homescreen=1', 'theme_color': DEFAULT_THEME_COLOR } for size in (192, 384, 512, 1024): MANIFEST_JSON['icons'].append({ 'src': '/static/icons/favicon-{size}x{size}.png'.format(size=size), 'sizes': '{size}x{size}'.format(size=size), 'type': 'image/png' }) DATA_FINALIZE_PANEL = 'frontend_finalize_panel' DATA_PANELS = 'frontend_panels' DATA_JS_VERSION = 'frontend_js_version' DATA_EXTRA_HTML_URL = 'frontend_extra_html_url' DATA_EXTRA_HTML_URL_ES5 = 'frontend_extra_html_url_es5' DATA_THEMES = 'frontend_themes' DATA_DEFAULT_THEME = 'frontend_default_theme' DEFAULT_THEME = 'default' PRIMARY_COLOR = 'primary-color' _LOGGER = logging.getLogger(__name__) CONFIG_SCHEMA = vol.Schema({ DOMAIN: vol.Schema({ vol.Optional(CONF_FRONTEND_REPO): cv.isdir, vol.Optional(CONF_THEMES): vol.Schema({ cv.string: {cv.string: cv.string} }), vol.Optional(CONF_EXTRA_HTML_URL): vol.All(cv.ensure_list, [cv.string]), # We no longer use these options. vol.Optional(CONF_EXTRA_HTML_URL_ES5): cv.match_all, vol.Optional(CONF_JS_VERSION): cv.match_all, }), }, extra=vol.ALLOW_EXTRA) SERVICE_SET_THEME = 'set_theme' SERVICE_RELOAD_THEMES = 'reload_themes' SERVICE_SET_THEME_SCHEMA = vol.Schema({ vol.Required(CONF_NAME): cv.string, }) WS_TYPE_GET_PANELS = 'get_panels' SCHEMA_GET_PANELS = websocket_api.BASE_COMMAND_MESSAGE_SCHEMA.extend({ vol.Required('type'): WS_TYPE_GET_PANELS, }) WS_TYPE_GET_THEMES = 'frontend/get_themes' SCHEMA_GET_THEMES = websocket_api.BASE_COMMAND_MESSAGE_SCHEMA.extend({ vol.Required('type'): WS_TYPE_GET_THEMES, }) WS_TYPE_GET_TRANSLATIONS = 'frontend/get_translations' SCHEMA_GET_TRANSLATIONS = websocket_api.BASE_COMMAND_MESSAGE_SCHEMA.extend({ vol.Required('type'): WS_TYPE_GET_TRANSLATIONS, vol.Required('language'): str, }) def generate_negative_index_regex(): """Generate regex for index.""" skip = [ # files "service_worker.js", "robots.txt", "onboarding.html", "manifest.json", ] for folder in ( "static", "frontend_latest", "frontend_es5", "local", "auth", "api", ): # Regex matching static, static/, static/index.html skip.append("{}(/|/.+|)".format(folder)) return r"(?!(" + "|".join(skip) + r")).*" class Panel: """Abstract class for panels.""" # Name of the webcomponent component_name = None # Icon to show in the sidebar (optional) sidebar_icon = None # Title to show in the sidebar (optional) sidebar_title = None # Url to show the panel in the frontend frontend_url_path = None # Config to pass to the webcomponent config = None # If the panel should only be visible to admins require_admin = False def __init__(self, component_name, sidebar_title, sidebar_icon, frontend_url_path, config, require_admin): """Initialize a built-in panel.""" self.component_name = component_name self.sidebar_title = sidebar_title self.sidebar_icon = sidebar_icon self.frontend_url_path = frontend_url_path or component_name self.config = config self.require_admin = require_admin @callback def to_response(self): """Panel as dictionary.""" return { 'component_name': self.component_name, 'icon': self.sidebar_icon, 'title': self.sidebar_title, 'config': self.config, 'url_path': self.frontend_url_path, 'require_admin': self.require_admin, } @bind_hass @callback def async_register_built_in_panel(hass, component_name, sidebar_title=None, sidebar_icon=None, frontend_url_path=None, config=None, require_admin=False): """Register a built-in panel.""" panel = Panel(component_name, sidebar_title, sidebar_icon, frontend_url_path, config, require_admin) panels = hass.data.setdefault(DATA_PANELS, {}) if panel.frontend_url_path in panels: _LOGGER.warning("Overwriting component %s", panel.frontend_url_path) panels[panel.frontend_url_path] = panel hass.bus.async_fire(EVENT_PANELS_UPDATED) @bind_hass @callback def async_remove_panel(hass, frontend_url_path): """Remove a built-in panel.""" panel = hass.data.get(DATA_PANELS, {}).pop(frontend_url_path, None) if panel is None: _LOGGER.warning("Removing unknown panel %s", frontend_url_path) hass.bus.async_fire(EVENT_PANELS_UPDATED) @bind_hass @callback def add_extra_html_url(hass, url, es5=False): """Register extra html url to load.""" key = DATA_EXTRA_HTML_URL_ES5 if es5 else DATA_EXTRA_HTML_URL url_set = hass.data.get(key) if url_set is None: url_set = hass.data[key] = set() url_set.add(url) def add_manifest_json_key(key, val): """Add a keyval to the manifest.json.""" MANIFEST_JSON[key] = val def _frontend_root(dev_repo_path): """Return root path to the frontend files.""" if dev_repo_path is not None: return pathlib.Path(dev_repo_path) / 'hass_frontend' import hass_frontend return hass_frontend.where() async def async_setup(hass, config): """Set up the serving of the frontend.""" await async_setup_frontend_storage(hass) hass.components.websocket_api.async_register_command( WS_TYPE_GET_PANELS, websocket_get_panels, SCHEMA_GET_PANELS) hass.components.websocket_api.async_register_command( WS_TYPE_GET_THEMES, websocket_get_themes, SCHEMA_GET_THEMES) hass.components.websocket_api.async_register_command( WS_TYPE_GET_TRANSLATIONS, websocket_get_translations, SCHEMA_GET_TRANSLATIONS) hass.http.register_view(ManifestJSONView) conf = config.get(DOMAIN, {}) repo_path = conf.get(CONF_FRONTEND_REPO) is_dev = repo_path is not None root_path = _frontend_root(repo_path) for path, should_cache in ( ("service_worker.js", False), ("robots.txt", False), ("onboarding.html", True), ("static", True), ("frontend_latest", True), ("frontend_es5", True), ): hass.http.register_static_path( "/{}".format(path), str(root_path / path), should_cache) hass.http.register_static_path( "/auth/authorize", str(root_path / "authorize.html"), False) local = hass.config.path('www') if os.path.isdir(local): hass.http.register_static_path("/local", local, not is_dev) hass.http.register_view(IndexView(repo_path)) for panel in ('kiosk', 'states', 'profile'): async_register_built_in_panel(hass, panel) for panel in ('dev-event', 'dev-info', 'dev-service', 'dev-state', 'dev-template', 'dev-mqtt'): async_register_built_in_panel(hass, panel, require_admin=True) if DATA_EXTRA_HTML_URL not in hass.data: hass.data[DATA_EXTRA_HTML_URL] = set() for url in conf.get(CONF_EXTRA_HTML_URL, []): add_extra_html_url(hass, url, False) _async_setup_themes(hass, conf.get(CONF_THEMES)) return True @callback def _async_setup_themes(hass, themes): """Set up themes data and services.""" hass.data[DATA_DEFAULT_THEME] = DEFAULT_THEME if themes is None: hass.data[DATA_THEMES] = {} return hass.data[DATA_THEMES] = themes @callback def update_theme_and_fire_event(): """Update theme_color in manifest.""" name = hass.data[DATA_DEFAULT_THEME] themes = hass.data[DATA_THEMES] if name != DEFAULT_THEME and PRIMARY_COLOR in themes[name]: MANIFEST_JSON['theme_color'] = themes[name][PRIMARY_COLOR] else: MANIFEST_JSON['theme_color'] = DEFAULT_THEME_COLOR hass.bus.async_fire(EVENT_THEMES_UPDATED, { 'themes': themes, 'default_theme': name, }) @callback def set_theme(call): """Set backend-preferred theme.""" data = call.data name = data[CONF_NAME] if name == DEFAULT_THEME or name in hass.data[DATA_THEMES]: _LOGGER.info("Theme %s set as default", name) hass.data[DATA_DEFAULT_THEME] = name update_theme_and_fire_event() else: _LOGGER.warning("Theme %s is not defined.", name) @callback def reload_themes(_): """Reload themes.""" path = find_config_file(hass.config.config_dir) new_themes = load_yaml_config_file(path)[DOMAIN].get(CONF_THEMES, {}) hass.data[DATA_THEMES] = new_themes if hass.data[DATA_DEFAULT_THEME] not in new_themes: hass.data[DATA_DEFAULT_THEME] = DEFAULT_THEME update_theme_and_fire_event() hass.services.async_register( DOMAIN, SERVICE_SET_THEME, set_theme, schema=SERVICE_SET_THEME_SCHEMA) hass.services.async_register(DOMAIN, SERVICE_RELOAD_THEMES, reload_themes) class IndexView(HomeAssistantView): """Serve the frontend.""" url = '/' name = 'frontend:index' requires_auth = False extra_urls = [ "/{extra:%s}" % generate_negative_index_regex() ] def __init__(self, repo_path): """Initialize the frontend view.""" self.repo_path = repo_path self._template_cache = None def get_template(self): """Get template.""" tpl = self._template_cache if tpl is None: with open( str(_frontend_root(self.repo_path) / 'index.html') ) as file: tpl = jinja2.Template(file.read()) # Cache template if not running from repository if self.repo_path is None: self._template_cache = tpl return tpl async def get(self, request, extra=None): """Serve the index view.""" hass = request.app['hass'] if (request.path != '/' and request.url.parts[1] not in hass.data[DATA_PANELS]): raise web.HTTPNotFound if not hass.components.onboarding.async_is_onboarded(): return web.Response(status=302, headers={ 'location': '/onboarding.html' }) template = self._template_cache if template is None: template = await hass.async_add_executor_job(self.get_template) return web.Response( text=template.render( theme_color=MANIFEST_JSON['theme_color'], extra_urls=hass.data[DATA_EXTRA_HTML_URL], ), content_type='text/html' ) class ManifestJSONView(HomeAssistantView): """View to return a manifest.json.""" requires_auth = False url = '/manifest.json' name = 'manifestjson' @callback def get(self, request): # pylint: disable=no-self-use """Return the manifest.json.""" msg = json.dumps(MANIFEST_JSON, sort_keys=True) return web.Response(text=msg, content_type="application/manifest+json") @callback def websocket_get_panels(hass, connection, msg): """Handle get panels command. Async friendly. """ user_is_admin = connection.user.is_admin panels = { panel_key: panel.to_response() for panel_key, panel in connection.hass.data[DATA_PANELS].items() if user_is_admin or not panel.require_admin} connection.send_message(websocket_api.result_message( msg['id'], panels)) @callback def websocket_get_themes(hass, connection, msg): """Handle get themes command. Async friendly. """ connection.send_message(websocket_api.result_message(msg['id'], { 'themes': hass.data[DATA_THEMES], 'default_theme': hass.data[DATA_DEFAULT_THEME], })) @websocket_api.async_response async def websocket_get_translations(hass, connection, msg): """Handle get translations command. Async friendly. """ resources = await async_get_translations(hass, msg['language']) connection.send_message(websocket_api.result_message( msg['id'], { 'resources': resources, } ))