#5 #4 pantherale0/ha-familysafety#37 pantherale0/ha-familysafety#33

pyfamilysafety/__init__.py

@@ -14,15 +14,20 @@ class FamilySafety:
     def __init__(self, api) -> None:
         self.api: FamilySafetyAPI = api
         self.accounts: list[Account] = None
+        self.experimental: bool = False
         self.pending_requests = []
 
     @classmethod
-    async def create(cls, token, use_refresh_token: bool=False) -> 'FamilySafety':
+    async def create(cls, token,
+                     use_refresh_token: bool=False,
+                     experimental: bool=False) -> 'FamilySafety':
         """Create an instance of the family safety module."""
         self = cls(await FamilySafetyAPI.create(token, use_refresh_token))
         accounts = await self.api.send_request("get_accounts")
-        self.accounts = await Account.from_dict(self.api, accounts.get("json"))
-        await self._get_pending_requests()
+        self.accounts = await Account.from_dict(self.api, accounts.get("json"), experimental)
+        self.experimental = experimental
+        if experimental:
+            await self._get_pending_requests()
         return self
 
     def get_account(self, user_id) -> Account:
@@ -99,7 +104,8 @@ async def deny_pending_request(self, request_id) -> bool:
     async def update(self):
         """Updates submodules"""
         try:
-            await self._get_pending_requests()
+            if self.experimental:
+                await self._get_pending_requests()
             for account in self.accounts:
                 await account.update()
         except AggregatorException:

pyfamilysafety/account.py

@@ -2,14 +2,15 @@
 """Family safety account handler."""
 
 import logging
-from datetime import datetime, date, time
+from datetime import datetime, date, time, timedelta
 from urllib.parse import quote_plus
 
 from .api import FamilySafetyAPI
 from .device import Device
 from .application import Application
 from .enum import OverrideTarget, OverrideType
 from .helpers import localise_datetime, API_TIMEZONE
+from .requests import PendingRequest
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -30,6 +31,7 @@ def __init__(self, api) -> None:
         self.screentime_usage: dict = None
         self.application_usage: dict = None
         self.blocked_platforms: list[OverrideTarget] = None
+        self.experimental: bool = False
         self._api: FamilySafetyAPI = api
         self.account_balance: float = 0.0
         self.account_currency: str = ""
@@ -81,9 +83,6 @@ async def _get_account_balance(self):
         if len(balances) == 1:
             self.account_balance = balances[0]["balance"]
             self.account_currency = balances[0]["currency"]
