Skip to content

Adding a Provider

Jump to bottom Edit New page
raman325 edited this page Jun 10, 2026 · 4 revisions

Adding a New Lock Provider

This tutorial walks through adding support for a new lock integration to Lock Code Manager. We'll build a provider for a hypothetical "SmartLock" integration.

For background on the contract, see Architecture and Provider-State-Management.

Prerequisites

Before starting:

  1. The lock integration exists in Home Assistant and creates lock.* entities.
  2. The integration provides a way to read, set, and clear PIN credentials.
  3. You understand the integration's user/credential model: does it expose users with credentials (Matter, Z-Wave User Credential CC), or just slot-indexed credentials (Z-Wave User Code CC, ZHA, Akuvox, Schlage)?

Choose Your Provider Flavor

LCM providers split into two flavors based on the lock's user model. Choose the one that matches your lock:

  • Slot-only — the lock indexes credentials by slot directly with no user concept (Z-Wave UC, ZHA, Zigbee2MQTT, Virtual), OR the lock's "user" is just the credential's friendly name (Akuvox, Schlage). The provider implements async_set_credential, async_delete_credential, async_get_users. The base seam writes credentials directly; the user_id parameter in async_set_credential is ignored.
  • Native-user — the lock has a real user record that owns one or more credentials (Matter, Z-Wave User Credential CC). The provider additionally implements async_set_user, async_delete_user, async_get_capabilities. The base seam runs a user-first lifecycle: create-or-find the user (via lcm:<slot>:<name> tag), then write the credential under it.

This tutorial covers the slot-only path first, then adds the native-user extensions.

Step 1: Create the Provider File

Create a new file in custom_components/lock_code_manager/providers/:

# custom_components/lock_code_manager/providers/smartlock.py
"""Module for SmartLock integration."""

from __future__ import annotations

from dataclasses import dataclass
from datetime import timedelta
import logging
from typing import Literal

from homeassistant.config_entries import ConfigEntry, ConfigEntryState
from homeassistant.core import callback

from ..domain.credentials import (
    Credential,
    CredentialRef,
    CredentialType,
    User,
    user_from_slot,
)
from ..domain.exceptions import (
    CodeRejectedError,
    DuplicateCodeError,
    LockDisconnected,
    LockOperationFailed,
)
from ..domain.models import SlotCredential
from ._base import BaseLock

_LOGGER = logging.getLogger(__name__)


@dataclass(repr=False, eq=False)
class SmartLockLock(BaseLock):
    """Class to represent SmartLock lock."""

    @property
    def domain(self) -> str:
        """Return integration domain."""
        return "smartlock"

Step 2: Override the Property Surface

Update Intervals

@property
def usercode_scan_interval(self) -> timedelta:
    """Return scan interval for usercodes (polling mode only)."""
    return timedelta(minutes=2)

@property
def hard_refresh_interval(self) -> timedelta | None:
    """Return drift-detection interval (None to disable)."""
    return timedelta(hours=1)

@property
def connection_check_interval(self) -> timedelta | None:
    """Return interval for connection state checks."""
    return timedelta(seconds=30)

For push-capable integrations, set connection_check_interval = None and rely on config-entry state changes instead.

Capability Flags

@property
def supports_push(self) -> bool:
    """Return True if the integration emits real-time events."""
    return False  # change to True if you'll implement push

@property
def supports_native_users(self) -> bool:
    """Return True for locks with a real user record (Matter, Z-Wave U3C)."""
    return False  # slot-only path; see "Native-User Extension" below

Step 3: Implement Connection Check

def is_integration_connected(self) -> bool:
    """Return whether the integration is reachable."""
    # Option A: check config entry state
    if self.lock_config_entry:
        return self.lock_config_entry.state == ConfigEntryState.LOADED

    # Option B: check the entity state
    state = self.hass.states.get(self.lock.entity_id)
    return state is not None and state.state not in ("unavailable", "unknown")

Step 4: Implement Read

Return the lock's current users and credentials as list[User]. Slot-only providers synthesize a single-credential user per occupied slot via user_from_slot.

async def async_get_users(self) -> list[User]:
    """Read credential state from the device.

    Slot-only path: synthesize one user per occupied slot. The base
    class projects these to dict[int, SlotCredential] via the
    default async_get_usercodes() projection.
    """
    try:
        codes = await self._fetch_codes()  # {slot: pin_or_None}
    except SomeDeviceError as err:
        raise LockDisconnected(f"get codes failed: {err}") from err

    return [
        user_from_slot(
            slot,
            SlotCredential.known(pin) if pin else SlotCredential.empty(),
        )
        for slot, pin in codes.items()
    ]

