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:

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

from homeassistant.core import callback

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

_LOGGER = logging.getLogger(__name__)


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

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

The package doesn't use from __future__ import annotations. Providers may import from model/ but never from ha/; tests/test_architecture.py fails the build if they do.

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)

usercode_scan_interval is the poll for providers that don't push. hard_refresh_interval schedules a drift read whether or not the provider pushes; None disables it. For integrations whose config-entry state changes signal reconnects, set connection_check_interval = None.

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 core reads through this (via the base's async_get_usercodes(slots) projection) to check every write it makes, so this read must ask the lock, never a cache. If your integration's plain read is cached, set the class attribute fresh_read_is_hard_refresh = True and make async_hard_refresh_codes(slots) (Step 6) the fresh read; the core then reads through that instead.

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()
    ]

Choose each slot's state carefully; the wrong one is a defect:

• SlotCredential.known(pin) — the lock holds this code and is accepting it. Never report a code the lock is refusing this way.
• SlotCredential.empty() — the lock said the slot is free. LCM treats it as proof a clear landed, and allocation hands the slot number out.
• SlotCredential.unreadable() — something is there that can't be read or trusted: a write-only PIN, a disabled code, a status you can't interpret.

When in doubt, unreadable().

If the lock answers a read without saying anything about a slot, set self.last_silent_slots to those slots during the read; if it answered that it cannot tell what a slot holds (a write-only lock), put them in self.last_cannot_tell_slots too. See Provider State Management for what the core does with them.

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 once the lock acknowledges the write. Map the integration's errors to the typed exceptions: CodeRejectedError / DuplicateCodeError for a PIN the lock refuses, LockDisconnected / LockOperationFailed for failures that may pass. Don't raise bare exceptions; the core treats anything untyped as a provider bug.

The full contract (each argument, the other WriteResult values) is in Provider State Management, and what each exception costs is in Provider State Management.

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

Return False only when the lock answered that there was nothing to clear; when you can't tell, return True. Full contract: Provider State Management.

Step 6: Implement Hard Refresh

The hard_refresh_usercodes service calls async_hard_refresh_codes() with no arguments to re-read the whole lock, and the base raises ProviderNotImplementedError if you don't implement it. slots names the slots to re-read from the device (None means everything); the return value is always the full projection, every managed slot included.

@override
async def async_hard_refresh_codes(
    self, slots: Collection[int] | None = None
) -> 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()

If you set fresh_read_is_hard_refresh = True, the core reads through this method with the slots it is checking, so re-read at least those from the device and make sure each of them is in the map you return. A slot missing from the answer reads as silent.

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, also record the use with async_fire_code_slot_event. Pass the slot, the direction the lock moved (None when you can't tell), and the kind of credential the lock said was presented; credential_type is required. See Provider State Management.

@callback
def _handle_lock_event(self, event_data) -> None:
    """Record a code use for automations."""
    self.async_fire_code_slot_event(
        code_slot=event_data["slot"],
        to_locked=event_data.get("locked"),
        credential_type=CredentialType.PIN,
    )

If a push handler has to ask the lock something before it can push (Matter reads the user list to resolve an event), run that call with await self.coordinator.async_in_turn(call) so it takes its turn with the lock's other exchanges.

Step 8: Register the Provider

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

"""Lock provider implementations."""

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...
}

The map is keyed by the lock entity's platform. Locks published through MQTT (Zigbee2MQTT, zwave-js-ui) all have platform mqtt, so they are not in the map: resolve_provider_class picks the provider from the device registry identifier the bridge published. A new MQTT bridge provider subclasses BaseMqttLock (providers/_mqtt.py) and adds a rule there.

Native-User Extension (for Matter-class locks)

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

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 run the user lifecycle and to size the tagged user name.

from ..model.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),
    )

Full contract: Provider State Management.

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

The base passes user.name already tagged (lcm:<slot>:<display>). Find the lock user LCM owns for that slot, update it or create one, and return its user_id and whether you created it:

from ..model.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, _ = parse_tag(user.name or "")
    if slot is None:
        slot = user.user_id
    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,
    )

_find_user_index_for_slot(slot, *, adopt_untagged=True) is yours to write. It looks for the user self.slot_tag(user) says is tagged for the slot, and, on the set path only, falls back to adopting the untagged user your lock used for that slot before 4.0.0. The two passes, the rules for each, and an example are in Provider State Management.

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

Full contract: Provider State Management.

Slot teardown hook

When a user is removed from LCM config, or a lock is removed from an entry, the core releases each freed slot by calling async_release_managed_slot(slot, written=...). The default clears the slot, which is all a slot-only provider needs.