-            return
-        _LOGGER.warning("Unknown account spending response, please report to developer. %s", response)
-        return
 
     async def get_screentime_usage(self,
                                    start_time: datetime = None,
@@ -153,8 +152,8 @@ async def override_device(self,
         response = await self._api.send_request(
             endpoint="override_device_restriction",
             body={
-                "overrideType": str(override),
-                "target": str(target),
+                "overrideType": str(override).upper(),
+                "target": str(target).upper(),
                 "validUntil": valid_until.strftime("%Y-%m-%dT%H:%M:%SZ")
             },
             USER_ID=self.user_id
@@ -179,7 +178,7 @@ def _update_device_blocked(self, raw_response: dict):
         self.blocked_platforms = blocked_platforms
 
     @classmethod
-    async def from_dict(cls, api: FamilySafetyAPI, raw_response: dict) -> list['Account']:
+    async def from_dict(cls, api: FamilySafetyAPI, raw_response: dict, experimental: bool) -> list['Account']:
         """Converts a roster request response to an array."""
         response = []
         if "members" in raw_response.keys():
@@ -192,6 +191,7 @@ async def from_dict(cls, api: FamilySafetyAPI, raw_response: dict) -> list['Acco
                     self.profile_picture = member.get("profilePicUrl")
                     self.first_name = member.get("user").get("firstName")
                     self.surname = member.get("user").get("lastName")
+                    self.experimental = experimental
                     await self.update()
                     response.append(self)
 

pyfamilysafety/api.py

@@ -8,7 +8,7 @@
 import aiohttp.client_exceptions
 
 from .authenticator import Authenticator
-from .const import ENDPOINTS, BASE_URL, AGGREGATOR_ERROR
+from .const import ENDPOINTS, BASE_URL, AGGREGATOR_ERROR, USER_AGENT
 from .exceptions import HttpException, AggregatorException, Unauthorized, RequestDenied
 
 _LOGGER = logging.getLogger(__name__)
@@ -65,8 +65,19 @@ async def send_request(self, endpoint: str, body: object=None, headers: dict=Non
         if self._session.headers.get("Authorization", None) is None:
             # Add the auth token
             self._session.headers.add("Authorization", self._auth_token)
+        # add headers to override
+        if headers is None:
+            headers = {}
+            headers["Authorization"] = self._auth_token
+            headers["User-Agent"] = USER_AGENT
+            headers["Content-Type"] = "application/json"
+
         # format the URL using the kwargs
-        url = e_point.get("url").format(BASE_URL=BASE_URL, **kwargs)
+        url = e_point.get("url")
+        if "{BASE_URL" in url:
+            url = url.format(BASE_URL=BASE_URL, **kwargs)
+        else:
+            url = url.format(**kwargs)
         _LOGGER.debug("Built URL %s", url)
         # now send the HTTP request
         resp: dict = {

pyfamilysafety/const.py

@@ -32,6 +32,10 @@
         "url": "{BASE_URL}/v4/activityreport/deviceScreenTimeUsage/{USER_ID}?beginTime={BEGIN_TIME}&endTime={END_TIME}&topDeviceCount={DEVICE_COUNT}",
         "method": "GET"
     },
+    "update_schedule": {
+        "url": "{BASE_URL}/v4/devicelimits/schedules/{USER_ID}",
+        "method": "PATCH"
+    },
     "get_user_devices": {
         "url": "{BASE_URL}/v1/devices/{USER_ID}",
         "method": "GET"
@@ -87,17 +91,5 @@
     "get_additional_permission_token": {
         "url": "{BASE_URL}/v1/FamilyPermission/permissiontoken/{USER_ID}?scopes={SCOPES}",
         "method": "GET"
-    },
-    "get_named_locations": {
-        "url": "{BASE_URL}/v1/namedlocation",
-        "method": "GET"
-    },
-    "add_named_location": {
-        "url": "{BASE_URL}/v1/namedlocation",
-        "method": "POST"
-    },
-    "delete_named_location": {
-        "url": "{BASE_URL}/v1/namedlocation/{LOCATION_ID}",
-        "method": "DELETE"
     }
 }

pyfamilysafety/enum.py

@@ -7,11 +7,8 @@ class OverrideTarget(Enum):
     WINDOWS = 0
     XBOX = 1
     MOBILE = 2
-    ALL_DEVICES = 4
 
     def __str__(self) -> str:
-        if self.name == "ALL_DEVICES":
-            return "AllDevices"
         if self.name == "WINDOWS":
             return "Windows"
         if self.name == "MOBILE":
@@ -23,8 +20,6 @@ def __str__(self) -> str:
     @classmethod
     def from_pretty(cls, pretty) -> 'OverrideTarget':
         """Returns from pretty."""
-        if pretty == "AllDevices":
-            return cls.ALL_DEVICES
         if pretty == "Windows":
             return cls.WINDOWS
         if pretty == "Mobile":

test.py

@@ -3,7 +3,9 @@
 
 import logging
 import asyncio
+from datetime import datetime, timedelta
 from pyfamilysafety import FamilySafety
+from pyfamilysafety.enum import OverrideType
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -12,7 +14,7 @@ async def main():
     login = True
     while login:
         try:
-            auth = await FamilySafety.create(token=input("Response URL: "), use_refresh_token=False)
+            auth = await FamilySafety.create(token=input("Response URL: "), use_refresh_token=False, experimental=True)
             _LOGGER.info("Logged in, ready.")
             _LOGGER.debug("Access token is: %s", auth.api.authenticator.refresh_token)
             login = False
@@ -24,6 +26,7 @@ async def main():
             _LOGGER.debug("Discovered account %s, label %s", account.user_id, account.first_name)
             _LOGGER.debug(account)
             _LOGGER.debug("Usage today %s", account.today_screentime_usage)
+            await account.override_device("AllDevices", OverrideType.UNTIL, valid_until=datetime.now()+timedelta(days=5))
 
         _LOGGER.debug("ping")
         await asyncio.sleep(15)