working services to sitch wifi and wifree

custom_components/telenet_telemeter/__init__.py

@@ -111,42 +111,70 @@ async def handle_switch_wireless(enableWifi, enableWifree):
         if not v2:
             return
 
-        customerDetails = await hass.async_add_executor_job(lambda: session.customerdetails())
-        customerLocationId = customerDetails.get('customerLocations')[0].get('id')
+        # customerDetails = await hass.async_add_executor_job(lambda: session.customerdetails())
+        # customerLocationId = customerDetails.get('customerLocations')[0].get('id')
 
         internetProductDetails = await hass.async_add_executor_job(lambda: session.productSubscriptions("INTERNET"))
         internetProductIdentifier = internetProductDetails[0].get('identifier')
 
         modemDetails = await hass.async_add_executor_job(lambda: session.modemdetails(internetProductIdentifier))
         modemMac = modemDetails.get('mac')
 
+        productServiceDetails = await hass.async_add_executor_job(lambda: session.productService(internetProductIdentifier, "INTERNET"))
+        customerLocationId = productServiceDetails.get('locationId')
+
+        for lineLevelProduct in productServiceDetails.get('lineLevelProducts',[]):
+            if lineLevelProduct.get('specurl'):
+                 urlDetails = await hass.async_add_executor_job(lambda: session.urldetails(lineLevelProduct.get('specurl')))
+
+        for option in productServiceDetails.get('options',[]):
+            if option.get('specurl'):
+                 urlDetails = await hass.async_add_executor_job(lambda: session.urldetails(option.get('specurl')))
+
         wifiDetails = await hass.async_add_executor_job(lambda: session.wifidetails(internetProductIdentifier, modemMac))
-        wifiEnabled = wifiDetails.get('wirelessEnabled')
+        # wifiEnabled = wifiDetails.get('wirelessEnabled')
+        wifiEnabled = wifiDetails.get("wirelessInterfaces")[0].get('active')
         wifreeEnabled = wifiDetails.get('homeSpotEnabled')
 
         if enableWifi is None:
             enableWifi = wifiEnabled
+        if enableWifi == "Yes":
+            enableWifi = True
+        else:
+            enableWifi = False
         if enableWifree is None:
             enableWifree = wifreeEnabled
+
+        # if wifiEnabled and enableWifi and wifreeEnabled == enableWifree:
+        #     _LOGGER.debug(f"no wifi change required: wifiEnabled: {wifiEnabled}, enableWifi: {enableWifi}, wifreeEnabled: {wifreeEnabled}, enableWifree: {enableWifree}")
+        #     return
+        # else:
+        _LOGGER.debug(f"wifi change required: wifiEnabled: {wifiEnabled}, enableWifi: {enableWifi}, wifreeEnabled: {wifreeEnabled}, enableWifree: {enableWifree}")
+
 
-        response  = await hass.async_add_executor_job(lambda: session.switchWifi(enableWifree, enableWifi, internetProductIdentifier, modemMac, customerLocationId))
+        await hass.async_add_executor_job(lambda: session.switchWifi(enableWifree, enableWifi, internetProductIdentifier, modemMac, customerLocationId))
 
-        _LOGGER.debug(f"{NAME} handle_switch_wifi switch executed, old state: wifiEnabled: {wifiEnabled}, wifreeEnabled: {wifreeEnabled}, new state: enableWifi: {enableWifi}, enableWifree: {enableWifree}, response: {response}")
+        _LOGGER.debug(f"{NAME} handle_switch_wifi switch executed, old state: wifiEnabled: {wifiEnabled}, wifreeEnabled: {wifreeEnabled}, new state: enableWifi: {enableWifi}, enableWifree: {enableWifree}")
+        return
 
     async def handle_switch_wifi(call):
         """Handle the service call."""
 
         enable = call.data.get('enable')
         if not(enable in ["Yes","No"]):
             return
+        _LOGGER.debug(f"handle_switch_wifi: enable : {enable}")
         await handle_switch_wireless(enable, None)
