Skip to content
This repository was archived by the owner on Mar 13, 2026. It is now read-only.
Open
Changes from all commits
Commits
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
29 changes: 29 additions & 0 deletions pyeight/user.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,8 @@ def __init__(self, device: "EightSleep", user_id: str, side: str):
self.presence: bool = False
self.observed_low: int = 0

self._device_state: dict[str, Any] = {}

def _get_trend(self, trend_num: int, keys: str | tuple[str, ...]) -> Any:
"""Get trend value for specified key."""
if len(self.trends) < trend_num + 1:
Expand Down Expand Up @@ -147,6 +149,11 @@ def heating_level(self) -> int | None:
self.observed_low = level
return level

@property
def device_state(self) -> dict[str, Any]:
"""Return state of device associated with user"""
return self._device_state

def past_heating_level(self, num) -> int:
"""Return a heating level from the past."""
if num > 9 or len(self.device.device_data_history) < num + 1:
Expand Down Expand Up @@ -622,6 +629,22 @@ async def update_user(self) -> None:
start.strftime(DATE_FORMAT), end.strftime(DATE_FORMAT)
)

async def set_device_state(self, state: bool, level: int = 0) -> None:
"""Update device temperature"""
url = f"{API_URL}/users/{self.user_id}/temperature"

level = max(-100, level)
level = min(100, level)

data = {"currentLevel": level}
if state:
# If device is currently off, turn it on
data["currentState"] = {"type": "smart"}
else:
data["currentState"] = {"type": "off"}

self._device_state = await self.device.api_request("put", url, data=data)

async def set_heating_level(self, level: int, duration: int = 0) -> None:
"""Update heating data json."""
url = f"{API_URL}/devices/{self.device.device_id}"
Expand Down Expand Up @@ -686,3 +709,9 @@ async def update_intervals_data(self) -> None:

intervals = await self.device.api_request("get", url)
self.intervals = intervals.get("intervals", [])

async def update_device_state(self) -> None:
"""Update device state using 'temperature' endpoint"""
url = f"{API_URL}/users/{self.user_id}/temperature"

self._device_state = await self.device.api_request("get", url)