custom_components/keymaster/__init__.py
async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
    """Set up is called when Home Assistant is loading our component."""
    hass.data.setdefault(DOMAIN, {})
    _LOGGER.info(
        "Version %s is starting, if you have any issues please report" " them here: %s",
        VERSION,
        ISSUE_URL,
    )
    should_generate_package = config_entry.data.get(CONF_GENERATE)

    updated_config = config_entry.data.copy()

    # pop CONF_GENERATE if it is in data
    updated_config.pop(CONF_GENERATE, None)

    # If CONF_PATH is absolute, make it relative. This can be removed in the future,
    # it is only needed for entries that are being migrated from using the old absolute
    # path
    config_path = hass.config.path()
    if config_entry.data[CONF_PATH].startswith(config_path):
        num_chars_config_path = len(config_path)
        updated_config[CONF_PATH] = updated_config[CONF_PATH][num_chars_config_path:]
        # Remove leading slashes
        updated_config[CONF_PATH] = updated_config[CONF_PATH].lstrip("/").lstrip("\\")

    if "parent" not in config_entry.data.keys():
        updated_config[CONF_PARENT] = None
    elif config_entry.data[CONF_PARENT] == "(none)":
        updated_config[CONF_PARENT] = None

    if updated_config != config_entry.data:
        hass.config_entries.async_update_entry(config_entry, data=updated_config)

    config_entry.add_update_listener(update_listener)

    primary_lock, child_locks = await generate_keymaster_locks(hass, config_entry)

    hass.data[DOMAIN][config_entry.entry_id] = {
        PRIMARY_LOCK: primary_lock,
        CHILD_LOCKS: child_locks,
        UNSUB_LISTENERS: [],
    }
    coordinator = LockUsercodeUpdateCoordinator(
        hass, config_entry, async_get_entity_registry(hass)
    )
    hass.data[DOMAIN][config_entry.entry_id][COORDINATOR] = coordinator

    # Button Press
    async def _refresh_codes(service: ServiceCall) -> None:
        """Refresh lock codes."""
        _LOGGER.debug("Refresh Codes service: %s", service)
        entity_id = service.data[ATTR_ENTITY_ID]
        instance_id = 1
        await refresh_codes(hass, entity_id, instance_id)

    hass.services.async_register(
        DOMAIN,
        SERVICE_REFRESH_CODES,
        _refresh_codes,
        schema=vol.Schema(
            {
                vol.Required(ATTR_ENTITY_ID): vol.Coerce(str),
            }
        ),
    )

    # Add code
    async def _add_code(service: ServiceCall) -> None:
        """Set a user code."""
        _LOGGER.debug("Add Code service: %s", service)
        entity_id = service.data[ATTR_ENTITY_ID]
        code_slot = service.data[ATTR_CODE_SLOT]
        usercode = service.data[ATTR_USER_CODE]
        await add_code(hass, entity_id, code_slot, usercode)

    hass.services.async_register(
        DOMAIN,
        SERVICE_ADD_CODE,
        _add_code,
        schema=vol.Schema(
            {
                vol.Required(ATTR_ENTITY_ID): vol.Coerce(str),
                vol.Required(ATTR_CODE_SLOT): vol.Coerce(int),
                vol.Required(ATTR_USER_CODE): vol.Coerce(str),
            }
        ),
    )

    # Clear code
    async def _clear_code(service: ServiceCall) -> None:
        """Clear a user code."""
        _LOGGER.debug("Clear Code service: %s", service)
        entity_id = service.data[ATTR_ENTITY_ID]
        code_slot = service.data[ATTR_CODE_SLOT]
        await clear_code(hass, entity_id, code_slot)

    hass.services.async_register(
        DOMAIN,
        SERVICE_CLEAR_CODE,
        _clear_code,
        schema=vol.Schema(
            {
                vol.Required(ATTR_ENTITY_ID): vol.Coerce(str),
                vol.Required(ATTR_CODE_SLOT): vol.Coerce(int),
            }
        ),
    )

    # Generate package files
    def _generate_package(service: ServiceCall) -> None:
        """Generate the package files."""
        _LOGGER.debug("DEBUG: %s", service)
        name = service.data[ATTR_NAME]
        generate_package_files(hass, name)

    hass.services.async_register(
        DOMAIN,
        SERVICE_GENERATE_PACKAGE,
        _generate_package,
        schema=vol.Schema({vol.Optional(ATTR_NAME): vol.Coerce(str)}),
    )

    await async_reset_code_slot_if_pin_unknown(
        hass,
        primary_lock.lock_name,
        config_entry.data[CONF_SLOTS],
        config_entry.data[CONF_START],
    )

    for platform in PLATFORMS:
        hass.async_create_task(
            hass.config_entries.async_forward_entry_setup(config_entry, platform)
        )

    # if the use turned on the bool generate the files
    if should_generate_package:
        servicedata = {"lockname": primary_lock.lock_name}
        await hass.services.async_call(
            DOMAIN, SERVICE_GENERATE_PACKAGE, servicedata, blocking=True
        )

    if async_using_zwave_js(lock=primary_lock):
        # Listen to Z-Wave JS events so we can fire our own events
        hass.data[DOMAIN][config_entry.entry_id][UNSUB_LISTENERS].append(
            hass.bus.async_listen(
                ZWAVE_JS_NOTIFICATION_EVENT,
                functools.partial(handle_zwave_js_event, hass, config_entry),
            )
        )
        return True

    # We only get here if we are not using zwave_js

    # Check if we need to check alarm type/alarm level sensors, in which case
    # we need to listen for lock state changes
    locks_to_watch = []
    for lock in [primary_lock, *child_locks]:
        if (
            lock.alarm_level_or_user_code_entity_id
            not in (
                None,
                "sensor.fake",
            )
            and lock.alarm_type_or_access_control_entity_id not in (None, "sensor.fake")
        ):
            locks_to_watch.append(lock)

    if locks_to_watch:
        if hass.state == CoreState.running:
            await homeassistant_started_listener(hass, config_entry, locks_to_watch)
        else:
            hass.bus.async_listen_once(
                EVENT_HOMEASSISTANT_STARTED,
                functools.partial(
                    homeassistant_started_listener, hass, config_entry, locks_to_watch
                ),
            )

    if primary_lock.parent is not None:
        await init_child_locks(
            hass,
            config_entry.data[CONF_START],
            config_entry.data[CONF_SLOTS],
            config_entry.data[CONF_LOCK_NAME],
        )

    return True