+        return
 
     async def handle_switch_wifree(call):
         """Handle the service call."""
         enable = call.data.get('enable')
         if not(enable in ["Yes","No"]):
             return
+        _LOGGER.debug(f"handle_switch_wifree: enable : {enable}")
         await handle_switch_wireless(None, enable)
+        return
 
 
     hass.services.async_register(DOMAIN, 'switch_wifi', handle_switch_wifi)

custom_components/telenet_telemeter/switch.py

@@ -0,0 +1,42 @@
+"""Support for Wifi switches"""
+import logging
+
+from homeassistant.helpers.entity import ToggleEntity
+from homeassistant.const import CONF_USERNAME
+
+from .const import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
+    """Old way."""
+
+
+async def async_setup_entry(hass, config_entry, async_add_entities):
+
+    switches = []
+    account = config_entry.data.get(CONF_USERNAME)
+    audiData = hass.data[DOMAIN][account]
+
+    switch = ComponentSwitch()
+    switches.append(switch)
+
+    async_add_entities(switches)
+
+
+class ComponentSwitch():
+    """Representation of a Audi switch."""
+
+    @property
+    def is_on(self):
+        """Return true if switch is on."""
+        return self._instrument.state
+
+    async def async_turn_on(self, **kwargs):
+        """Turn the switch on."""
+        await self._instrument.turn_on()
+
+    async def async_turn_off(self, **kwargs):
+        """Turn the switch off."""
+        await self._instrument.turn_off()

custom_components/telenet_telemeter/utils.py

@@ -6,6 +6,7 @@
 from typing import List
 import requests
 from pydantic import BaseModel
+from enum import Enum
 
 import voluptuous as vol
 from homeassistant.helpers.aiohttp_client import async_get_clientsession
@@ -39,6 +40,15 @@ def check_settings(config, hass):
 
     raise vol.Invalid("Missing settings to setup the sensor.")
 
+class HttpMethod(Enum):
+    GET = 'GET'
+    POST = 'POST'
+    PUT = 'PUT'
+    PATCH = 'PATCH'
+    DELETE = 'DELETE'
+    HEAD = 'HEAD'
+    OPTIONS = 'OPTIONS'
+
 
 class TelenetSession(object):
     def __init__(self):
@@ -47,27 +57,32 @@ def __init__(self):
         # self.s.headers["x-alt-referer"] = "https://www2.telenet.be/nl/klantenservice/#/pages=1/menu=selfservice"
         self.s.headers["x-alt-referer"] = "https://www2.telenet.be/residential/nl/mijn-telenet"
 
-    def callTelenet(self, url, caller = "Not set", data = None, expectedStatusCode = "200", printResponse = False, patch = False):
-        if patch:
-            _LOGGER.debug(f"[{caller}] Calling PATCH {url}")
-            response = self.s.patch(url,data,timeout=30)
-        else:
-            if data == None:
-                _LOGGER.debug(f"[{caller}] Calling GET {url}")
-                response = self.s.get(url,timeout=30)
-            else:
-                _LOGGER.debug(f"[{caller}] Calling POST {url}")
-                response = self.s.post(url,data,timeout=30)
+    def callTelenet(self, url, caller = "Not set", expectedStatusCode = 200, data = None, printResponse = False, method : HttpMethod  = HttpMethod.GET):
+        if method == HttpMethod.GET:
+            _LOGGER.debug(f"[{caller}] Calling GET {url}")
+            response = self.s.get(url,timeout=60)
+        elif method == HttpMethod.POST:
+            # self.s.headers["Content-Type"] = "application/json;charset=UTF-8"
+            _LOGGER.debug(f"[{caller}] Calling POST {url} with data {data}")
+            response = self.s.post(url,data,timeout=30)
+        elif method == HttpMethod.PATCH:
+            self.s.headers["Content-Type"] = "application/json;charset=UTF-8"
+            _LOGGER.debug(f"[{caller}] Calling PATCH {url} with data: {data}")
+            response = self.s.patch(url,json=data,timeout=60)
+        elif method == HttpMethod.OPTIONS:
+            self.s.headers["Content-Type"] = "application/json;charset=UTF-8"
+            _LOGGER.debug(f"[{caller}] Calling OPTIONS {url} with data: {data}")
+            response = self.s.options(url,timeout=60)
         _LOGGER.debug(f"[{caller}] http status code = {response.status_code} (expecting {expectedStatusCode})")
         if printResponse:
