<s> """<STR_LIT>""" <EOL> import atexit <EOL> import json <EOL> import logging <EOL> import queue <EOL> import sqlite3 <EOL> import threading <EOL> from datetime import date , datetime <EOL> import homeassistant . util . dt as dt_util <EOL> from homeassistant . const import ( <EOL> EVENT_HOMEASSISTANT_START , EVENT_HOMEASSISTANT_STOP , EVENT_STATE_CHANGED , <EOL> EVENT_TIME_CHANGED , MATCH_ALL ) <EOL> from homeassistant . core import Event , EventOrigin , State <EOL> from homeassistant . remote import JSONEncoder <EOL> DOMAIN = "<STR_LIT>" <EOL> DB_FILE = '<STR_LIT>' <EOL> RETURN_ROWCOUNT = "<STR_LIT>" <EOL> RETURN_LASTROWID = "<STR_LIT>" <EOL> RETURN_ONE_ROW = "<STR_LIT>" <EOL> _INSTANCE = None <EOL> _LOGGER = logging . getLogger ( __name__ ) <EOL> def query ( sql_query , arguments = None ) : <EOL> """<STR_LIT>""" <EOL> _verify_instance ( ) <EOL> return _INSTANCE . query ( sql_query , arguments ) <EOL> def query_states ( state_query , arguments = None ) : <EOL> """<STR_LIT>""" <EOL> return [ <EOL> row for row in <EOL> ( row_to_state ( row ) for row in query ( state_query , arguments ) ) <EOL> if row is not None ] <EOL> def query_events ( event_query , arguments = None ) : <EOL> """<STR_LIT>""" <EOL> return [ <EOL> row for row in <EOL> ( row_to_event ( row ) for row in query ( event_query , arguments ) ) <EOL> if row is not None ] <EOL> def row_to_state ( row ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return State ( <EOL> row [ <NUM_LIT:1> ] , row [ <NUM_LIT:2> ] , json . loads ( row [ <NUM_LIT:3> ] ) , <EOL> dt_util . utc_from_timestamp ( row [ <NUM_LIT:4> ] ) , <EOL> dt_util . utc_from_timestamp ( row [ <NUM_LIT:5> ] ) ) <EOL> except ValueError : <EOL> _LOGGER . exception ( "<STR_LIT>" , row ) <EOL> return None <EOL> def row_to_event ( row ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return Event ( row [ <NUM_LIT:1> ] , json . loads ( row [ <NUM_LIT:2> ] ) , EventOrigin ( row [ <NUM_LIT:3> ] ) , <EOL> dt_util . utc_from_timestamp ( row [ <NUM_LIT:5> ] ) ) <EOL> except ValueError : <EOL> _LOGGER . exception ( "<STR_LIT>" , row ) <EOL> return None <EOL> def run_information ( point_in_time = None ) : <EOL> """<STR_LIT>""" <EOL> _verify_instance ( ) <EOL> if point_in_time is None or point_in_time > _INSTANCE . recording_start : <EOL> return RecorderRun ( ) <EOL> run = _INSTANCE . query ( <EOL> "<STR_LIT>" , <EOL> ( point_in_time , point_in_time ) , return_value = RETURN_ONE_ROW ) <EOL> return RecorderRun ( run ) if run else None <EOL> def setup ( hass , config ) : <EOL> """<STR_LIT>""" <EOL> global _INSTANCE <EOL> _INSTANCE = Recorder ( hass ) <EOL> return True <EOL> class RecorderRun ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , row = None ) : <EOL> """<STR_LIT>""" <EOL> self . end = None <EOL> if row is None : <EOL> self . start = _INSTANCE . recording_start <EOL> self . closed_incorrect = False <EOL> else : <EOL> self . start = dt_util . utc_from_timestamp ( row [ <NUM_LIT:1> ] ) <EOL> if row [ <NUM_LIT:2> ] is not None : <EOL> self . end = dt_util . utc_from_timestamp ( row [ <NUM_LIT:2> ] ) <EOL> self . closed_incorrect = bool ( row [ <NUM_LIT:3> ] ) <EOL> def entity_ids ( self , point_in_time = None ) : <EOL> """<STR_LIT>""" <EOL> where = self . where_after_start_run <EOL> where_data = [ ] <EOL> if point_in_time is not None or self . end is not None : <EOL> where += "<STR_LIT>" <EOL> where_data . append ( point_in_time or self . end ) <EOL> return [ row [ <NUM_LIT:0> ] for row in query ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" . format ( where ) , where_data ) ] <EOL> @ property <EOL> def where_after_start_run ( self ) : <EOL> """<STR_LIT>""" <EOL> return "<STR_LIT>" . format ( _adapt_datetime ( self . start ) ) <EOL> @ property <EOL> def where_limit_to_run ( self ) : <EOL> """<STR_LIT>""" <EOL> where = self . where_after_start_run <EOL> if self . end is not None : <EOL> where += "<STR_LIT>" . format ( _adapt_datetime ( self . end ) ) <EOL> return where <EOL> class Recorder ( threading . Thread ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , hass ) : <EOL> """<STR_LIT>""" <EOL> threading . Thread . __init__ ( self ) <EOL> self . hass = hass <EOL> self . conn = None <EOL> self . queue = queue . Queue ( ) <EOL> self . quit_object = object ( ) <EOL> self . lock = threading . Lock ( ) <EOL> self . recording_start = dt_util . utcnow ( ) <EOL> self . utc_offset = dt_util . now ( ) . utcoffset ( ) . total_seconds ( ) <EOL> self . db_path = self . hass . config . path ( DB_FILE ) <EOL> def start_recording ( event ) : <EOL> """<STR_LIT>""" <EOL> self . start ( ) <EOL> hass . bus . listen_once ( EVENT_HOMEASSISTANT_START , start_recording ) <EOL> hass . bus . listen_once ( EVENT_HOMEASSISTANT_STOP , self . shutdown ) <EOL> hass . bus . listen ( MATCH_ALL , self . event_listener ) <EOL> def run ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _setup_connection ( ) <EOL> self . _setup_run ( ) <EOL> while True : <EOL> event = self . queue . get ( ) <EOL> if event == self . quit_object : <EOL> self . _close_run ( ) <EOL> self . _close_connection ( ) <EOL> self . queue . task_done ( ) <EOL> return <EOL> elif event . event_type == EVENT_TIME_CHANGED : <EOL> self . queue . task_done ( ) <EOL> continue <EOL> event_id = self . record_event ( event ) <EOL> if event . event_type == EVENT_STATE_CHANGED : <EOL> self . record_state ( <EOL> event . data [ '<STR_LIT>' ] , event . data . get ( '<STR_LIT>' ) , <EOL> event_id ) <EOL> self . queue . task_done ( ) <EOL> def event_listener ( self , event ) : <EOL> """<STR_LIT>""" <EOL> self . queue . put ( event ) <EOL> def shutdown ( self , event ) : <EOL> """<STR_LIT>""" <EOL> self . queue . put ( self . quit_object ) <EOL> self . block_till_done ( ) <EOL> def record_state ( self , entity_id , state , event_id ) : <EOL> """<STR_LIT>""" <EOL> now = dt_util . utcnow ( ) <EOL> if state is None : <EOL> state_state = '<STR_LIT>' <EOL> state_domain = '<STR_LIT>' <EOL> state_attr = '<STR_LIT:{}>' <EOL> last_changed = last_updated = now <EOL> else : <EOL> state_domain = state . domain <EOL> state_state = state . state <EOL> state_attr = json . dumps ( dict ( state . attributes ) ) <EOL> last_changed = state . last_changed <EOL> last_updated = state . last_updated <EOL> info = ( <EOL> entity_id , state_domain , state_state , state_attr , <EOL> last_changed , last_updated , <EOL> now , self . utc_offset , event_id ) <EOL> self . query ( <EOL> """<STR_LIT>""" , <EOL> info ) <EOL> def record_event ( self , event ) : <EOL> """<STR_LIT>""" <EOL> info = ( <EOL> event . event_type , json . dumps ( event . data , cls = JSONEncoder ) , <EOL> str ( event . origin ) , dt_util . utcnow ( ) , event . time_fired , <EOL> self . utc_offset <EOL> ) <EOL> return self . query ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , info , RETURN_LASTROWID ) <EOL> def query ( self , sql_query , data = None , return_value = None ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> with self . conn , self . lock : <EOL> _LOGGER . debug ( "<STR_LIT>" , sql_query ) <EOL> cur = self . conn . cursor ( ) <EOL> if data is not None : <EOL> cur . execute ( sql_query , data ) <EOL> else : <EOL> cur . execute ( sql_query ) <EOL> if return_value == RETURN_ROWCOUNT : <EOL> return cur . rowcount <EOL> elif return_value == RETURN_LASTROWID : <EOL> return cur . lastrowid <EOL> elif return_value == RETURN_ONE_ROW : <EOL> return cur . fetchone ( ) <EOL> else : <EOL> return cur . fetchall ( ) <EOL> except ( sqlite3 . IntegrityError , sqlite3 . OperationalError , <EOL> sqlite3 . ProgrammingError ) : <EOL> _LOGGER . exception ( <EOL> "<STR_LIT>" , sql_query ) <EOL> return [ ] <EOL> def block_till_done ( self ) : <EOL> """<STR_LIT>""" <EOL> self . queue . join ( ) <EOL> def _setup_connection ( self ) : <EOL> """<STR_LIT>""" <EOL> self . conn = sqlite3 . connect ( self . db_path , check_same_thread = False ) <EOL> self . conn . row_factory = sqlite3 . Row <EOL> atexit . register ( self . _close_connection ) <EOL> sqlite3 . register_adapter ( date , _adapt_datetime ) <EOL> sqlite3 . register_adapter ( datetime , _adapt_datetime ) <EOL> cur = self . conn . cursor ( ) <EOL> def save_migration ( migration_id ) : <EOL> """<STR_LIT>""" <EOL> cur . execute ( '<STR_LIT>' , <EOL> ( migration_id , dt_util . utcnow ( ) ) ) <EOL> self . conn . commit ( ) <EOL> _LOGGER . info ( "<STR_LIT>" , migration_id ) <EOL> try : <EOL> cur . execute ( '<STR_LIT>' ) <EOL> migration_id = cur . fetchone ( ) [ <NUM_LIT:0> ] or <NUM_LIT:0> <EOL> except sqlite3 . OperationalError : <EOL> cur . execute ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> migration_id = <NUM_LIT:0> <EOL> if migration_id < <NUM_LIT:1> : <EOL> cur . execute ( """<STR_LIT>""" ) <EOL> cur . execute ( """<STR_LIT>""" ) <EOL> cur . execute ( <EOL> '<STR_LIT>' ) <EOL> cur . execute ( """<STR_LIT>""" ) <EOL> cur . execute ( '<STR_LIT>' ) <EOL> save_migration ( <NUM_LIT:1> ) <EOL> if migration_id < <NUM_LIT:2> : <EOL> cur . execute ( """<STR_LIT>""" ) <EOL> cur . execute ( '<STR_LIT>' ) <EOL> save_migration ( <NUM_LIT:2> ) <EOL> if migration_id < <NUM_LIT:3> : <EOL> utc_offset = self . utc_offset <EOL> cur . execute ( """<STR_LIT>""" ) <EOL> cur . execute ( """<STR_LIT>""" ) <EOL> cur . execute ( """<STR_LIT>""" ) <EOL> cur . execute ( "<STR_LIT>" , [ utc_offset ] ) <EOL> cur . execute ( "<STR_LIT>" , [ utc_offset ] ) <EOL> cur . execute ( "<STR_LIT>" , [ utc_offset ] ) <EOL> save_migration ( <NUM_LIT:3> ) <EOL> if migration_id < <NUM_LIT:4> : <EOL> cur . execute ( <EOL> """<STR_LIT>""" , [ self . utc_offset ] ) <EOL> cur . execute ( """<STR_LIT>""" ) <EOL> save_migration ( <NUM_LIT:4> ) <EOL> if migration_id < <NUM_LIT:5> : <EOL> try : <EOL> cur . execute ( """<STR_LIT>""" ) <EOL> except sqlite3 . OperationalError : <EOL> pass <EOL> self . conn . create_function ( <EOL> "<STR_LIT>" , <NUM_LIT:2> , <EOL> lambda string , substring : string . find ( substring ) + <NUM_LIT:1> ) <EOL> cur . execute ( """<STR_LIT>""" ) <EOL> cur . execute ( """<STR_LIT>""" ) <EOL> cur . execute ( """<STR_LIT>""" ) <EOL> save_migration ( <NUM_LIT:5> ) <EOL> def _close_connection ( self ) : <EOL> """<STR_LIT>""" <EOL> _LOGGER . info ( "<STR_LIT>" ) <EOL> atexit . unregister ( self . _close_connection ) <EOL> self . conn . close ( ) <EOL> def _setup_run ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . query ( """<STR_LIT>""" , ( self . recording_start , ) , <EOL> return_value = RETURN_ROWCOUNT ) : <EOL> _LOGGER . warning ( "<STR_LIT>" ) <EOL> self . query ( <EOL> """<STR_LIT>""" , <EOL> ( self . recording_start , dt_util . utcnow ( ) , self . utc_offset ) ) <EOL> def _close_run ( self ) : <EOL> """<STR_LIT>""" <EOL> self . query ( <EOL> "<STR_LIT>" , <EOL> ( dt_util . utcnow ( ) , self . recording_start ) ) <EOL> def _adapt_datetime ( datetimestamp ) : <EOL> """<STR_LIT>""" <EOL> return dt_util . as_utc ( datetimestamp ) . timestamp ( ) <EOL> def _verify_instance ( ) : <EOL> """<STR_LIT>""" <EOL> if _INSTANCE is None : <EOL> raise RuntimeError ( "<STR_LIT>" ) </s>