For write-only locks (the lock confirms occupancy but won't reveal the PIN), use SlotCredential.unreadable() instead of SlotCredential.known(...).

Step 5: Implement Write

async_set_credential

async def async_set_credential(
    self,
    user_id: int,
    credential: Credential,
    pin: str,
    *,
    name: str | None,
    source: Literal["sync", "direct"],
) -> bool:
    """Write a PIN to a slot; return whether the lock changed.

    Slot-only: user_id is ignored; address by credential.slot.
    """
    try:
        await self._device.set_code(credential.slot, pin, name=name)
    except DuplicatePinError as err:
        raise DuplicateCodeError(
            code_slot=credential.slot,
            lock_entity_id=self.lock.entity_id,
        ) from err
    except TransientError as err:
        # Transient: route to retry, not slot suspension.
        raise LockDisconnected(
            f"set slot {credential.slot} failed: {err}"
        ) from err
    except RejectedError as err:
        raise CodeRejectedError(
            code_slot=credential.slot,
            lock_entity_id=self.lock.entity_id,
            reason=str(err),
        ) from err
    return True

Return True if the value changed, False if it was already set. When the integration doesn't tell you (write-only locks), return True — the seam uses this to drive a coordinator refresh.

async_delete_credential

async def async_delete_credential(self, ref: CredentialRef) -> bool:
    """Delete the credential addressed by ref."""
    try:
        await self._device.clear_code(ref.slot)
    except TransientError as err:
        raise LockDisconnected(
            f"clear slot {ref.slot} failed: {err}"
        ) from err
    except LockApiError as err:
        raise LockOperationFailed(
            f"clear slot {ref.slot} failed: {err}"
        ) from err
    return True

The seam does not delete the owning user after async_delete_credential for native-user providers — the user is a persistent slot anchor; see Architecture for the lifecycle rationale.

Step 6: Implement Hard Refresh (Recommended)

If your integration caches data, bypass that cache to detect drift.

async def async_hard_refresh_codes(self) -> dict[int, SlotCredential]:
    """Force a fresh read and return the projected slot map."""
    try:
        await self._device.invalidate_cache()
    except SomeDeviceError as err:
        raise LockDisconnected(f"refresh failed: {err}") from err

    return await self.async_get_usercodes()

Step 7: Add Push Support (Optional)

If the integration emits real-time events:

@property
def supports_push(self) -> bool:
    return True

@dataclass(repr=False, eq=False)
class SmartLockLock(BaseLock):
    # ...existing fields...
    _event_unsub: Callable[[], None] | None = field(init=False, default=None)

    @callback
    def subscribe_push_updates(self) -> None:
        """Subscribe to real-time updates. Must be idempotent."""
        if self._event_unsub is not None:
            return

        @callback
        def on_code_changed(event_data) -> None:
            slot = event_data["slot"]
            state = (
                SlotCredential.known(event_data["pin"])
                if event_data.get("pin")
                else SlotCredential.empty()
            )
            if self.coordinator:
                self.coordinator.push_update({slot: state})

        self._event_unsub = self._device.subscribe(on_code_changed)

    @callback
    def unsubscribe_push_updates(self) -> None:
        """Unsubscribe. Must be idempotent."""
        if self._event_unsub:
            self._event_unsub()
            self._event_unsub = None

For locks that report which code was used to unlock, also fire LCM events for automations:

@callback
def _handle_lock_event(self, event_data) -> None:
    """Fire an LCM code-slot event for automations."""
    self.async_fire_code_slot_event(
        code_slot=event_data["slot"],
        to_locked=event_data.get("locked"),
        action_text=event_data.get("action"),
        source_data=event_data,
    )

Step 8: Register the Provider

Add your provider to the registry in providers/__init__.py:

"""Integrations module."""

from __future__ import annotations

from ._base import BaseLock
from .smartlock import SmartLockLock  # add the import
from .virtual import VirtualLock
from .zwave_js import ZWaveJSLock

INTEGRATIONS_CLASS_MAP: dict[str, type[BaseLock]] = {
    "smartlock": SmartLockLock,  # add the mapping
    "virtual": VirtualLock,
    "zwave_js": ZWaveJSLock,
}

Native-User Extension (for Matter-class locks)

If your lock has a real user record that owns credentials, set supports_native_users = True and add three more primitives.

async_get_capabilities

The seam reads this to decide whether to invoke the user lifecycle and to size the tagged user name.

from ..domain.credentials import (
    CredentialTypeCapability,
    LockCapabilities,
)


async def async_get_capabilities(self) -> LockCapabilities:
    """Report the lock's user/credential capabilities."""
    caps = await self._driver.get_capabilities()
    return LockCapabilities(
        supports_user_management=caps["supports_user_management"],
        max_users=caps["max_users"],
        credential_types={
            CredentialType.PIN: CredentialTypeCapability(
                num_slots=caps["pin_num_slots"],
                min_length=caps["pin_min_length"],
                max_length=caps["pin_max_length"],
                supports_learn=False,
            ),
        },
        max_user_name_length=caps.get("max_user_name_length", 0),
    )

async_set_user (find-or-create-by-tag)

The base seam passes a tagged user.name (lcm:<slot>:<display>). Discover or allocate the underlying lock user_id:

from ..domain.credentials import SetUserResult
from ._util import parse_tag


async def async_set_user(self, user: User) -> SetUserResult:
    """Find-or-create the lock user for the LCM slot encoded in user.name."""
    slot = self._slot_from_seam_user(user)
    existing_user_id = await self._find_user_index_for_slot(slot)
    try:
        result = await self._driver.set_user(
            user_id=existing_user_id,  # None on CREATE -> driver auto-allocates
            user_name=user.name,
            active=user.active,
        )
    except SomeDeviceError as err:
        raise LockDisconnected(
            f"set user for slot {slot} failed: {err}"
        ) from err
    return SetUserResult(
        user_id=result["user_id"],
        created=existing_user_id is None,
    )


def _slot_from_seam_user(self, user: User) -> int:
    """Recover LCM slot from the tagged user name; fall back to user_id."""
    if user.name:
        slot, _ = parse_tag(user.name)
        if slot is not None:
            return slot
    return user.user_id


async def _find_user_index_for_slot(self, slot: int) -> int | None:
    """Return the lock user_id LCM owns for slot.

    Two-pass:
    1. Canonical: a user whose name parses to this LCM slot.
    2. Legacy adoption: an untagged user at the pre-PR-C invariant
       (the position your lock used pre-4.0.0) that owns a PIN at
       the slot. The legacy pass MUST skip users already tagged for
       a different LCM slot.
    """
    users = await self.async_get_users()
    try:
        return next(
            u.user_id
            for u in users
            if u.name and parse_tag(u.name)[0] == slot
        )
    except StopIteration:
        # Replace the legacy-position predicate with whatever your
        # pre-4.0.0 invariant was (e.g. user_id == slot for U3C,
        # credential_index == slot for Matter).
        return next(
            (
                u.user_id
                for u in users
                if u.user_id == slot
                and parse_tag(u.name or "")[0] is None
                for cred in u.pin_credentials
                if cred.slot == slot
            ),
            None,
        )

async_delete_user

async def async_delete_user(self, user_id: int) -> None:
    """Delete a lock user (cascades its credentials per spec)."""
    try:
        await self._driver.delete_user(user_id)
    except SomeDeviceError as err:
        raise LockDisconnected(f"delete user {user_id} failed: {err}") from err

Slot teardown hook

When a slot is removed from LCM config, the base seam calls async_release_managed_slot(slot). Native-user providers override to release the persistent user anchor:

async def async_release_managed_slot(self, slot: int) -> None:
    """Remove the lock user that anchors this LCM slot."""
    user_id = await self._find_user_index_for_slot(slot)
    if user_id is None:
        return
    try:
        await self.async_delete_user(user_id)
    except (LockDisconnected, LockOperationFailed) as err:
        _LOGGER.debug(
            "Lock %s: could not release slot %s anchor: %s",
            self.lock.entity_id,
            slot,
            err,
        )

Wrap in try/except so slot teardown doesn't block on lock connectivity.

Complete Minimal Slot-Only Example

"""Module for SmartLock integration."""

from __future__ import annotations

from dataclasses import dataclass
from datetime import timedelta
from typing import Literal

from homeassistant.config_entries import ConfigEntry, ConfigEntryState

from ..domain.credentials import (
    Credential,
    CredentialRef,
    User,
    user_from_slot,
)
from ..domain.exceptions import (
    CodeRejectedError,
    DuplicateCodeError,
    LockDisconnected,
)
from ..domain.models import SlotCredential
from ._base import BaseLock


@dataclass(repr=False, eq=False)
class SmartLockLock(BaseLock):
    """Class to represent SmartLock lock."""

    @property
    def domain(self) -> str:
        return "smartlock"

    @property
    def usercode_scan_interval(self) -> timedelta:
        return timedelta(minutes=2)

    def is_integration_connected(self) -> bool:
        if self.lock_config_entry:
            return self.lock_config_entry.state == ConfigEntryState.LOADED
        return True

    async def async_get_users(self) -> list[User]:
        try:
            codes = await self._fetch_codes_from_device()
        except Exception as err:
            raise LockDisconnected(str(err)) from err
        return [
            user_from_slot(
                slot,
                SlotCredential.known(pin) if pin else SlotCredential.empty(),
            )
            for slot, pin in codes.items()
        ]

    async def async_set_credential(
        self,
        user_id: int,
        credential: Credential,
        pin: str,
        *,
        name: str | None,
        source: Literal["sync", "direct"],
    ) -> bool:
        try:
            await self._set_code_on_device(credential.slot, pin)
        except DuplicatePinError as err:
            raise DuplicateCodeError(
                code_slot=credential.slot,
                lock_entity_id=self.lock.entity_id,
            ) from err
        except Exception as err:
            raise LockDisconnected(str(err)) from err
        return True

    async def async_delete_credential(self, ref: CredentialRef) -> bool:
        try:
            await self._clear_code_on_device(ref.slot)
        except Exception as err:
            raise LockDisconnected(str(err)) from err
        return True

    # ─── Integration-specific helpers ──────────────────────────────

    async def _fetch_codes_from_device(self) -> dict[int, str]:
        """Replace with your integration's API."""
        raise NotImplementedError

    async def _set_code_on_device(self, slot: int, pin: str) -> None:
        raise NotImplementedError

    async def _clear_code_on_device(self, slot: int) -> None:
        raise NotImplementedError

Testing Your Provider

Unit tests

Create tests/providers/smartlock/test_provider.py. Mirror the structure of an existing provider's tests (tests/providers/zwave_js/test_provider.py for native-user, tests/providers/akuvox/test_provider.py for slot-only).

Key fixtures to exercise:

  • async_get_users() returns the expected User list given seeded device state
  • async_set_credential() calls the integration's API with the right slot + pin and returns True
  • async_delete_credential() calls the integration's API and returns True
  • Exception mapping: device errors become LockDisconnected, duplicate-pin errors become DuplicateCodeError, rejections become CodeRejectedError
  • For native-user: async_set_user() does find-or-create-by-tag; legacy adoption picks up the pre-upgrade user; legacy pass skips users tagged for other slots

Manual testing

  1. Restart Home Assistant after registering the provider.
  2. Configure LCM with a lock using your integration.
  3. Verify:
    • Codes can be set and cleared from LCM
    • The code sensor shows correct values (or "unreadable" for write-only locks)
    • Sync status updates correctly
    • For native-user providers: lock-side users carry lcm:<slot>:<name> tags
    • For push providers: changes made at the keypad show up in LCM in real time
    • Hard refresh detects out-of-band changes

Troubleshooting

"Entity not found" errors: ensure the lock entity exists and is in the entity registry. Check that domain returns the correct integration domain.

Codes not syncing: verify async_get_users() returns valid User objects with pin_credentials populated. Check that async_set_credential() actually writes to the device. Enable debug logging.

Push updates not arriving: ensure supports_push returns True. Verify subscribe_push_updates() is called (the base seam calls it in async_setup). Check that coordinator.push_update({slot: SlotCredential}) is called with the right slot.

For native-user providers, the wrong slot is reported on push: under the user-tag identity model, the lock's user_id and credential_index may differ from the LCM slot. The push handler must resolve the LCM slot via parse_tag on the owning user's name, not from lock-side indices. See the Matter provider's _dispatch_lock_user_change for the canonical pattern.

Debug logging

# configuration.yaml
logger:
  default: info
  logs:
    custom_components.lock_code_manager.providers.smartlock: debug

Next Steps

  • Review existing providers as references:
    • Slot-only: virtual.py, akuvox.py, schlage.py, zha.py, zigbee2mqtt.py
    • Native-user: matter.py, zwave_js.py
  • Read Provider-State-Management for the coordinator + exception contract in depth.
  • Read Architecture for the cross-cutting design (user-tag identity, persistent user anchor lifecycle, masked PINs, duplicate detection).
  • Submit a PR to add your provider to the main repository.
Add a custom footer

Home

Getting Started

UI

Features

Advanced

Development

Troubleshooting

FAQ

Supported Integrations

Clone this wiki locally