Set up is called when Home Assistant is loading our component.
Set up is called when Home Assistant is loading our component.
Set up is called when Home Assistant is loading our component .
async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
    hass.data.setdefault(DOMAIN, {})
    _LOGGER.info(
        "Version %s is starting, if you have any issues please report" " them here: %s",
        VERSION,
        ISSUE_URL,
    )
    should_generate_package = config_entry.data.get(CONF_GENERATE)
    updated_config = config_entry.data.copy()
    updated_config.pop(CONF_GENERATE, None)
    config_path = hass.config.path()
    if config_entry.data[CONF_PATH].startswith(config_path):
        num_chars_config_path = len(config_path)
        updated_config[CONF_PATH] = updated_config[CONF_PATH][num_chars_config_path:]
        updated_config[CONF_PATH] = updated_config[CONF_PATH].lstrip("/").lstrip("\\")
    if "parent" not in config_entry.data.keys():
        updated_config[CONF_PARENT] = None
    elif config_entry.data[CONF_PARENT] == "(none)":
        updated_config[CONF_PARENT] = None
    if updated_config != config_entry.data:
        hass.config_entries.async_update_entry(config_entry, data=updated_config)
    config_entry.add_update_listener(update_listener)
    primary_lock, child_locks = await generate_keymaster_locks(hass, config_entry)
    hass.data[DOMAIN][config_entry.entry_id] = {
        PRIMARY_LOCK: primary_lock,
        CHILD_LOCKS: child_locks,
        UNSUB_LISTENERS: [],
    }
    coordinator = LockUsercodeUpdateCoordinator(
        hass, config_entry, async_get_entity_registry(hass)
    )
    hass.data[DOMAIN][config_entry.entry_id][COORDINATOR] = coordinator
    async def _refresh_codes(service: ServiceCall) -> None:
        _LOGGER.debug("Refresh Codes service: %s", service)
        entity_id = service.data[ATTR_ENTITY_ID]
        instance_id = 1
        await refresh_codes(hass, entity_id, instance_id)
    hass.services.async_register(
        DOMAIN,
        SERVICE_REFRESH_CODES,
        _refresh_codes,
        schema=vol.Schema(
            {
                vol.Required(ATTR_ENTITY_ID): vol.Coerce(str),
            }
        ),
    )
    async def _add_code(service: ServiceCall) -> None:
        _LOGGER.debug("Add Code service: %s", service)
        entity_id = service.data[ATTR_ENTITY_ID]
        code_slot = service.data[ATTR_CODE_SLOT]
        usercode = service.data[ATTR_USER_CODE]
        await add_code(hass, entity_id, code_slot, usercode)
    hass.services.async_register(
        DOMAIN,
        SERVICE_ADD_CODE,
        _add_code,
        schema=vol.Schema(
            {
                vol.Required(ATTR_ENTITY_ID): vol.Coerce(str),
                vol.Required(ATTR_CODE_SLOT): vol.Coerce(int),
                vol.Required(ATTR_USER_CODE): vol.Coerce(str),
            }
        ),
    )
    async def _clear_code(service: ServiceCall) -> None:
        _LOGGER.debug("Clear Code service: %s", service)
        entity_id = service.data[ATTR_ENTITY_ID]
        code_slot = service.data[ATTR_CODE_SLOT]
        await clear_code(hass, entity_id, code_slot)
    hass.services.async_register(
        DOMAIN,
        SERVICE_CLEAR_CODE,
        _clear_code,
        schema=vol.Schema(
            {
                vol.Required(ATTR_ENTITY_ID): vol.Coerce(str),
                vol.Required(ATTR_CODE_SLOT): vol.Coerce(int),
            }
        ),
    )
    def _generate_package(service: ServiceCall) -> None:
        _LOGGER.debug("DEBUG: %s", service)
        name = service.data[ATTR_NAME]
        generate_package_files(hass, name)
    hass.services.async_register(
        DOMAIN,
        SERVICE_GENERATE_PACKAGE,
        _generate_package,
        schema=vol.Schema({vol.Optional(ATTR_NAME): vol.Coerce(str)}),
    )
    await async_reset_code_slot_if_pin_unknown(
        hass,
        primary_lock.lock_name,
        config_entry.data[CONF_SLOTS],
        config_entry.data[CONF_START],
    )
    for platform in PLATFORMS:
        hass.async_create_task(
            hass.config_entries.async_forward_entry_setup(config_entry, platform)
        )
    if should_generate_package:
        servicedata = {"lockname": primary_lock.lock_name}
        await hass.services.async_call(
            DOMAIN, SERVICE_GENERATE_PACKAGE, servicedata, blocking=True
        )
    if async_using_zwave_js(lock=primary_lock):
        hass.data[DOMAIN][config_entry.entry_id][UNSUB_LISTENERS].append(
            hass.bus.async_listen(
                ZWAVE_JS_NOTIFICATION_EVENT,
                functools.partial(handle_zwave_js_event, hass, config_entry),
            )
        )
        return True
    locks_to_watch = []
    for lock in [primary_lock, *child_locks]:
        if (
            lock.alarm_level_or_user_code_entity_id
            not in (
                None,
                "sensor.fake",
            )
            and lock.alarm_type_or_access_control_entity_id not in (None, "sensor.fake")
        ):
            locks_to_watch.append(lock)
    if locks_to_watch:
        if hass.state == CoreState.running:
            await homeassistant_started_listener(hass, config_entry, locks_to_watch)
        else:
            hass.bus.async_listen_once(
                EVENT_HOMEASSISTANT_STARTED,
                functools.partial(
                    homeassistant_started_listener, hass, config_entry, locks_to_watch
                ),
            )
    if primary_lock.parent is not None:
        await init_child_locks(
            hass,
            config_entry.data[CONF_START],
            config_entry.data[CONF_SLOTS],
            config_entry.data[CONF_LOCK_NAME],
        )
    return True
