codey/api
1
0
Fork 0
Code Issues Pull Requests Activity

minor/formatting

Browse Source
This commit is contained in:
codey
2025-10-02 10:45:30 -04:00
parent fa3c8e8861
commit eae490ddde
3 changed files with 114 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
base.py
View File

@@ -101,7 +101,9 @@ routes: dict = {
"meme": importlib.import_module("endpoints.meme").Meme(app, util, constants),
"trip": importlib.import_module("endpoints.rip").RIP(app, util, constants),
"auth": importlib.import_module("endpoints.auth").Auth(app),
"lighting": importlib.import_module("endpoints.lighting").Lighting(app, util, constants),
"lighting": importlib.import_module("endpoints.lighting").Lighting(
app, util, constants
),
}
# Misc endpoint depends on radio endpoint instance

154
endpoints/lighting.py
View File

@@ -10,8 +10,8 @@ import redis
from lyric_search.sources import private
from auth.deps import get_current_user
from dotenv import load_dotenv
from pycync.user import User # type: ignore
from pycync.cync import Cync as Cync # type: ignore
from pycync.user import User # type: ignore
from pycync.cync import Cync as Cync # type: ignore
from pycync import Auth # type: ignore
from pycync.exceptions import TwoFactorRequiredError, AuthFailedError # type: ignore
@@ -28,26 +28,35 @@ class Lighting(FastAPI):
if not self.cync_device_name:
missing_vars.append("CYNC_DEVICE_NAME")
if missing_vars:
raise Exception(f"Missing required environment variables: {', '.join(missing_vars)}")
raise Exception(
f"Missing required environment variables: {', '.join(missing_vars)}"
)
# Cast to str after check to silence linter
cync_email: str = self.cync_email # type: ignore
cync_password: str = self.cync_password # type: ignore
# Check if session is closed or missing
if not self.session or getattr(self.session, 'closed', False):
if not self.session or getattr(self.session, "closed", False):
self.session = aiohttp.ClientSession()
cached_user = self._load_cached_user()
if cached_user:
self.auth = Auth(session=self.session,
user=cached_user,
username=cync_email,
password=cync_password)
self.auth = Auth(
session=self.session,
user=cached_user,
username=cync_email,
password=cync_password,
)
else:
self.auth = Auth(session=self.session, username=cync_email,
password=cync_password)
self.auth = Auth(
session=self.session, username=cync_email, password=cync_password
)
# Try to refresh token
self.cync_user = None
if self.auth.user and hasattr(self.auth.user, 'expires_at') and self.auth.user.expires_at > time.time():
if (
self.auth.user
and hasattr(self.auth.user, "expires_at")
and self.auth.user.expires_at > time.time()
):
try:
await self.auth.async_refresh_user_token()
self.cync_user = self.auth.user
@@ -60,10 +69,12 @@ class Lighting(FastAPI):
self.cync_user = await self.auth.login()
self._save_cached_user(self.cync_user)
except TwoFactorRequiredError:
logging.error("Cync 2FA required. Set CYNC_2FA_CODE in env if needed.")
logging.error(
"Cync 2FA required. Set CYNC_2FA_CODE in env if needed."
)
raise Exception("Cync 2FA required.")
except AuthFailedError as e:
logging.error(f"Failed to authenticate with Cync API: {e}")
logging.error("Failed to authenticate with Cync API: %s", e)
raise Exception("Cync authentication failed.")
self.cync_api = await Cync.create(self.auth)
# Also check if cync_api is None (shouldn't happen, but just in case)
@@ -72,6 +83,7 @@ class Lighting(FastAPI):
logging.critical("self.auth: %s", self.auth)
return
self.cync_api = await Cync.create(self.auth)
"""
Lighting Endpoints
"""
@@ -82,7 +94,9 @@ class Lighting(FastAPI):
self.app: FastAPI = app
self.util = util
self.constants = constants
self.redis_client = redis.Redis(password=private.REDIS_PW, decode_responses=True)
self.redis_client = redis.Redis(
password=private.REDIS_PW, decode_responses=True
)
self.lighting_key = "lighting:state"
# Cync config
@@ -107,20 +121,31 @@ class Lighting(FastAPI):
if not self.cync_device_name:
missing_vars.append("CYNC_DEVICE_NAME")
if missing_vars:
raise Exception(f"Missing required environment variables: {', '.join(missing_vars)}")
raise Exception(
f"Missing required environment variables: {', '.join(missing_vars)}"
)
self.session = aiohttp.ClientSession()
cached_user = self._load_cached_user()
if cached_user:
self.auth = Auth(session=self.session, user=cached_user,
username=self.cync_email or "",
password=self.cync_password or "")
self.auth = Auth(
session=self.session,
user=cached_user,
username=self.cync_email or "",
password=self.cync_password or "",
)
else:
self.auth = Auth(session=self.session,
username=self.cync_email or "",
password=self.cync_password or "")
self.auth = Auth(
session=self.session,
username=self.cync_email or "",
password=self.cync_password or "",
)
# Try to refresh token
if self.auth.user and hasattr(self.auth.user, 'expires_at') and self.auth.user.expires_at > time.time():
if (
self.auth.user
and hasattr(self.auth.user, "expires_at")
and self.auth.user.expires_at > time.time()
):
try:
await self.auth.async_refresh_user_token()
self.cync_user = self.auth.user
@@ -133,10 +158,12 @@ class Lighting(FastAPI):
self.cync_user = await self.auth.login()
self._save_cached_user(self.cync_user)
except TwoFactorRequiredError:
logging.error("Cync 2FA required. Set CYNC_2FA_CODE in env if needed.")
logging.error(
"Cync 2FA required. Set CYNC_2FA_CODE in env if needed."
)
raise Exception("Cync 2FA required.")
except AuthFailedError as e:
logging.error(f"Failed to authenticate with Cync API: {e}")
logging.error("Failed to authenticate with Cync API: %s", e)
raise Exception("Cync authentication failed.")
# Create persistent Cync API object
self.cync_api = await Cync.create(self.auth)
@@ -151,7 +178,10 @@ class Lighting(FastAPI):
handler,
methods=["GET"],
include_in_schema=True,
dependencies=[Depends(RateLimiter(times=10, seconds=2)), Depends(get_current_user)],
dependencies=[
Depends(RateLimiter(times=10, seconds=2)),
Depends(get_current_user),
],
)
app.add_api_route(
@@ -159,7 +189,10 @@ class Lighting(FastAPI):
self.set_lighting_state,
methods=["POST"],
include_in_schema=True,
dependencies=[Depends(RateLimiter(times=10, seconds=2)), Depends(get_current_user)],
dependencies=[
Depends(RateLimiter(times=10, seconds=2)),
Depends(get_current_user),
],
)
def _load_cached_user(self):
@@ -172,10 +205,10 @@ class Lighting(FastAPI):
refresh_token=data["refresh_token"],
authorize=data["authorize"],
user_id=data["user_id"],
expires_at=data["expires_at"]
expires_at=data["expires_at"],
)
except Exception as e:
logging.warning(f"Failed to load cached Cync user: {e}")
logging.warning("Failed to load cached Cync user: %s", e)
return None
def _save_cached_user(self, user):
@@ -185,13 +218,13 @@ class Lighting(FastAPI):
"refresh_token": user.refresh_token,
"authorize": user.authorize,
"user_id": user.user_id,
"expires_at": user.expires_at
"expires_at": user.expires_at,
}
with open(self.token_cache_path, "w") as f:
json.dump(data, f)
logging.info("Saved Cync user tokens to disk.")
except Exception as e:
logging.warning(f"Failed to save Cync user tokens: {e}")
logging.warning("Failed to save Cync user tokens: %s", e)
async def get_lighting_state(self) -> JSONResponse:
"""
@@ -209,11 +242,11 @@ class Lighting(FastAPI):
default_state = {
"power": "off",
"brightness": 50,
"color": {"r": 255, "g": 255, "b": 255}
"color": {"r": 255, "g": 255, "b": 255},
}
return JSONResponse(content=default_state)
except Exception as e:
logging.error(f"Error getting lighting state: {e}")
logging.error("Error getting lighting state: %s", e)
raise HTTPException(status_code=500, detail="Internal server error")
async def set_lighting_state(self, request: Request) -> JSONResponse:
@@ -224,7 +257,9 @@ class Lighting(FastAPI):
state = await request.json()
# Validate state (basic validation)
if not isinstance(state, dict):
raise HTTPException(status_code=400, detail="State must be a JSON object")
raise HTTPException(
status_code=400, detail="State must be a JSON object"
)
# Store in Redis
self.redis_client.set(self.lighting_key, json.dumps(state))
@@ -234,20 +269,30 @@ class Lighting(FastAPI):
# Apply to Cync device
power = state.get("power", "off")
if power not in ["on", "off"]:
raise HTTPException(status_code=400, detail=f"Invalid power state: {power}")
raise HTTPException(
status_code=400, detail=f"Invalid power state: {power}"
)
brightness = state.get("brightness", 50)
if not isinstance(brightness, (int, float)) or not (0 <= brightness <= 100):
raise HTTPException(status_code=400, detail=f"Invalid brightness: {brightness}")
raise HTTPException(
status_code=400, detail=f"Invalid brightness: {brightness}"
)
color = state.get("color")
if color and isinstance(color, dict) and all(k in color for k in ["r", "g", "b"]):
if (
color
and isinstance(color, dict)
and all(k in color for k in ["r", "g", "b"])
):
rgb = (color["r"], color["g"], color["b"])
elif all(k in state for k in ["red", "green", "blue"]):
rgb = (state["red"], state["green"], state["blue"])
for val, name in zip(rgb, ["red", "green", "blue"]):
if not isinstance(val, int) or not (0 <= val <= 255):
raise HTTPException(status_code=400, detail=f"Invalid {name} color value: {val}")
raise HTTPException(
status_code=400, detail=f"Invalid {name} color value: {val}"
)
else:
rgb = None
@@ -256,10 +301,22 @@ class Lighting(FastAPI):
raise HTTPException(status_code=500, detail="Cync API not initialized.")
devices = self.cync_api.get_devices()
if not devices or not isinstance(devices, (list, tuple)):
raise HTTPException(status_code=500, detail="No devices returned from Cync API.")
light = next((d for d in devices if hasattr(d, 'name') and d.name == self.cync_device_name), None)
raise HTTPException(
status_code=500, detail="No devices returned from Cync API."
)
light = next(
(
d
for d in devices
if hasattr(d, "name") and d.name == self.cync_device_name
),
None,
)
if not light:
raise HTTPException(status_code=404, detail=f"Device '{self.cync_device_name}' not found")
raise HTTPException(
status_code=404,
detail=f"Device '{self.cync_device_name}' not found",
)
# Set power
if power == "on":
@@ -275,10 +332,19 @@ class Lighting(FastAPI):
if rgb:
await light.set_rgb(rgb)
logging.info(f"Successfully applied state to device '{self.cync_device_name}': {state}")
return JSONResponse(content={"message": "Lighting state updated and applied", "state": state})
logging.info(
"Successfully applied state to device '%s': %s",
self.cync_device_name,
state,
)
return JSONResponse(
content={
"message": "Lighting state updated and applied",
"state": state,
}
)
except HTTPException:
raise
except Exception as e:
logging.error(f"Error setting lighting state: {e}")
raise HTTPException(status_code=500, detail="Internal server error")
logging.error("Error setting lighting state: %s", e)
raise HTTPException(status_code=500, detail="Internal server error")

2
utils/jwt_utils.py
View File

@@ -16,4 +16,4 @@ def decode_jwt(token: str) -> dict | None:
return None
return jwt.decode(token, key, algorithms=[JWT_ALGORITHM])
except JWTError:
return None
return None