-            _LOGGER.debug(f"[{caller}] Response:\n{response.text}")
+            _LOGGER.debug(f"[{caller}] Response: {response.text}")
         if expectedStatusCode != None:
             assert response.status_code == expectedStatusCode
 
         return response
 
     def login(self, username, password):
-        response = self.callTelenet("https://api.prd.telenet.be/ocapi/oauth/userdetails","login", None, None)
+        response = self.callTelenet("https://api.prd.telenet.be/ocapi/oauth/userdetails","login", None)
         if response.status_code == 200:
             # Return if already authenticated
             return
@@ -76,86 +91,100 @@ def login(self, username, password):
         _LOGGER.debug(f"loging response to split state, nonce: {response.text}")
         state, nonce = response.text.split(",", maxsplit=2)
         # Log in
-        self.callTelenet(f'https://login.prd.telenet.be/openid/oauth/authorize?client_id=ocapi&response_type=code&claims={{"id_token":{{"http://telenet.be/claims/roles":null,"http://telenet.be/claims/licenses":null}}}}&lang=nl&state={state}&nonce={nonce}&prompt=login',"login", None, None)
-        self.callTelenet("https://login.prd.telenet.be/openid/login.do","login",{"j_username": username,"j_password": password,"rememberme": True}, 200)
+        self.callTelenet(f'https://login.prd.telenet.be/openid/oauth/authorize?client_id=ocapi&response_type=code&claims={{"id_token":{{"http://telenet.be/claims/roles":null,"http://telenet.be/claims/licenses":null}}}}&lang=nl&state={state}&nonce={nonce}&prompt=login',"login", None)
+        self.callTelenet("https://login.prd.telenet.be/openid/login.do","login", 200, {"j_username": username,"j_password": password,"rememberme": True}, False, HttpMethod.POST)
         self.s.headers["X-TOKEN-XSRF"] = self.s.cookies.get("TOKEN-XSRF")
-        self.callTelenet("https://api.prd.telenet.be/ocapi/oauth/userdetails","login", None, 200)
+        self.callTelenet("https://api.prd.telenet.be/ocapi/oauth/userdetails","login")
 
     def userdetails(self):
-        response = self.callTelenet("https://api.prd.telenet.be/ocapi/oauth/userdetails","userdetails", None, 200)
+        response = self.callTelenet("https://api.prd.telenet.be/ocapi/oauth/userdetails","userdetails", None)
         return response.json()
 
     def customerdetails(self):
-        response = self.callTelenet("https://api.prd.telenet.be/ocapi/public/api/customer-service/v1/customers","customerdetails", None, 200)
+        response = self.callTelenet("https://api.prd.telenet.be/ocapi/public/api/customer-service/v1/customers","customerdetails")
         return response.json()
 
     def telemeter(self):
-        response = self.callTelenet("https://api.prd.telenet.be/ocapi/public/?p=internetusage,internetusagereminder","telemeter", None, 200)
+        response = self.callTelenet("https://api.prd.telenet.be/ocapi/public/?p=internetusage,internetusagereminder","telemeter")
         return response.json()
 
     def telemeter_product_details(self, url):
-        response = self.callTelenet(url,"telemeter_product_details",None, 200)
+        response = self.callTelenet(url,"telemeter_product_details")
         return response.json()
 
     def modemdetails(self, productIdentifier):
-        response = self.callTelenet(f"https://api.prd.telenet.be/ocapi/public/api/resource-service/v1/modems?productIdentifier={productIdentifier}","modemdetails", None, 200)
+        response = self.callTelenet(f"https://api.prd.telenet.be/ocapi/public/api/resource-service/v1/modems?productIdentifier={productIdentifier}","modemdetails")
         return response.json()
 
     def wifidetails(self, productIdentifier, modemMac):
