diff --git a/base.py b/base.py index 248a071..5ecb0ee 100644 --- a/base.py +++ b/base.py @@ -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 diff --git a/endpoints/lighting.py b/endpoints/lighting.py index 73d140d..3b60fcf 100644 --- a/endpoints/lighting.py +++ b/endpoints/lighting.py @@ -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") \ No newline at end of file + logging.error("Error setting lighting state: %s", e) + raise HTTPException(status_code=500, detail="Internal server error") diff --git a/utils/jwt_utils.py b/utils/jwt_utils.py index f1dce97..9e9e2bb 100644 --- a/utils/jwt_utils.py +++ b/utils/jwt_utils.py @@ -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 \ No newline at end of file + return None