A native-user provider overrides it to remove only what LCM wrote: find the user by its LCM tag (_find_user_index_for_slot(slot, adopt_untagged=False)), remove LCM's PIN, and delete the user only when that PIN is all it holds; otherwise rename the user so it carries no tag. Never touch an untagged user, and on a lock without user records call super(). Let errors propagate: a failure you catch and log reads as a release that worked, and the code stays on the lock.

The full rules and an example are in Provider State Management; ZWaveJSLock.async_release_managed_slot in providers/zwave_js.py is the reference implementation.

Complete Minimal Slot-Only Example

The whole provider in one file

"""Module for SmartLock integration."""

from collections.abc import Collection
from dataclasses import dataclass
from datetime import timedelta
from typing import Literal, override

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


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

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

    @property
    @override
    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.

    @override
    async def async_get_users(
        self, slots: Collection[int] | None = None
    ) -> list[User]:
        try:
            codes = await self._fetch_codes_from_device()
        except SmartLockConnectionError 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()
        ]

    @override
    async def async_hard_refresh_codes(
        self, slots: Collection[int] | None = None
    ) -> dict[int, SlotCredential]:
        # async_get_users asks the device every time: nothing to bypass.
        return await self.async_get_usercodes()

    @override
    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 SmartLockDuplicateError as err:
            raise DuplicateCodeError(
                code_slot=credential.slot,
                lock_entity_id=self.lock.entity_id,
            ) from err
        except SmartLockRejectedError as err:
            raise CodeRejectedError(
                code_slot=credential.slot,
                lock_entity_id=self.lock.entity_id,
                reason=str(err),
            ) from err
        except SmartLockConnectionError as err:
            raise LockDisconnected(str(err)) from err
        return WriteResult.CONFIRMED

    @override
    async def async_delete_credential(self, ref: CredentialRef) -> bool:
        try:
            await self._clear_code_on_device(ref.slot)
        except SmartLockConnectionError 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

SmartLockConnectionError, SmartLockDuplicateError and SmartLockRejectedError stand for your integration's own exception types. Catch those, not Exception: anything untyped that escapes is treated as a provider bug, which is what it should be.

Testing Your Provider

Tests

Add tests/providers/smartlock/ with:

• conftest.py — fixtures that set up the integration's lock (usually against a fake of its API: see fake_service.py, fake_node.py or fake_bridge.py in the existing provider test folders) and an LCM entry using it.
• test_provider.py — the provider in isolation:
  • async_get_users() returns the expected User list given seeded device state, with the right SlotCredential state for each kind of slot (code, empty, masked, disabled)
  • async_set_credential() calls the integration's API with the right slot + PIN
  • async_delete_credential() calls the integration's API and returns whether anything changed
  • Exception mapping: transport 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 and skips users tagged for other slots; async_release_managed_slot() removes only LCM's PIN, leaves untagged users alone, and lets errors propagate
• test_e2e.py — the full lifecycle through Home Assistant: add a user, see the PIN reach the lock and the slot report in sync, change and disable it, remove the user and see the code released.

Mirror an existing provider: tests/providers/akuvox/ or tests/providers/virtual/ for slot-only, tests/providers/zwave_js/ or tests/providers/matter/ for native-user. Run only the tests you touched locally; CI runs the full suite, including tests/test_architecture.py.

Manual testing

1. Restart Home Assistant after registering the provider.
2. Configure LCM with a lock using your integration and add a user with a PIN.
3. Verify:
   • The user's PIN reaches the lock, and clears when the user is disabled or removed
   • The code sensor shows correct values (or unknown for write-only locks until the write is trusted)
   • 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_usercodes picks up 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 and that the platform is in INTEGRATIONS_CLASS_MAP.

Codes not syncing: verify async_get_users() returns valid User objects with pin_credentials populated, and that it reads the device rather than a cache — a stale read makes every write look failed, and the slot is held after three tries. Check that async_set_credential() actually writes to the device. Enable debug logging.

Lock shown offline or backing off: every exchange that raises LockDisconnected / LockOperationFailed, or a read where every slot is silent, counts as no answer. Check that a successful read reports the slots the lock described.

Push updates not arriving: ensure supports_push returns True. Verify setup_push_subscription() runs (the base 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 from the owning user's tag, not from lock-side indices, and must count a bare-digit name as LCM's only where LCM could have written it (self.slot_tag(user) applies that rule to a User). 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
    # every exchange the lock's worker makes, and what each provider call took
    custom_components.lock_code_manager.sync.worker: debug
    custom_components.lock_code_manager.ha.adapter: 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
  • MQTT bridges: _mqtt.py (BaseMqttLock), zigbee2mqtt.py, zwave_js_ui.py
• Read Provider-State-Management for the read rules, hooks and exception contract in depth.
• Read Architecture for the cross-cutting design (the lock worker, user-tag identity, releasing slots, masked PINs, duplicate detection).
• Submit a PR to add your provider to the main repository.