-        response = self.callTelenet(f"https://api.prd.telenet.be/ocapi/public/api/resource-service/v1/modems/{modemMac}/wireless-settings?withmetadata=true&withwirelessservice=true&productidentifier={productIdentifier}","wifidetails", None, 200)
+        response = self.callTelenet(f"https://api.prd.telenet.be/ocapi/public/api/resource-service/v1/modems/{modemMac}/wireless-settings?withmetadata=true&withwirelessservice=true&productidentifier={productIdentifier}","wifidetails")
+        return response.json()
+
+    def urldetails(self, url):
+        response = self.callTelenet(url,"urldetails")
         return response.json()
 
     def switchWifi(self, homeSpotEnabled, wirelessEnabled, productIdentifier, modemMac, locationId):
-        if homeSpotEnabled:
+        if homeSpotEnabled == "Yes":
             homeSpotEnabled = "Yes"
         else:
             homeSpotEnabled = "No"
 
-        if wirelessEnabled:
+        if wirelessEnabled == "Yes":
             wirelessEnabled = "Yes"
         else:
             wirelessEnabled = "No"
-        data = {"productIdentifier":productIdentifier,"homeSpotEnabled":homeSpotEnabled,"wirelessEnabled":wirelessEnabled,"locationId":locationId,"patchOperations":[{"op":"replace","path":"/wirelessInterfaces/2.4GHZ/ssids/PRIMARY/active","value":True},{"op":"replace","path":"/wirelessInterfaces/5GHZ/ssids/PRIMARY/active","value":True}]}
-        response = self.callTelenet(f"https://api.prd.telenet.be/ocapi/public/api/resource-service/v1/modems/{modemMac}/wireless-settings?withmetadata=true&withwirelessservice=true&productidentifier={productIdentifier}","wifidetails", data, 200, True)
-        return response.json()
+
+        if wirelessEnabled == "Yes":
+            wirelessEnabled = True
+        else:
+            wirelessEnabled = False
+        data = {"productIdentifier":productIdentifier,"homeSpotEnabled":homeSpotEnabled,"wirelessEnabled":"Yes","locationId":locationId,"patchOperations":[{"op":"replace","path":"/wirelessInterfaces/2.4GHZ/ssids/PRIMARY/active","value":wirelessEnabled},{"op":"replace","path":"/wirelessInterfaces/5GHZ/ssids/PRIMARY/active","value":wirelessEnabled}]}
+        self.callTelenet(f"https://api.prd.telenet.be/ocapi/public/api/resource-service/v1/modems/{modemMac}/wireless-settings","optionswifi", 200, None, True, HttpMethod.OPTIONS)
+        self.callTelenet(f"https://api.prd.telenet.be/ocapi/public/api/resource-service/v1/modems/{modemMac}/wireless-settings","patchwifi", 200, data, True, HttpMethod.PATCH)
+        return
 
     def mobile(self):
-        response = self.callTelenet("https://api.prd.telenet.be/ocapi/public/?p=mobileusage","mobile", None, 200)
+        response = self.callTelenet("https://api.prd.telenet.be/ocapi/public/?p=mobileusage","mobile")
         return response.json()
 
     def planInfo(self):
-        response = self.callTelenet("https://api.prd.telenet.be/ocapi/public/api/product-service/v1/product-subscriptions?producttypes=PLAN","planInfo", None, 200)
+        response = self.callTelenet("https://api.prd.telenet.be/ocapi/public/api/product-service/v1/product-subscriptions?producttypes=PLAN","planInfo")
         return response.json()
 
     def billCycles(self, productType, productIdentifier):
-        response = self.callTelenet(f"https://api.prd.telenet.be/ocapi/public/api/billing-service/v1/account/products/{productIdentifier}/billcycle-details?producttype={productType}&count=3","billCycles", None, 200)
+        response = self.callTelenet(f"https://api.prd.telenet.be/ocapi/public/api/billing-service/v1/account/products/{productIdentifier}/billcycle-details?producttype={productType}&count=3","billCycles")
         return response.json()
 
     def productUsage(self, productType, productIdentifier,startDate, endDate):
