<s> """<STR_LIT>""" <EOL> import logging <EOL> import os <EOL> import socket <EOL> import time <EOL> import voluptuous as vol <EOL> from homeassistant . bootstrap import prepare_setup_platform <EOL> from homeassistant . config import load_yaml_config_file <EOL> from homeassistant . exceptions import HomeAssistantError <EOL> from homeassistant . helpers import template <EOL> import homeassistant . helpers . config_validation as cv <EOL> from homeassistant . const import ( <EOL> EVENT_HOMEASSISTANT_START , EVENT_HOMEASSISTANT_STOP , <EOL> CONF_PLATFORM , CONF_SCAN_INTERVAL , CONF_VALUE_TEMPLATE ) <EOL> _LOGGER = logging . getLogger ( __name__ ) <EOL> DOMAIN = "<STR_LIT>" <EOL> MQTT_CLIENT = None <EOL> SERVICE_PUBLISH = '<STR_LIT>' <EOL> EVENT_MQTT_MESSAGE_RECEIVED = '<STR_LIT>' <EOL> REQUIREMENTS = [ '<STR_LIT>' ] <EOL> CONF_EMBEDDED = '<STR_LIT>' <EOL> CONF_BROKER = '<STR_LIT>' <EOL> CONF_PORT = '<STR_LIT:port>' <EOL> CONF_CLIENT_ID = '<STR_LIT>' <EOL> CONF_KEEPALIVE = '<STR_LIT>' <EOL> CONF_USERNAME = '<STR_LIT:username>' <EOL> CONF_PASSWORD = '<STR_LIT:password>' <EOL> CONF_CERTIFICATE = '<STR_LIT>' <EOL> CONF_PROTOCOL = '<STR_LIT>' <EOL> CONF_STATE_TOPIC = '<STR_LIT>' <EOL> CONF_COMMAND_TOPIC = '<STR_LIT>' <EOL> CONF_QOS = '<STR_LIT>' <EOL> CONF_RETAIN = '<STR_LIT>' <EOL> PROTOCOL_31 = '<STR_LIT>' <EOL> PROTOCOL_311 = '<STR_LIT>' <EOL> DEFAULT_PORT = <NUM_LIT> <EOL> DEFAULT_KEEPALIVE = <NUM_LIT> <EOL> DEFAULT_QOS = <NUM_LIT:0> <EOL> DEFAULT_RETAIN = False <EOL> DEFAULT_PROTOCOL = PROTOCOL_311 <EOL> ATTR_TOPIC = '<STR_LIT>' <EOL> ATTR_PAYLOAD = '<STR_LIT>' <EOL> ATTR_PAYLOAD_TEMPLATE = '<STR_LIT>' <EOL> ATTR_QOS = CONF_QOS <EOL> ATTR_RETAIN = CONF_RETAIN <EOL> MAX_RECONNECT_WAIT = <NUM_LIT> <EOL> def valid_subscribe_topic ( value , invalid_chars = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( value , str ) and all ( c not in value for c in invalid_chars ) : <EOL> return vol . Length ( min = <NUM_LIT:1> , max = <NUM_LIT> ) ( value ) <EOL> raise vol . Invalid ( '<STR_LIT>' ) <EOL> def valid_publish_topic ( value ) : <EOL> """<STR_LIT>""" <EOL> return valid_subscribe_topic ( value , invalid_chars = '<STR_LIT>' ) <EOL> _VALID_QOS_SCHEMA = vol . All ( vol . Coerce ( int ) , vol . In ( [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:2> ] ) ) <EOL> _HBMQTT_CONFIG_SCHEMA = vol . Schema ( dict ) <EOL> CONFIG_SCHEMA = vol . Schema ( { <EOL> DOMAIN : vol . Schema ( { <EOL> vol . Optional ( CONF_CLIENT_ID ) : cv . string , <EOL> vol . Optional ( CONF_KEEPALIVE , default = DEFAULT_KEEPALIVE ) : <EOL> vol . All ( vol . Coerce ( int ) , vol . Range ( min = <NUM_LIT:15> ) ) , <EOL> vol . Optional ( CONF_BROKER ) : cv . string , <EOL> vol . Optional ( CONF_PORT , default = DEFAULT_PORT ) : <EOL> vol . All ( vol . Coerce ( int ) , vol . Range ( min = <NUM_LIT:1> , max = <NUM_LIT> ) ) , <EOL> vol . Optional ( CONF_USERNAME ) : cv . string , <EOL> vol . Optional ( CONF_PASSWORD ) : cv . string , <EOL> vol . Optional ( CONF_CERTIFICATE ) : cv . isfile , <EOL> vol . Optional ( CONF_PROTOCOL , default = DEFAULT_PROTOCOL ) : <EOL> vol . All ( cv . string , vol . In ( [ PROTOCOL_31 , PROTOCOL_311 ] ) ) , <EOL> vol . Optional ( CONF_EMBEDDED ) : _HBMQTT_CONFIG_SCHEMA , <EOL> } ) , <EOL> } , extra = vol . ALLOW_EXTRA ) <EOL> MQTT_BASE_PLATFORM_SCHEMA = vol . Schema ( { <EOL> vol . Required ( CONF_PLATFORM ) : DOMAIN , <EOL> vol . Optional ( CONF_SCAN_INTERVAL ) : <EOL> vol . All ( vol . Coerce ( int ) , vol . Range ( min = <NUM_LIT:1> ) ) , <EOL> vol . Optional ( CONF_QOS , default = DEFAULT_QOS ) : _VALID_QOS_SCHEMA , <EOL> } ) <EOL> MQTT_RO_PLATFORM_SCHEMA = MQTT_BASE_PLATFORM_SCHEMA . extend ( { <EOL> vol . Required ( CONF_STATE_TOPIC ) : valid_subscribe_topic , <EOL> vol . Optional ( CONF_VALUE_TEMPLATE ) : cv . template , <EOL> } ) <EOL> MQTT_RW_PLATFORM_SCHEMA = MQTT_BASE_PLATFORM_SCHEMA . extend ( { <EOL> vol . Required ( CONF_COMMAND_TOPIC ) : valid_publish_topic , <EOL> vol . Optional ( CONF_RETAIN , default = DEFAULT_RETAIN ) : cv . boolean , <EOL> vol . Optional ( CONF_STATE_TOPIC ) : valid_subscribe_topic , <EOL> vol . Optional ( CONF_VALUE_TEMPLATE ) : cv . template , <EOL> } ) <EOL> MQTT_PUBLISH_SCHEMA = vol . Schema ( { <EOL> vol . Required ( ATTR_TOPIC ) : valid_publish_topic , <EOL> vol . Exclusive ( ATTR_PAYLOAD , '<STR_LIT>' ) : object , <EOL> vol . Exclusive ( ATTR_PAYLOAD_TEMPLATE , '<STR_LIT>' ) : cv . string , <EOL> vol . Required ( ATTR_QOS , default = DEFAULT_QOS ) : _VALID_QOS_SCHEMA , <EOL> vol . Required ( ATTR_RETAIN , default = DEFAULT_RETAIN ) : cv . boolean , <EOL> } , required = True ) <EOL> def _build_publish_data ( topic , qos , retain ) : <EOL> """<STR_LIT>""" <EOL> data = { ATTR_TOPIC : topic } <EOL> if qos is not None : <EOL> data [ ATTR_QOS ] = qos <EOL> if retain is not None : <EOL> data [ ATTR_RETAIN ] = retain <EOL> return data <EOL> def publish ( hass , topic , payload , qos = None , retain = None ) : <EOL> """<STR_LIT>""" <EOL> data = _build_publish_data ( topic , qos , retain ) <EOL> data [ ATTR_PAYLOAD ] = payload <EOL> hass . services . call ( DOMAIN , SERVICE_PUBLISH , data ) <EOL> def publish_template ( hass , topic , payload_template , qos = None , retain = None ) : <EOL> """<STR_LIT>""" <EOL> data = _build_publish_data ( topic , qos , retain ) <EOL> data [ ATTR_PAYLOAD_TEMPLATE ] = payload_template <EOL> hass . services . call ( DOMAIN , SERVICE_PUBLISH , data ) <EOL> def subscribe ( hass , topic , callback , qos = DEFAULT_QOS ) : <EOL> """<STR_LIT>""" <EOL> def mqtt_topic_subscriber ( event ) : <EOL> """<STR_LIT>""" <EOL> if _match_topic ( topic , event . data [ ATTR_TOPIC ] ) : <EOL> callback ( event . data [ ATTR_TOPIC ] , event . data [ ATTR_PAYLOAD ] , <EOL> event . data [ ATTR_QOS ] ) <EOL> hass . bus . listen ( EVENT_MQTT_MESSAGE_RECEIVED , mqtt_topic_subscriber ) <EOL> MQTT_CLIENT . subscribe ( topic , qos ) <EOL> def _setup_server ( hass , config ) : <EOL> """<STR_LIT>""" <EOL> conf = config . get ( DOMAIN , { } ) <EOL> if CONF_EMBEDDED not in conf and CONF_BROKER in conf : <EOL> return None <EOL> server = prepare_setup_platform ( hass , config , DOMAIN , '<STR_LIT>' ) <EOL> if server is None : <EOL> _LOGGER . error ( '<STR_LIT>' ) <EOL> return None <EOL> success , broker_config = server . start ( hass , conf . get ( CONF_EMBEDDED ) ) <EOL> return success and broker_config <EOL> def setup ( hass , config ) : <EOL> """<STR_LIT>""" <EOL> conf = config . get ( DOMAIN , { } ) <EOL> client_id = conf . get ( CONF_CLIENT_ID ) <EOL> keepalive = conf . get ( CONF_KEEPALIVE ) <EOL> broker_config = _setup_server ( hass , config ) <EOL> if broker_config and CONF_EMBEDDED not in conf : <EOL> broker , port , username , password , certificate , protocol = broker_config <EOL> elif not broker_config and CONF_BROKER not in conf : <EOL> _LOGGER . error ( '<STR_LIT>' ) <EOL> return False <EOL> if CONF_BROKER in conf : <EOL> broker = conf [ CONF_BROKER ] <EOL> port = conf [ CONF_PORT ] <EOL> username = conf . get ( CONF_USERNAME ) <EOL> password = conf . get ( CONF_PASSWORD ) <EOL> certificate = conf . get ( CONF_CERTIFICATE ) <EOL> protocol = conf [ CONF_PROTOCOL ] <EOL> if certificate is None and <NUM_LIT> < port < <NUM_LIT> and broker . endswith ( '<STR_LIT>' ) : <EOL> certificate = os . path . join ( os . path . dirname ( __file__ ) , <EOL> '<STR_LIT>' ) <EOL> global MQTT_CLIENT <EOL> try : <EOL> MQTT_CLIENT = MQTT ( hass , broker , port , client_id , keepalive , username , <EOL> password , certificate , protocol ) <EOL> except socket . error : <EOL> _LOGGER . exception ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> return False <EOL> def stop_mqtt ( event ) : <EOL> """<STR_LIT>""" <EOL> MQTT_CLIENT . stop ( ) <EOL> def start_mqtt ( event ) : <EOL> """<STR_LIT>""" <EOL> MQTT_CLIENT . start ( ) <EOL> hass . bus . listen_once ( EVENT_HOMEASSISTANT_STOP , stop_mqtt ) <EOL> def publish_service ( call ) : <EOL> """<STR_LIT>""" <EOL> msg_topic = call . data [ ATTR_TOPIC ] <EOL> payload = call . data . get ( ATTR_PAYLOAD ) <EOL> payload_template = call . data . get ( ATTR_PAYLOAD_TEMPLATE ) <EOL> qos = call . data [ ATTR_QOS ] <EOL> retain = call . data [ ATTR_RETAIN ] <EOL> try : <EOL> payload = ( payload if payload_template is None else <EOL> template . render ( hass , payload_template ) ) or '<STR_LIT>' <EOL> except template . jinja2 . TemplateError as exc : <EOL> _LOGGER . error ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> msg_topic , payload_template , exc ) <EOL> return <EOL> MQTT_CLIENT . publish ( msg_topic , payload , qos , retain ) <EOL> hass . bus . listen_once ( EVENT_HOMEASSISTANT_START , start_mqtt ) <EOL> descriptions = load_yaml_config_file ( <EOL> os . path . join ( os . path . dirname ( __file__ ) , '<STR_LIT>' ) ) <EOL> hass . services . register ( DOMAIN , SERVICE_PUBLISH , publish_service , <EOL> descriptions . get ( SERVICE_PUBLISH ) , <EOL> schema = MQTT_PUBLISH_SCHEMA ) <EOL> return True <EOL> class MQTT ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , hass , broker , port , client_id , keepalive , username , <EOL> password , certificate , protocol ) : <EOL> """<STR_LIT>""" <EOL> import paho . mqtt . client as mqtt <EOL> self . hass = hass <EOL> self . topics = { } <EOL> self . progress = { } <EOL> if protocol == PROTOCOL_31 : <EOL> proto = mqtt . MQTTv31 <EOL> else : <EOL> proto = mqtt . MQTTv311 <EOL> if client_id is None : <EOL> self . _mqttc = mqtt . Client ( protocol = proto ) <EOL> else : <EOL> self . _mqttc = mqtt . Client ( client_id , protocol = proto ) <EOL> if username is not None : <EOL> self . _mqttc . username_pw_set ( username , password ) <EOL> if certificate is not None : <EOL> self . _mqttc . tls_set ( certificate ) <EOL> self . _mqttc . on_subscribe = self . _mqtt_on_subscribe <EOL> self . _mqttc . on_unsubscribe = self . _mqtt_on_unsubscribe <EOL> self . _mqttc . on_connect = self . _mqtt_on_connect <EOL> self . _mqttc . on_disconnect = self . _mqtt_on_disconnect <EOL> self . _mqttc . on_message = self . _mqtt_on_message <EOL> self . _mqttc . connect ( broker , port , keepalive ) <EOL> def publish ( self , topic , payload , qos , retain ) : <EOL> """<STR_LIT>""" <EOL> self . _mqttc . publish ( topic , payload , qos , retain ) <EOL> def start ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _mqttc . loop_start ( ) <EOL> def stop ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _mqttc . disconnect ( ) <EOL> self . _mqttc . loop_stop ( ) <EOL> def subscribe ( self , topic , qos ) : <EOL> """<STR_LIT>""" <EOL> assert isinstance ( topic , str ) <EOL> if topic in self . topics : <EOL> return <EOL> result , mid = self . _mqttc . subscribe ( topic , qos ) <EOL> _raise_on_error ( result ) <EOL> self . progress [ mid ] = topic <EOL> self . topics [ topic ] = None <EOL> def unsubscribe ( self , topic ) : <EOL> """<STR_LIT>""" <EOL> result , mid = self . _mqttc . unsubscribe ( topic ) <EOL> _raise_on_error ( result ) <EOL> self . progress [ mid ] = topic <EOL> def _mqtt_on_connect ( self , _mqttc , _userdata , _flags , result_code ) : <EOL> """<STR_LIT>""" <EOL> if result_code != <NUM_LIT:0> : <EOL> _LOGGER . error ( '<STR_LIT>' , { <EOL> <NUM_LIT:1> : '<STR_LIT>' , <EOL> <NUM_LIT:2> : '<STR_LIT>' , <EOL> <NUM_LIT:3> : '<STR_LIT>' , <EOL> <NUM_LIT:4> : '<STR_LIT>' , <EOL> <NUM_LIT:5> : '<STR_LIT>' <EOL> } . get ( result_code , '<STR_LIT>' ) ) <EOL> self . _mqttc . disconnect ( ) <EOL> return <EOL> old_topics = self . topics <EOL> self . topics = { key : value for key , value in self . topics . items ( ) <EOL> if value is None } <EOL> for topic , qos in old_topics . items ( ) : <EOL> if qos is not None : <EOL> self . subscribe ( topic , qos ) <EOL> def _mqtt_on_subscribe ( self , _mqttc , _userdata , mid , granted_qos ) : <EOL> """<STR_LIT>""" <EOL> topic = self . progress . pop ( mid , None ) <EOL> if topic is None : <EOL> return <EOL> self . topics [ topic ] = granted_qos [ <NUM_LIT:0> ] <EOL> def _mqtt_on_message ( self , _mqttc , _userdata , msg ) : <EOL> """<STR_LIT>""" <EOL> self . hass . bus . fire ( EVENT_MQTT_MESSAGE_RECEIVED , { <EOL> ATTR_TOPIC : msg . topic , <EOL> ATTR_QOS : msg . qos , <EOL> ATTR_PAYLOAD : msg . payload . decode ( '<STR_LIT:utf-8>' ) , <EOL> } ) <EOL> def _mqtt_on_unsubscribe ( self , _mqttc , _userdata , mid , granted_qos ) : <EOL> """<STR_LIT>""" <EOL> topic = self . progress . pop ( mid , None ) <EOL> if topic is None : <EOL> return <EOL> self . topics . pop ( topic , None ) <EOL> def _mqtt_on_disconnect ( self , _mqttc , _userdata , result_code ) : <EOL> """<STR_LIT>""" <EOL> self . progress = { } <EOL> self . topics = { key : value for key , value in self . topics . items ( ) <EOL> if value is not None } <EOL> for key in list ( self . topics ) : <EOL> if self . topics [ key ] is None : <EOL> self . topics . pop ( key ) <EOL> if result_code == <NUM_LIT:0> : <EOL> return <EOL> tries = <NUM_LIT:0> <EOL> wait_time = <NUM_LIT:0> <EOL> while True : <EOL> try : <EOL> if self . _mqttc . reconnect ( ) == <NUM_LIT:0> : <EOL> _LOGGER . info ( '<STR_LIT>' ) <EOL> break <EOL> except socket . error : <EOL> pass <EOL> wait_time = min ( <NUM_LIT:2> ** tries , MAX_RECONNECT_WAIT ) <EOL> _LOGGER . warning ( <EOL> '<STR_LIT>' , <EOL> result_code , wait_time ) <EOL> time . sleep ( wait_time ) <EOL> tries += <NUM_LIT:1> <EOL> def _raise_on_error ( result ) : <EOL> """<STR_LIT>""" <EOL> if result != <NUM_LIT:0> : <EOL> raise HomeAssistantError ( '<STR_LIT>' . format ( result ) ) <EOL> def _match_topic ( subscription , topic ) : <EOL> """<STR_LIT>""" <EOL> if subscription . endswith ( '<STR_LIT:#>' ) : <EOL> return ( subscription [ : - <NUM_LIT:2> ] == topic or <EOL> topic . startswith ( subscription [ : - <NUM_LIT:1> ] ) ) <EOL> sub_parts = subscription . split ( '<STR_LIT:/>' ) <EOL> topic_parts = topic . split ( '<STR_LIT:/>' ) <EOL> return ( len ( sub_parts ) == len ( topic_parts ) and <EOL> all ( a == b for a , b in zip ( sub_parts , topic_parts ) if a != '<STR_LIT:+>' ) ) </s>