async def async_setup_entry ( hass : HomeAssistant , config_entry : ConfigEntry ) -> bool : hass . data . setdefault ( DOMAIN , { } ) _LOGGER . info ( "Version %s is starting, if you have any issues please report" " them here: %s" , VERSION , ISSUE_URL , ) should_generate_package = config_entry . data . get ( CONF_GENERATE ) updated_config = config_entry . data . copy ( ) updated_config . pop ( CONF_GENERATE , None ) config_path = hass . config . path ( ) if config_entry . data [ CONF_PATH ] . startswith ( config_path ) : num_chars_config_path = len ( config_path ) updated_config [ CONF_PATH ] = updated_config [ CONF_PATH ] [ num_chars_config_path : ] updated_config [ CONF_PATH ] = updated_config [ CONF_PATH ] . lstrip ( "/" ) . lstrip ( "\\" ) if "parent" not in config_entry . data . keys ( ) : updated_config [ CONF_PARENT ] = None elif config_entry . data [ CONF_PARENT ] == "(none)" : updated_config [ CONF_PARENT ] = None if updated_config != config_entry . data : hass . config_entries . async_update_entry ( config_entry , data = updated_config ) config_entry . add_update_listener ( update_listener ) primary_lock , child_locks = await generate_keymaster_locks ( hass , config_entry ) hass . data [ DOMAIN ] [ config_entry . entry_id ] = { PRIMARY_LOCK : primary_lock , CHILD_LOCKS : child_locks , UNSUB_LISTENERS : [ ] , } coordinator = LockUsercodeUpdateCoordinator ( hass , config_entry , async_get_entity_registry ( hass ) ) hass . data [ DOMAIN ] [ config_entry . entry_id ] [ COORDINATOR ] = coordinator async def _refresh_codes ( service : ServiceCall ) -> None : """Refresh lock codes.""" _LOGGER . debug ( "Refresh Codes service: %s" , service ) entity_id = service . data [ ATTR_ENTITY_ID ] instance_id = 1 await refresh_codes ( hass , entity_id , instance_id ) hass . services . async_register ( DOMAIN , SERVICE_REFRESH_CODES , _refresh_codes , schema = vol . Schema ( { vol . Required ( ATTR_ENTITY_ID ) : vol . Coerce ( str ) , } ) , ) async def _add_code ( service : ServiceCall ) -> None : """Set a user code.""" _LOGGER . debug ( "Add Code service: %s" , service ) entity_id = service . data [ ATTR_ENTITY_ID ] code_slot = service . data [ ATTR_CODE_SLOT ] usercode = service . data [ ATTR_USER_CODE ] await add_code ( hass , entity_id , code_slot , usercode ) hass . services . async_register ( DOMAIN , SERVICE_ADD_CODE , _add_code , schema = vol . Schema ( { vol . Required ( ATTR_ENTITY_ID ) : vol . Coerce ( str ) , vol . Required ( ATTR_CODE_SLOT ) : vol . Coerce ( int ) , vol . Required ( ATTR_USER_CODE ) : vol . Coerce ( str ) , } ) , ) async def _clear_code ( service : ServiceCall ) -> None : """Clear a user code.""" _LOGGER . debug ( "Clear Code service: %s" , service ) entity_id = service . data [ ATTR_ENTITY_ID ] code_slot = service . data [ ATTR_CODE_SLOT ] await clear_code ( hass , entity_id , code_slot ) hass . services . async_register ( DOMAIN , SERVICE_CLEAR_CODE , _clear_code , schema = vol . Schema ( { vol . Required ( ATTR_ENTITY_ID ) : vol . Coerce ( str ) , vol . Required ( ATTR_CODE_SLOT ) : vol . Coerce ( int ) , } ) , ) def _generate_package ( service : ServiceCall ) -> None : """Generate the package files.""" _LOGGER . debug ( "DEBUG: %s" , service ) name = service . data [ ATTR_NAME ] generate_package_files ( hass , name ) hass . services . async_register ( DOMAIN , SERVICE_GENERATE_PACKAGE , _generate_package , schema = vol . Schema ( { vol . Optional ( ATTR_NAME ) : vol . Coerce ( str ) } ) , ) await async_reset_code_slot_if_pin_unknown ( hass , primary_lock . lock_name , config_entry . data [ CONF_SLOTS ] , config_entry . data [ CONF_START ] , ) for platform in PLATFORMS : hass . async_create_task ( hass . config_entries . async_forward_entry_setup ( config_entry , platform ) ) if should_generate_package : servicedata = { "lockname" : primary_lock . lock_name } await hass . services . async_call ( DOMAIN , SERVICE_GENERATE_PACKAGE , servicedata , blocking = True ) if async_using_zwave_js ( lock = primary_lock ) : hass . data [ DOMAIN ] [ config_entry . entry_id ] [ UNSUB_LISTENERS ] . append ( hass . bus . async_listen ( ZWAVE_JS_NOTIFICATION_EVENT , functools . partial ( handle_zwave_js_event , hass , config_entry ) , ) ) return True locks_to_watch = [ ] for lock in [ primary_lock , * child_locks ] : if ( lock . alarm_level_or_user_code_entity_id not in ( None , "sensor.fake" , ) and lock . alarm_type_or_access_control_entity_id not in ( None , "sensor.fake" ) ) : locks_to_watch . append ( lock ) if locks_to_watch : if hass . state == CoreState . running : await homeassistant_started_listener ( hass , config_entry , locks_to_watch ) else : hass . bus . async_listen_once ( EVENT_HOMEASSISTANT_STARTED , functools . partial ( homeassistant_started_listener , hass , config_entry , locks_to_watch ) , ) if primary_lock . parent is not None : await init_child_locks ( hass , config_entry . data [ CONF_START ] , config_entry . data [ CONF_SLOTS ] , config_entry . data [ CONF_LOCK_NAME ] , ) return True
Set up is called when Home Assistant is loading our component.
Set up is called when Home Assistant is loading our component .
"""Set up is called when Home Assistant is loading our component.""" # pop CONF_GENERATE if it is in data # If CONF_PATH is absolute, make it relative. This can be removed in the future, # it is only needed for entries that are being migrated from using the old absolute # path # Remove leading slashes # Button Press """Refresh lock codes.""" # Add code """Set a user code.""" # Clear code """Clear a user code.""" # Generate package files """Generate the package files.""" # if the use turned on the bool generate the files # Listen to Z-Wave JS events so we can fire our own events # We only get here if we are not using zwave_js # Check if we need to check alarm type/alarm level sensors, in which case # we need to listen for lock state changes
hass
HomeAssistant
config_entry
ConfigEntry