What about something like this ?
"""Sync with firestore"""
import logging
import os
from typing import Any, Dict
from homeassistant.const import (EVENT_STATE_CHANGED, STATE_UNAVAILABLE, STATE_UNKNOWN)
from homeassistant.core import Event, HomeAssistant
_LOGGER = logging.getLogger(__name__)
REQUIREMENTS = ['firebase-admin']
DOMAIN = 'google_firebase'
def setup(hass: HomeAssistant, yaml_config: Dict[str, Any]):
import firebase_admin
from firebase_admin import credentials
from firebase_admin import firestore
certificate = os.path.join(hass.config.config_dir, 'firestore.json')
cred = credentials.Certificate(certificate)
firebase_admin.initialize_app(cred)
db = firestore.client()
col_ref = db.collection(u'states')
def update_firestore(event: Event):
new_state = event.data.get('new_state')
old_state = event.data.get('old_state')
if (new_state is None
or new_state.state in (STATE_UNKNOWN, '', STATE_UNAVAILABLE)
or ( not new_state.entity_id.startswith('light.')
and not new_state.entity_id.startswith('switch.'))):
return
col_ref.document(new_state.entity_id).set({u'state': new_state.state})
# col_ref.document(new_state.entity_id).set(new_state.as_dict())
_LOGGER.debug('added to firestore id %s, from %s to %s',
new_state.entity_id,
old_state.state if old_state is not None else None,
new_state.state)
def on_snapshot(col_snapshot, changes, read_time):
for change in changes:
id = change.document.id
state = change.document.to_dict().get('state')
if state in ['on', 'off']:
method_string = 'turn_' + state
_LOGGER.debug('Calling homeassistant.%s %s', method_string, id)
hass.services.call('homeassistant', method_string, { 'entity_id': id })
query_watch = col_ref.on_snapshot(on_snapshot)
hass.bus.listen(EVENT_STATE_CHANGED, update_firestore)
return True
!
.
!
.
!