I am working on a custom component to integrate my bosma door locks. I have checked all python code for syntax errors and am not finding anything. I am currently getting: Integration error: bosmalock - Integration ‘bosmalock’ not found. In config I have created a custom_components folder and inside that added another folder named bosmalock. I placed my init.py, lock.py, manifest.json and api.py into this directory. Then I also placed an entry in secrets.yaml with my api key. I keep getting this error ehen I update my configuration.yaml to use this custom component. What could be Causing this?
Any help would be appriciated.
I am wondering if we are approching this in the wrong way. while it would be cool to keep the cloud functionality involved I did dig up FCC filings and show the gateway is a ESP32 C3 https://fccid.io/2AEZA-GWBOB01/Internal-Photos/EUT-Internal-Photo-5978359 and that the lock uses a BK3431Q https://fccid.io/2AEZA-DLBOB02/Internal-Photos/Internal-Photo-5620418 So I wonder if we figure out how the gateway talks to the lock we can trigger the lock FROM the gateway Im new to all of this and know nothing about programming so I might just be spit balling here but I am getting closer and closer to just trying to learn this just so I can do this. I knew going into buying this thing I was gonna hate the cloud aspect of it but at the time smart locks werent a huge thing so I had to take the good with the evil on it. hopefully over time I learn and we can put our heads together in figuring this thing out together but maybe you get it before I learn in which case I owe you a beer!
1 Like
Ya, I hadn’t thought to do it that way. I feel like I am getting pretty close here, or at least meeting all requirements laid out in the API documentation. I’m thinking I found a solution to the issue I was having loading the custom component. My manifest.json was done in windows CRLF instead of Unix LF. I got it converted, but I haven’t got a chance to test it yet. I’m hoping it should fix that issue at least. I am still pretty new to coding as well. Defiantly let me know what you find out and if I am able to get this all working I will be posting it on Github. I have almost given up on this a few times, but I feel like I have sunk too much time into it to turn back now.
1 Like
Here is an (all files included) overview of this custom_component. I feel that it is close but it is still causing errors within home assistant. I am no expert working with this stuff. Hopefully someone with a bit more knowledge and skill could make this work one day.
manifest.json
{
“domain”: “bosmalock”,
“name”: “bosmalock”,
“documentation”: “https://www.example.com”,
“dependencies”: ,
“codeowners”: [“zperkd”],
“version”: “1.0.0”
}
init.py
from homeassistant.const import CONF_API_KEY, CONF_DEVICES
from .api import BosmaAPIClient
from .lock import BosmaLock
def setup(hass, config):
bosma_config = config[‘bosmalock’]
api_client = BosmaAPIClient(bosma_config[CONF_API_KEY], ‘Your Company ID’) # Replace with your company ID
for device in bosma_config[CONF_DEVICES]:
hass.data[f"bosmalock_{device[‘name’]}"] = BosmaLock(device[‘name’], device[‘device_id’], api_client)
return True
api.py
import hashlib
import requests
import time
import uuid
import logging
class BosmaAPIClient:
def init(self, api_secret, company_id):
self.api_secret = ‘your_api_secret’ # Replace with your actual API secret
self.company_id = ‘your_company_id’ # Replace with your company ID provided by Bosma
self.base_url = “https://www.bosma-iot.com/bosma-smart-global”
def generatesignature(self, nonce, timestamp):
sign_string = f"nonce={nonce}×tamp={timestamp}{self.api_secret}"
return hashlib.md5(sign_string.encode('utf-8')).hexdigest()
def generatenonce(self):
return uuid.uuid4().hex
def getheaders(self):
nonce = self.generate_nonce()
timestamp = str(int(time.time()))
signature = self.generate_signature(nonce, timestamp)
return {
"corp": self.company_id,
"nonce": nonce,
"timestamp": timestamp,
"sign": signature
}
def get_token(self, app_package, platform, openid, country):
url = f"{self.base_url}/sdk/globalcorpuser/applyToken"
headers = self.get_headers()
params = {
"appPackage": 'your_app_package', # Replace with your app package path
"platform": 2, # Replace with 1 for iOS or 2 for Android
"openid": 'your_openid', # Replace with the user ID of the client platform
"country": 'your_country' # Replace with the user registration country ID
}
response = requests.post(url, headers=headers, params=params)
if response.status_code == 200:
return response.json().get('data', {}).get('ticket')
else:
logging.error(f"Error getting token: {response.text}")
return None
Implement methods for locking, unlocking, and checking lock status
lock.py
custom_components/bosmalock/lock.py
import aiohttp
from homeassistant.components.lock import LockEntity, LockEntityFeature
from homeassistant.core import HomeAssistant
Constants for Bosma API
API_BASE_URL = “https://api.bosma.com”
LOCK_ENDPOINT = “/path/to/lock/endpoint” # Replace with the actual lock endpoint
UNLOCK_ENDPOINT = “/path/to/unlock/endpoint” # Replace with the actual unlock endpoint
BOSMA_API_KEY = “YOUR_API_KEY_PLACEHOLDER”
class BosmaLockEntity(LockEntity):
“”“Represent a Bosma lock.”“”
def __init__(self, hass: HomeAssistant, lock_id: str):
self.hass = hass
self.lock_id = lock_id
self._is_locked = None
@property
def supported_features(self):
"""Flag supported features."""
return LockEntityFeature.LOCK | LockEntityFeature.UNLOCK
@property
def is_locked(self):
"""Return true if the lock is locked."""
return self._is_locked
async def async_lock(self, **kwargs):
"""Lock the device."""
url = f"{API_BASE_URL}{LOCK_ENDPOINT}"
headers = {'Authorization': f'Bearer {BOSMA_API_KEY}'}
data = {'lock_id': self.lock_id}
async with aiohttp.ClientSession() as session:
async with session.post(url, headers=headers, json=data) as response:
if response.status == 200:
self._is_locked = True
else:
# Handle error responses
pass
self.async_write_ha_state()
async def async_unlock(self, **kwargs):
"""Unlock the device."""
url = f"{API_BASE_URL}{UNLOCK_ENDPOINT}"
headers = {'Authorization': f'Bearer {BOSMA_API_KEY}'}
data = {'lock_id': self.lock_id}
async with aiohttp.ClientSession() as session:
async with session.post(url, headers=headers, json=data) as response:
if response.status == 200:
self._is_locked = False
else:
# Handle error responses
pass
self.async_write_ha_state()
Example usage remains the same
secrets.yaml
secrets.yaml
bosma_api_key: “YOUR_API_KEY_PLACEHOLDER” # Replace with your actual Bosma API key
const.py
custom_components/bosma/const.py
Replace these with your actual API key and URL
BOSMA_API_KEY = “YOUR_API_KEY_PLACEHOLDER”
BOSMA_API_URL = “https://api.bosma.com”
hardware.py
import requests
from homeassistant.exceptions import PlatformNotReady
Assuming ‘BOSMA_API_KEY’ and ‘BOSMA_API_URL’ are constants defined in your component
from .const import BOSMA_API_KEY, BOSMA_API_URL
def check_hardware_status(lock_id):
“”“Check the status of the lock’s hardware components.”“”
url = f"{BOSMA_API_URL}/hardwareStatus"
headers = {
‘Authorization’: f’Bearer {BOSMA_API_KEY}',
‘Content-Type’: ‘application/json’
}
payload = {‘lock_id’: lock_id}
try:
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status()
# Assuming the API returns a JSON response
data = response.json()
# You might want to process the data or directly return it
return data
except requests.exceptions.HTTPError as errh:
# Handle specific HTTP errors
print(f"HTTP Error: {errh}“)
except requests.exceptions.ConnectionError as errc:
# Handle connection errors
print(f"Error Connecting: {errc}”)
except requests.exceptions.Timeout as errt:
# Handle timeout errors
print(f"Timeout Error: {errt}“)
except requests.exceptions.RequestException as err:
# Handle any other request errors
print(f"Error: {err}”)
# Return None or a default value if an error occurs
return None
You can test this function with a specific lock_id
test_lock_id = ‘your_lock_id_here’
print(check_hardware_status(test_lock_id))
media_source.py
import requests
from homeassistant.exceptions import PlatformNotReady
custom_components/bosmalock/media_source.py
from homeassistant.components.media_source import AsyncMediaSource
from homeassistant.core import HomeAssistant
class BosmaMediaSource(AsyncMediaSource):
“”“A media source implementation for Bosma locks.”“”
name = "Bosma Lock Media"
def __init__(self, hass: HomeAssistant):
self.hass = hass
async def async_browse_media(self, media_content_id=None):
"""Return a browsable media collection for the Bosma lock."""
# Implement retrieval and formatting of media data
# This could be camera feeds, recorded videos, etc.
pass
async def async_resolve_media(self, media_content_id):
"""Resolve media to a URL."""
# Implement the method to resolve a media item to a playable URL
pass
async def async_get_media_source(hass: HomeAssistant):
“”“Set up Bosma media source.”“”
return BosmaMediaSource(hass)
system_health.py
import requests
from homeassistant.exceptions import PlatformNotReady
Assuming ‘BOSMA_API_KEY’ and ‘BOSMA_API_URL’ are constants defined in your component
from .const import BOSMA_API_KEY, BOSMA_API_URL
def report_system_health(lock_id):
“”“Report the health of the Bosma lock system.”“”
url = f"{BOSMA_API_URL}/systemHealth"
headers = {
‘Authorization’: f’Bearer {BOSMA_API_KEY}',
‘Content-Type’: ‘application/json’
}
payload = {‘lock_id’: lock_id}
try:
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status()
# Assuming the API returns JSON data with health metrics
health_data = response.json()
# Process and return the health data
# This might include metrics like connectivity status, battery level, etc.
return health_data
except requests.exceptions.HTTPError as errh:
print(f"HTTP Error: {errh}“)
except requests.exceptions.ConnectionError as errc:
print(f"Error Connecting: {errc}”)
except requests.exceptions.Timeout as errt:
print(f"Timeout Error: {errt}“)
except requests.exceptions.RequestException as err:
print(f"Error: {err}”)
return None
Test the function with a specific lock_id
test_lock_id = ‘your_lock_id_here’
print(report_system_health(test_lock_id))
significant_change.py
import requests
from homeassistant.exceptions import PlatformNotReady
Assuming ‘BOSMA_API_KEY’ and ‘BOSMA_API_URL’ are constants defined in your component
from .const import BOSMA_API_KEY, BOSMA_API_URL
def detect_significant_change(lock_id):
“”“Detect and handle significant changes in lock state.”“”
url = f"{BOSMA_API_URL}/getStateChange"
headers = {
‘Authorization’: f’Bearer {BOSMA_API_KEY}',
‘Content-Type’: ‘application/json’
}
payload = {‘lock_id’: lock_id}
try:
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status()
# Assuming the API returns JSON data with state change information
state_change_data = response.json()
# Analyze the data to determine if a significant change has occurred
# This could involve checking for specific states or alerts
# Example:
# if state_change_data[‘status’] == ‘security_breach’:
# handle_security_breach(lock_id)
return state_change_data
except requests.exceptions.HTTPError as errh:
print(f"HTTP Error: {errh}“)
except requests.exceptions.ConnectionError as errc:
print(f"Error Connecting: {errc}”)
except requests.exceptions.Timeout as errt:
print(f"Timeout Error: {errt}“)
except requests.exceptions.RequestException as err:
print(f"Error: {err}”)
return None
Example function to handle a specific type of significant change
def handle_security_breach(lock_id):
# Implement the response to a security breach
# This could involve sending alerts, triggering automations, etc.
pass
Test the function with a specific lock_id
test_lock_id = ‘your_lock_id_here’
print(detect_significant_change(test_lock_id))
group.py
import requests
from homeassistant.exceptions import PlatformNotReady
Assuming ‘BOSMA_API_KEY’ and ‘BOSMA_API_URL’ are constants defined in your component
from .const import BOSMA_API_KEY, BOSMA_API_URL
def manage_group(group_id, lock_ids, action):
“”“Manage a group of locks.”“”
url = f"{BOSMA_API_URL}/manageGroup"
headers = {
‘Authorization’: f’Bearer {BOSMA_API_KEY}',
‘Content-Type’: ‘application/json’
}
payload = {
‘group_id’: group_id,
‘lock_ids’: lock_ids,
‘action’: action # action could be ‘lock’, ‘unlock’, etc.
}
try:
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status()
# Assuming the API returns a response indicating success or failure
return response.json()
except requests.exceptions.HTTPError as errh:
print(f"HTTP Error: {errh}“)
except requests.exceptions.ConnectionError as errc:
print(f"Error Connecting: {errc}”)
except requests.exceptions.Timeout as errt:
print(f"Timeout Error: {errt}“)
except requests.exceptions.RequestException as err:
print(f"Error: {err}”)
return None
Example usage
group_id = ‘example_group_id’
lock_ids = [‘lock1_id’, ‘lock2_id’]
action = ‘lock’ # or ‘unlock’
print(manage_group(group_id, lock_ids, action))
recorder.py
from homeassistant.core import HomeAssistant
from homeassistant.components.recorder import Recorder
def record_state_change(hass: HomeAssistant, entity_id: str, new_state: str):
“”“Record a state change for a Bosma lock.”“”
# Assuming entity_id is the entity representing the Bosma lock in Home Assistant
# Get the current state of the entity
current_state = hass.states.get(entity_id)
# Create a new state object with the new state
new_state_obj = hass.states.set(entity_id, new_state, current_state.attributes)
# Check if recorder is set up and working
if isinstance(hass.components.recorder, Recorder):
hass.components.recorder.record_state_change(entity_id, current_state, new_state_obj)
else:
# Log an error or handle cases where the recorder isn’t available
pass
Example usage:
Assuming ‘lock.bosma_front_door’ is the entity ID of a Bosma lock
record_state_change(hass, ‘lock.bosma_front_door’, ‘unlocked’)
diagnostics.py
Constants for Bosma API
API_BASE_URL = “https://api.bosma.com”
API_ENDPOINT_DIAGNOSTICS = “/getDiagnostics”
Replace ‘BOSMA_API_KEY’ with your actual Bosma API key
BOSMA_API_KEY = “YOUR_API_KEY_PLACEHOLDER”
LOGGER = logging.getLogger(name)
async def get_diagnostics(hass, lock_id):
“”“Retrieve diagnostic information for a specific Bosma lock asynchronously.”“”
url = f"{API_BASE_URL}{API_ENDPOINT_DIAGNOSTICS}"
headers = {
‘Authorization’: f’Bearer {BOSMA_API_KEY}',
‘Content-Type’: ‘application/json’
}
params = {‘lockId’: lock_id}
# Use Home Assistant’s aiohttp client session for asynchronous HTTP requests
session = async_get_clientsession(hass)
try:
response = await session.get(url, headers=headers, params=params)
response.raise_for_status()
return await response.json() # Assuming the API returns JSON formatted diagnostics
except requests.HTTPError as http_err:
# Handle HTTP errors (e.g., response code 4xx, 5xx)
LOGGER.error(f"HTTP error occurred: {http_err}“)
except Exception as err:
# Handle other possible errors (e.g., network issues)
LOGGER.error(f"Error occurred: {err}”)
return None
Example usage:
diagnostics = await get_diagnostics(hass, ‘your_lock_id’)
print(diagnostics)
logbook.py
custom_components/bosmalock/logbook.py
import logging
from homeassistant.core import HomeAssistant, Event
LOGGER = logging.getLogger(name)
def create_logbook_entry(hass: HomeAssistant, lock_id: str, message: str):
“”“Fire an event in Home Assistant to create a logbook entry.”“”
event_data = {
“lock_id”: lock_id,
“message”: message
}
hass.bus.fire(“bosmalock_logbook_entry”, event_data)
def setup(hass: HomeAssistant):
“”“Set up the logbook event listener.”“”
def handle_logbook_entry(event: Event):
“”“Handle the event to create a logbook entry.”“”
lock_id = event.data.get(“lock_id”)
message = event.data.get(“message”)
if not lock_id or not message:
LOGGER.error(“Missing lockid or message in logbook entry event”)
return
# Add the logbook entry
hass.components.logbook.log_entry(
name=“Bosma Lock”,
message=message,
entity_id=f"lock.bosmalock_{lock_id}",
domain=“bosmalock”
)
hass.bus.listen(“bosmalock_logbook_entry”, handle_logbook_entry)
return True
1 Like
I applied for the IoT thing thru bosma when you first posted and havent heard back, I also reattempted to sign up when you post this last part. I am still waiting to hear back how long did it take for you to get yours? communication doesnt seem like bosmas strong suit 
I did have to reach out to them a few times on their dev portal. I also used the contact us section on their main site. They seem to respond a bit quicker from there. It did take a little over a week, and there were several steps I had to complete and be approved for.