Skip to content

Tidy up codebase #106

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
KJonline wants to merge 7 commits into from
Closed

Tidy up codebase #106

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
e4ed948
refactor: replace dictionary access with attribute access for device …
Nov 9, 2025
f4aaa09
refactor: improve debugger code quality and fix trace functionality
Nov 9, 2025
4302121
refactor: remove unnecessary pylint skip-file comments
Nov 9, 2025
d366085
refactor: standardize attribute naming to snake_case
Nov 9, 2025
73a1d01
Merge branch 'dev' into tidy-up-codebase
KJonline Feb 20, 2026
31c0127
Remove alarm and camera modules with associated API endpoints and fun…
KJonline Feb 20, 2026
160ba16
Standardize attribute naming to snake_case convention
KJonline Feb 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 18 additions & 18 deletions src/action.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,21 +34,21 @@ async def getAction(self, device: dict):
"""
dev_data = {}

if device["hiveID"] in self.data["action"]:
if device.hive_id in self.data["action"]:
dev_data = {
"hiveID": device["hiveID"],
"hiveName": device["hiveName"],
"hiveType": device["hiveType"],
"haName": device["haName"],
"haType": device["haType"],
"hiveID": device.hive_id,
"hiveName": device.hive_name,
"hiveType": device.hive_type,
"haName": device.ha_name,
"haType": device.ha_type,
"status": {"state": await self.getState(device)},
"power_usage": None,
"deviceData": {},
"custom": device.get("custom", None),
"custom": getattr(device, "custom", None),
}

self.session.devices.update({device["hiveID"]: dev_data})
return self.session.devices[device["hiveID"]]
self.session.devices.update({device.hive_id: dev_data})
return self.session.devices[device.hive_id]
else:
exists = self.session.data.actions.get("hiveID", False)
if exists is False:
Expand All @@ -67,7 +67,7 @@ async def getState(self, device: dict):
final = None

try:
data = self.session.data.actions[device["hiveID"]]
data = self.session.data.actions[device.hive_id]
final = data["enabled"]
except KeyError as e:
_LOGGER.error(e)
Expand All @@ -87,16 +87,16 @@ async def setStatusOn(self, device: dict):

final = False

if device["hiveID"] in self.session.data.actions:
if device.hive_id in self.session.data.actions:
_LOGGER.debug("Enabling action %s.", device["haName"])
await self.session.hiveRefreshTokens()
data = self.session.data.actions[device["hiveID"]]
data = self.session.data.actions[device.hive_id]
data.update({"enabled": True})
send = json.dumps(data)
resp = await self.session.api.setAction(device["hiveID"], send)
resp = await self.session.api.setAction(device.hive_id, send)
if resp["original"] == 200:
final = True
await self.session.getDevices(device["hiveID"])
await self.session.getDevices(device.hive_id)

return final

Expand All @@ -113,15 +113,15 @@ async def setStatusOff(self, device: dict):

final = False

if device["hiveID"] in self.session.data.actions:
if device.hive_id in self.session.data.actions:
_LOGGER.debug("Disabling action %s.", device["haName"])
await self.session.hiveRefreshTokens()
data = self.session.data.actions[device["hiveID"]]
data = self.session.data.actions[device.hive_id]
data.update({"enabled": False})
send = json.dumps(data)
resp = await self.session.api.setAction(device["hiveID"], send)
resp = await self.session.api.setAction(device.hive_id, send)
if resp["original"] == 200:
final = True
await self.session.getDevices(device["hiveID"])
await self.session.getDevices(device.hive_id)

return final
135 changes: 0 additions & 135 deletions src/alarm.py
View file
Open in desktop

This file was deleted.

85 changes: 11 additions & 74 deletions src/api/hive_api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,6 @@ class HiveApi:

def __init__(self, hiveSession=None, websession=None, token=None):
"""Hive API initialisation."""
self.cameraBaseUrl = "prod.hcam.bgchtest.info"
self.urls = {
"properties": "https://sso.hivehome.com/",
"login": "https://beekeeper.hivehome.com/1.0/cognito/login",
Expand All @@ -25,9 +24,6 @@ def __init__(self, hiveSession=None, websession=None, token=None):
"weather": "https://weather.prod.bgchprod.info/weather",
"holiday_mode": "/holiday-mode",
"all": "/nodes/all?products=true&devices=true&actions=true",
"alarm": "/security-lite?homeId=",
"cameraImages": f"https://event-history-service.{self.cameraBaseUrl}/v1/events/cameras?latest=true&cameraId={{0}}",
"cameraRecordings": f"https://event-history-service.{self.cameraBaseUrl}/v1/playlist/cameras/{{0}}/events/{{1}}.m3u8",
"devices": "/devices",
"products": "/products",
"actions": "/actions",
Expand All @@ -41,36 +37,20 @@ def __init__(self, hiveSession=None, websession=None, token=None):
self.session = hiveSession
self.token = token

def request(self, type, url, jsc=None, camera=False):
def request(self, type, url, jsc=None):
"""Make API request."""
if self.session is not None:
if camera:
self.headers = {
"content-type": "application/json",
"Accept": "*/*",
"Authorization": f"Bearer {self.session.tokens.tokenData['token']}",
"x-jwt-token": self.session.tokens.tokenData["token"],
}
else:
self.headers = {
"content-type": "application/json",
"Accept": "*/*",
"authorization": self.session.tokens.tokenData["token"],
}
self.headers = {
"content-type": "application/json",
"Accept": "*/*",
"authorization": self.session.tokens.token_data["token"],
}
else:
if camera:
self.headers = {
"content-type": "application/json",
"Accept": "*/*",
"Authorization": f"Bearer {self.token}",
"x-jwt-token": self.token,
}
else:
self.headers = {
"content-type": "application/json",
"Accept": "*/*",
"authorization": self.token,
}
self.headers = {
"content-type": "application/json",
"Accept": "*/*",
"authorization": self.token,
}

if type == "GET":
return requests.get(
Expand Down Expand Up @@ -99,7 +79,6 @@ def refreshTokens(self, tokens={}):
if "token" in data and self.session:
self.session.updateTokens(data)
self.urls.update({"base": data["platform"]["endpoint"]})
self.urls.update({"camera": data["platform"]["cameraPlatform"]})
self.json_return.update({"original": info.status_code})
self.json_return.update({"parsed": info.json()})
except (OSError, RuntimeError, ZeroDivisionError):
Expand Down Expand Up @@ -143,48 +122,6 @@ def getAll(self):

return json_return

def getAlarm(self, homeID=None):
"""Build and query alarm endpoint."""
if self.session is not None:
homeID = self.session.config.homeID
url = self.urls["base"] + self.urls["alarm"] + homeID
try:
info = self.request("GET", url)
self.json_return.update({"original": info.status_code})
self.json_return.update({"parsed": info.json()})
except (OSError, RuntimeError, ZeroDivisionError):
self.error()

return self.json_return

def getCameraImage(self, device=None, accessToken=None):
"""Build and query camera endpoint."""
json_return = {}
url = self.urls["cameraImages"].format(device["props"]["hardwareIdentifier"])
try:
info = self.request("GET", url, camera=True)
json_return.update({"original": info.status_code})
json_return.update({"parsed": info.json()})
except (OSError, RuntimeError, ZeroDivisionError):
self.error()

return json_return

def getCameraRecording(self, device=None, eventId=None):
"""Build and query camera endpoint."""
json_return = {}
url = self.urls["cameraRecordings"].format(
device["props"]["hardwareIdentifier"], eventId
)
try:
info = self.request("GET", url, camera=True)
json_return.update({"original": info.status_code})
json_return.update({"parsed": info.text.split("\n")[3]})
except (OSError, RuntimeError, ZeroDivisionError):
self.error()

return json_return

def getDevices(self):
"""Call the get devices endpoint."""
url = self.urls["base"] + self.urls["devices"]
Expand Down
Loading
Loading