Adding a Provider

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:

The lock integration exists in Home Assistant and creates lock.* entities.
The integration provides a way to read, set, and clear PIN credentials.
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 (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). 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. A native-user provider can still serve lock models without user records: the base gates the user lifecycle on async_get_capabilities (supports_user_management and max_user_name_length), which is how the single Z-Wave provider covers both User Code CC and User Credential CC locks.

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 collections.abc import Collection
from dataclasses import dataclass
from datetime import timedelta
import logging
from typing import Literal

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

from ..domain.credentials import (
    Credential,
    CredentialRef,
    CredentialType,
    User,
    WriteResult,
    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)."""
    return False  # slot-only path; see "Native-User Extension" below

Step 3: Implement Connection Check

The base default checks lock_config_entry.state == ConfigEntryState.LOADED. Override only when your integration has a better signal (or when the lock has no config entry, in which case the base default raises and you must override):

async def async_is_integration_connected(self) -> bool:
    """Return whether the integration is reachable."""
    # Ask the integration's own client/gateway object — the ZHA provider
    # looks the device up through the gateway, and the Z-Wave JS provider
    # asks the client whether it is connected and ready.
    return self._client is not None and self._client.connected

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.

The slots parameter scopes which credential indices to report on; None means the slots LCM manages (self.managed_slots). A provider that must ask the lock about one index per round trip MUST honor it; a provider that reads the whole lock in one call may ignore it (the result is allowed to exceed the scope).

async def async_get_users(self, slots: Collection[int] | None = None) -> 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"],
) -> WriteResult:
    """Write a PIN to a slot; return the write outcome.

    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 WriteResult.CONFIRMED

Return WriteResult.CONFIRMED when the lock acknowledged the write, WriteResult.NO_CHANGE if the value was already set, and WriteResult.OPTIMISTIC when the result is ambiguous (write-only/masked locks: the write may have landed but is unconfirmed). The outcome drives the coordinator refresh and the verified/unverified slot lifecycle.

`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, override setup_push_subscription and teardown_push_subscription. (The base's subscribe_push_updates / unsubscribe_push_updates are final wrappers that call these and handle errors — don't override those.)

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

@callback
def setup_push_subscription(self) -> None:
    """Subscribe to real-time updates. Must be idempotent. Raise on failure."""
    if self._push_unsubs:
        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()
        )
        self._push_credential_update(slot, state)

    self._register_push_unsub(self._device.subscribe(on_code_changed))

@callback
def teardown_push_subscription(self) -> None:
    """Unsubscribe. Must be idempotent."""
    self._clear_push_unsubs()

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:

"""Lock provider implementations."""

from __future__ import annotations

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

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

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.

If the lock allocates its own credential index (Matter picks the next free index) rather than addressing credentials by the LCM slot number verbatim, also override credential_index_follows_slot to return False. That property defaults to True and gates the base's slot-within-capacity check and free-slot allocation: when the index is auto-allocated, the slot number is LCM-side bookkeeping with no device-side bound.

`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 user is removed from LCM config (or a lock is dropped from a user), the base seam calls async_release_managed_slot(slot) for the freed slot. The default implementation clears the credential, which is all a slot-only provider needs. Native-user providers override it to delete the persistent user anchor instead — the lock protocol's cascade removes the user's credentials, so don't also clear:

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 collections.abc import Collection
from dataclasses import dataclass
from datetime import timedelta
from typing import Literal

from ..domain.credentials import (
    Credential,
    CredentialRef,
    User,
    WriteResult,
    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)

    # async_is_integration_connected: base default (config entry LOADED)
    # is fine here; override only for a better signal.

    async def async_get_users(
        self, slots: Collection[int] | None = None
    ) -> 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"],
    ) -> WriteResult:
        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 WriteResult.CONFIRMED

    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 WriteResult.CONFIRMED
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

Restart Home Assistant after registering the provider.
Configure LCM with a lock using your integration and add a user with a PIN.
Verify:
- The user's PIN reaches the lock, and clears when the user is disabled or removed
- 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 setup_push_subscription() runs (the base seam calls it, via subscribe_push_updates(), during provider setup and on reconnect). Check that self._push_credential_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.

Home

Getting Started

UI

Add a UI for lock code management
- UI Strategies
- Custom Cards

Features

Advanced

Development

Troubleshooting

FAQ

Supported Integrations

Uh oh!

Adding a Provider

Adding a New Lock Provider

Prerequisites

Choose Your Provider Flavor

Step 1: Create the Provider File

Step 2: Override the Property Surface

Update Intervals

Capability Flags

Step 3: Implement Connection Check

Step 4: Implement Read

Step 5: Implement Write

async_set_credential

async_delete_credential

Step 6: Implement Hard Refresh (Recommended)

Step 7: Add Push Support (Optional)

Step 8: Register the Provider

Native-User Extension (for Matter-class locks)

async_get_capabilities

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

async_delete_user

Slot teardown hook

Complete Minimal Slot-Only Example

Testing Your Provider

Unit tests

Manual testing

Troubleshooting

Debug logging

Next Steps

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Clone this wiki locally

`async_set_credential`

`async_delete_credential`

`async_get_capabilities`

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

`async_delete_user`