-        response = self.callTelenet(f"https://api.prd.telenet.be/ocapi/public/api/product-service/v1/products/{productType}/{productIdentifier}/usage?fromDate={startDate}&toDate={endDate}","productUsage", None, 200)
+        response = self.callTelenet(f"https://api.prd.telenet.be/ocapi/public/api/product-service/v1/products/{productType}/{productIdentifier}/usage?fromDate={startDate}&toDate={endDate}","productUsage")
         return response.json()
 
     def productDailyUsage(self, productType, productIdentifier,startDate, endDate):
-        response = self.callTelenet(f"https://api.prd.telenet.be/ocapi/public/api/product-service/v1/products/{productType}/{productIdentifier}/dailyusage?billcycle=CURRENT&fromDate={startDate}&toDate={endDate}","productUsage", None, 200)
+        response = self.callTelenet(f"https://api.prd.telenet.be/ocapi/public/api/product-service/v1/products/{productType}/{productIdentifier}/dailyusage?billcycle=CURRENT&fromDate={startDate}&toDate={endDate}","productUsage")
         return response.json()
 
     def productSubscriptions(self, productType):
-        response = self.callTelenet(f"https://api.prd.telenet.be/ocapi/public/api/product-service/v1/product-subscriptions?producttypes={productType}","productSubscriptions", None, 200)
+        response = self.callTelenet(f"https://api.prd.telenet.be/ocapi/public/api/product-service/v1/product-subscriptions?producttypes={productType}","productSubscriptions")
+        return response.json()
+
+    def productService(self, productIdentifier, productType):
+        response = self.callTelenet(f"https://api.prd.telenet.be/ocapi/public/api/product-service/v1/products/{productIdentifier}?producttype={productType.lower()}","productService")
         return response.json()
 
     def mobileUsage(self, productIdentifier):
-        response = self.callTelenet(f"https://api.prd.telenet.be/ocapi/public/api/mobile-service/v3/mobilesubscriptions/{productIdentifier}/usages","mobileUsage", None, 200)
+        response = self.callTelenet(f"https://api.prd.telenet.be/ocapi/public/api/mobile-service/v3/mobilesubscriptions/{productIdentifier}/usages","mobileUsage")
         return response.json()
 
     def mobileBundleUsage(self, bundleIdentifier, lineIdentifier = None):
         if lineIdentifier != None:
-            response = self.callTelenet(f"https://api.prd.telenet.be/ocapi/public/api/mobile-service/v3/mobilesubscriptions/{bundleIdentifier}/usages?type=bundle&lineIdentifier={lineIdentifier}","mobileBundleUsage lineIdentifier", None, 200)
+            response = self.callTelenet(f"https://api.prd.telenet.be/ocapi/public/api/mobile-service/v3/mobilesubscriptions/{bundleIdentifier}/usages?type=bundle&lineIdentifier={lineIdentifier}","mobileBundleUsage lineIdentifier")
         else:
-            response = self.callTelenet(f"https://api.prd.telenet.be/ocapi/public/api/mobile-service/v3/mobilesubscriptions/{bundleIdentifier}/usages?type=bundle","mobileBundleUsage bundle", None, 200)
+            response = self.callTelenet(f"https://api.prd.telenet.be/ocapi/public/api/mobile-service/v3/mobilesubscriptions/{bundleIdentifier}/usages?type=bundle","mobileBundleUsage bundle")
         return response.json()
 
     def apiVersion2(self):
-        response = self.callTelenet("https://api.prd.telenet.be/ocapi/public/api/product-service/v1/product-subscriptions?producttypes=PLAN","apiVersion2", None, None)
+        response = self.callTelenet("https://api.prd.telenet.be/ocapi/public/api/product-service/v1/product-subscriptions?producttypes=PLAN","apiVersion2", None)
         if response.status_code == 200:
             return True
         return False