diff --git a/googledevices/api/cast/alarm.py b/googledevices/api/cast/alarm.py deleted file mode 100644 index 8546a0e..0000000 --- a/googledevices/api/cast/alarm.py +++ /dev/null @@ -1,64 +0,0 @@ -"""Retrieve alarms from Google Home.""" -from googledevices.utils.const import HEADERS, CASTPORT -import googledevices.utils.log as log -from googledevices.helpers import gdh_request - - -class Alarm(object): - """A class for getting the alarms from a Google Home.""" - - def __init__(self, host, loop, session): - """Initialize the class.""" - self.loop = loop - self.host = host - self.session = session - self._alarms = [] - self._alarmvolume = None - - async def get_alarms(self): - """Get the alarms from the device.""" - endpoint = 'setup/assistant/alarms' - response = await gdh_request(host=self.host, port=CASTPORT, - loop=self.loop, session=self.session, - endpoint=endpoint) - self._alarms = response - log.debug(self._alarms) - return self._alarms - - async def get_alarm_volume(self): - """Get the alarm volume for the device.""" - endpoint = 'setup/assistant/alarms/volume' - response = await gdh_request(host=self.host, port=CASTPORT, - loop=self.loop, session=self.session, - endpoint=endpoint, method='post') - self._alarmvolume = response - log.debug(self._alarmvolume) - return self._alarmvolume - - async def set_alarm_volume(self, volume): - """Set the alarm volume for the device.""" - data = {'volume': volume} - endpoint = 'setup/assistant/alarms/volume' - returnvalue = False - result = await gdh_request(host=self.host, port=CASTPORT, - endpoint=endpoint, method='post', - loop=self.loop, session=self.session, - json_data=data, headers=HEADERS, - json=False) - try: - if result.status == 200: - returnvalue = True - except AttributeError: - msg = "Error connecting to - {}".format(self.host) - log.error(msg) - return returnvalue - - @property - def alarms(self): - """Return the alarms.""" - return self._alarms - - @property - def alarm_volume(self): - """Return the alarm volume.""" - return self._alarmvolume diff --git a/googledevices/api/cast/assistant.py b/googledevices/api/cast/assistant.py new file mode 100644 index 0000000..aa7032d --- /dev/null +++ b/googledevices/api/cast/assistant.py @@ -0,0 +1,124 @@ +"""Controll Assistant settings on the unit.""" +from googledevices.utils.const import CASTPORT, HEADERS +from googledevices.helpers import gdh_request +import googledevices.utils.log as log + + +class Assistant(object): + """A class for Assistant settings.""" + + def __init__(self, host, loop, session): + """Initialize the class.""" + self.host = host + self.loop = loop + self.session = session + self._alarms = [] + self._alarmvolume = None + + async def set_night_mode_params(self, data): + """Set night mode options.""" + endpoint = 'setup/assistant/set_night_mode_params' + result = await gdh_request(host=self.host, port=CASTPORT, + endpoint=endpoint, method='post', + loop=self.loop, session=self.session, + json_data=data, headers=HEADERS) + return result + + async def notifications_enabled(self, mode=True): + """Set notifications_enabled True/False.""" + endpoint = 'setup/assistant/notifications' + data = {"notifications_enabled": mode} + result = await gdh_request(host=self.host, port=CASTPORT, + endpoint=endpoint, method='post', + loop=self.loop, session=self.session, + json_data=data, headers=HEADERS) + return result + + async def set_accessibility(self, start=True, end=False): + """Set accessibility True/False.""" + endpoint = 'setup/assistant/a11y_mode' + data = {"endpoint_enabled": end, "hotword_enabled": start} + result = await gdh_request(host=self.host, port=CASTPORT, + endpoint=endpoint, method='post', + loop=self.loop, session=self.session, + json_data=data, headers=HEADERS) + return result + + async def delete_alarms(self, data): + """Delete active alarms and timers.""" + endpoint = 'setup/assistant/alarms/delete' + result = await gdh_request(host=self.host, port=CASTPORT, + endpoint=endpoint, method='post', + loop=self.loop, session=self.session, + json_data=data, headers=HEADERS) + return result + + async def set_equalizer(self, low_gain=0, high_gain=0): + """Set equalizer db gain.""" + endpoint = 'setup/user_eq/set_equalizer' + returnvalue = False + data = { + "low_shelf": {"gain_db": low_gain}, + "high_shelf": {"gain_db": high_gain} + } + result = await gdh_request(host=self.host, port=CASTPORT, + endpoint=endpoint, method='post', + loop=self.loop, session=self.session, + json_data=data, headers=HEADERS, + json=False) + try: + if result.status == 200: + returnvalue = True + except AttributeError: + msg = "Error connecting to - {}".format(self.host) + log.error(msg) + return returnvalue + + async def get_alarms(self): + """Get the alarms from the device.""" + endpoint = 'setup/assistant/alarms' + response = await gdh_request(host=self.host, port=CASTPORT, + loop=self.loop, session=self.session, + endpoint=endpoint, headers=HEADERS) + self._alarms = response + log.debug(self._alarms) + return self._alarms + + async def get_alarm_volume(self): + """Get the alarm volume for the device.""" + endpoint = 'setup/assistant/alarms/volume' + response = await gdh_request(host=self.host, port=CASTPORT, + loop=self.loop, session=self.session, + endpoint=endpoint, headers=HEADERS, + method='post') + self._alarmvolume = response + log.debug(self._alarmvolume) + return self._alarmvolume + + async def set_alarm_volume(self, volume): + """Set the alarm volume for the device.""" + data = {'volume': volume} + endpoint = 'setup/assistant/alarms/volume' + returnvalue = False + result = await gdh_request(host=self.host, port=CASTPORT, + endpoint=endpoint, method='post', + loop=self.loop, session=self.session, + json_data=data, headers=HEADERS, + json=False) + try: + if result.status == 200: + returnvalue = True + except AttributeError: + msg = "Error connecting to - {}".format(self.host) + log.error(msg) + return returnvalue + + @property + def alarms(self): + """Return the alarms.""" + return self._alarms + + @property + def alarm_volume(self): + """Return the alarm volume.""" + return self._alarmvolume diff --git a/googledevices/api/cast/bluetooth.py b/googledevices/api/cast/bluetooth.py index 2b62f39..3adb137 100644 --- a/googledevices/api/cast/bluetooth.py +++ b/googledevices/api/cast/bluetooth.py @@ -13,6 +13,7 @@ def __init__(self, host, loop, session): self.host = host self.session = session self._devices = [] + self._paired_devices = [] self._status = {} async def get_bluetooth_status(self): @@ -20,11 +21,39 @@ async def get_bluetooth_status(self): endpoint = 'setup/bluetooth/status' response = await gdh_request(host=self.host, port=CASTPORT, loop=self.loop, session=self.session, - endpoint=endpoint) + endpoint=endpoint, headers=HEADERS) self._status = response log.debug(self._status) return self._status + async def get_paired_devices(self): + """Get paired devices.""" + endpoint = 'setup/bluetooth/get_bonded' + response = await gdh_request(host=self.host, port=CASTPORT, + loop=self.loop, session=self.session, + endpoint=endpoint, headers=HEADERS) + self._status = response + log.debug(self._status) + return self._status + + async def forget_paired_device(self, mac_address): + """Forget a paired device.""" + endpoint = 'setup/bluetooth/bond' + data = {"bond": False, "mac_address": mac_address} + returnvalue = False + result = await gdh_request(host=self.host, port=CASTPORT, + endpoint=endpoint, method='post', + loop=self.loop, session=self.session, + json_data=data, headers=HEADERS, + json=False) + try: + if result.status == 200: + returnvalue = True + except AttributeError: + msg = "Error connecting to - {}".format(self.host) + log.error(msg) + return returnvalue + async def set_discovery_enabled(self): """Enable bluetooth discoverablility.""" endpoint = 'setup/bluetooth/discovery' @@ -43,6 +72,24 @@ async def set_discovery_enabled(self): log.error(msg) return returnvalue + async def pair_with_mac(self, mac_address): + """Pair with bluetooth device.""" + endpoint = 'setup/bluetooth/scan' + data = {"connect": True, "mac_address": mac_address, "profile": 2} + returnvalue = False + result = await gdh_request(host=self.host, port=CASTPORT, + endpoint=endpoint, method='post', + loop=self.loop, session=self.session, + json_data=data, headers=HEADERS, + json=False) + try: + if result.status == 200: + returnvalue = True + except AttributeError: + msg = "Error connecting to - {}".format(self.host) + log.error(msg) + return returnvalue + async def scan_for_devices(self): """Scan for bluetooth devices.""" endpoint = 'setup/bluetooth/scan' @@ -66,7 +113,7 @@ async def get_scan_result(self): endpoint = 'setup/bluetooth/scan_results' response = await gdh_request(host=self.host, port=CASTPORT, loop=self.loop, session=self.session, - endpoint=endpoint) + endpoint=endpoint, headers=HEADERS) self._devices = response log.debug(self._devices) return self._devices @@ -78,5 +125,10 @@ def status(self): @property def devices(self): - """Return the device info if any.""" + """Return the devices if any.""" return self._devices + + @property + def paired_devices(self): + """Return paired devices if any.""" + return self._paired_devices diff --git a/googledevices/api/cast/info.py b/googledevices/api/cast/info.py index 6dbc053..ae8fa5a 100644 --- a/googledevices/api/cast/info.py +++ b/googledevices/api/cast/info.py @@ -1,10 +1,10 @@ """Get device information for the unit.""" -from googledevices.utils.const import DEFAULT_DEVICE_NAME, CASTPORT +from googledevices.utils.const import DEFAULT_DEVICE_NAME, CASTPORT, HEADERS import googledevices.utils.log as log from googledevices.helpers import gdh_request -class Info(object): +class Info(object): # pylint: disable=R0902 """A class for device info.""" def __init__(self, host, loop, session): @@ -14,20 +14,96 @@ def __init__(self, host, loop, session): self.session = session self._name = 'GoogleDevice' self._device_info = {} + self._offer = {} + self._timezones = [] + self._locales = [] + self._app_device_id = {} async def get_device_info(self): - """Get device information for the unit..""" + """Get device information for the unit.""" endpoint = 'setup/eureka_info' params = ("params=version,audio,name,build_info,detail,device_info," "net,wifi,setup,settings,opt_in,opencast,multizone,proxy," "night_mode_params,user_eq,room_equalizer&options=detail") response = await gdh_request(host=self.host, port=CASTPORT, loop=self.loop, session=self.session, - endpoint=endpoint, params=params) + endpoint=endpoint, params=params, + headers=HEADERS) self._device_info = response log.debug(self._device_info) return self._device_info + async def get_offer(self): + """Get offer token.""" + endpoint = 'setup/offer' + response = await gdh_request(host=self.host, port=CASTPORT, + loop=self.loop, session=self.session, + endpoint=endpoint, headers=HEADERS) + self._offer = response + log.debug(self._offer) + return self._offer + + async def get_timezones(self): + """Get supported timezones.""" + endpoint = 'setup/supported_timezones' + response = await gdh_request(host=self.host, port=CASTPORT, + loop=self.loop, session=self.session, + endpoint=endpoint, headers=HEADERS) + self._timezones = response + log.debug(self._timezones) + return self._timezones + + async def get_locales(self): + """Get supported locales.""" + endpoint = 'setup/supported_locales' + response = await gdh_request(host=self.host, port=CASTPORT, + loop=self.loop, session=self.session, + endpoint=endpoint, headers=HEADERS) + self._locales = response + log.debug(self._locales) + return self._locales + + async def speedtest(self): + """Run speedtest.""" + endpoint = 'setup/test_internet_download_speed' + url = "https://storage.googleapis.com/reliability-speedtest/random.txt" + data = {"url": url} + result = await gdh_request(host=self.host, port=CASTPORT, + endpoint=endpoint, method='post', + loop=self.loop, session=self.session, + json_data=data, headers=HEADERS) + return result + + async def get_app_device_id(self): + """Run speedtest.""" + endpoint = 'setup/get_app_device_id' + data = {"app_id": "E8C28D3C"} + result = await gdh_request(host=self.host, port=CASTPORT, + endpoint=endpoint, method='post', + loop=self.loop, session=self.session, + json_data=data, headers=HEADERS) + return result + + @property + def offer(self): + """Return the offer token.""" + return self._offer + + @property + def timezones(self): + """Return supported timezones.""" + return self._timezones + + @property + def locales(self): + """Return supported timezones.""" + return self._locales + + @property + def app_device_id(self): + """Return app_device_id.""" + return self._app_device_id + @property def device_info(self): """Return the device info if any.""" diff --git a/googledevices/api/cast/settings.py b/googledevices/api/cast/settings.py index 466d667..7244183 100644 --- a/googledevices/api/cast/settings.py +++ b/googledevices/api/cast/settings.py @@ -5,7 +5,7 @@ class Settings(object): - """A class for device dettings.""" + """A class for device settings.""" def __init__(self, host, loop, session): """Initialize the class.""" @@ -13,11 +13,16 @@ def __init__(self, host, loop, session): self.loop = loop self.session = session - async def reboot(self): + async def reboot(self, mode='now'): """Reboot the device.""" endpoint = 'setup/reboot' - data = {'params': 'now'} + supported_modes = ['now', 'fdr'] returnvalue = False + if mode not in supported_modes: + log_msg = "Mode {} is not supported.".format(mode) + log.error(log_msg) + return returnvalue + data = {'params': mode} result = await gdh_request(host=self.host, port=CASTPORT, endpoint=endpoint, method='post', loop=self.loop, session=self.session, diff --git a/googledevices/api/cast/wifi.py b/googledevices/api/cast/wifi.py new file mode 100644 index 0000000..ddc5443 --- /dev/null +++ b/googledevices/api/cast/wifi.py @@ -0,0 +1,81 @@ +"""Wifi handling on Google Home units.""" +from googledevices.helpers import gdh_request +from googledevices.utils.const import HEADERS, CASTPORT +import googledevices.utils.log as log + + +class Wifi(object): + """A class for Wifi.""" + + def __init__(self, host, loop, session): + """Initialize the class.""" + self.loop = loop + self.host = host + self.session = session + self._configured_networks = None + self._nearby_networks = None + + async def get_configured_networks(self): + """Get the configured networks of the device.""" + endpoint = 'setup/configured_networks' + response = await gdh_request(host=self.host, port=CASTPORT, + loop=self.loop, session=self.session, + endpoint=endpoint, headers=HEADERS) + self._configured_networks = response + log.debug(self._configured_networks) + return self._configured_networks + + async def get_wifi_scan_result(self): + """Get the result of a wifi scan.""" + endpoint = 'setup/configured_networks' + response = await gdh_request(host=self.host, port=CASTPORT, + loop=self.loop, session=self.session, + endpoint=endpoint, headers=HEADERS) + self._configured_networks = response + log.debug(self._configured_networks) + return self._configured_networks + + async def scan_for_wifi(self): + """Scan for nearby wifi networks.""" + endpoint = 'setup/scan_wifi' + returnvalue = False + result = await gdh_request(host=self.host, port=CASTPORT, + endpoint=endpoint, method='post', + loop=self.loop, session=self.session, + headers=HEADERS, json=False) + try: + if result.status == 200: + returnvalue = True + except AttributeError: + msg = "Error connecting to - {}".format(self.host) + log.error(msg) + return returnvalue + + async def forget_network(self, wpa_id): + """Forget a network.""" + endpoint = 'setup/forget_wifi' + returnvalue = False + data = {"wpa_id": int(wpa_id)} + returnvalue = False + result = await gdh_request(host=self.host, port=CASTPORT, + endpoint=endpoint, method='post', + loop=self.loop, session=self.session, + json_data=data, headers=HEADERS, + json=False) + try: + if result.status == 200: + returnvalue = True + except AttributeError: + msg = "Error connecting to - {}".format(self.host) + log.error(msg) + return returnvalue + + @property + def configured_networks(self): + """Return the configured networks of the device.""" + return self._configured_networks + + @property + def nearby_networks(self): + """Return the nearby networks of the device.""" + return self._nearby_networks diff --git a/googledevices/api/connect.py b/googledevices/api/connect.py index 2c8109f..8117113 100644 --- a/googledevices/api/connect.py +++ b/googledevices/api/connect.py @@ -14,11 +14,6 @@ def __init__(self, host=None, loop=None, session=None): self.loop = loop self.session = session - async def alarm(self): - """Return cast Alarm connector.""" - from googledevices.api.cast.alarm import Alarm - return Alarm(self.host, self.loop, self.session) - async def bluetooth(self): """Return cast Bluetooth connector.""" from googledevices.api.cast.bluetooth import Bluetooth @@ -34,6 +29,16 @@ async def settings(self): from googledevices.api.cast.settings import Settings return Settings(self.host, self.loop, self.session) + async def assistant(self): + """Return cast DeviceSettings connector.""" + from googledevices.api.cast.assistant import Assistant + return Assistant(self.host, self.loop, self.session) + + async def wifi(self): + """Return cast DeviceSettings connector.""" + from googledevices.api.cast.wifi import Wifi as CastWifi + return CastWifi(self.host, self.loop, self.session) + class Wifi(object): """Wifi class for Google WiFi.""" diff --git a/googledevices/cli/commands/alarm_volume.py b/googledevices/cli/commands/alarm_volume.py index 6d7147b..d179144 100644 --- a/googledevices/cli/commands/alarm_volume.py +++ b/googledevices/cli/commands/alarm_volume.py @@ -1,6 +1,6 @@ """Example usage of googledevices.""" from googledevices.helpers import gdh_session -from googledevices.api.cast.alarm import Alarm +from googledevices.api.cast.assistant import Assistant def alarm_volume(host, loop, mode, volume=None): @@ -8,13 +8,13 @@ def alarm_volume(host, loop, mode, volume=None): async def set_alarm_volume(): """Get alarms and timers from GH.""" async with gdh_session() as session: - googledevices = Alarm(host, loop, session) + googledevices = Assistant(host, loop, session) await googledevices.set_alarm_volume(volume) async def get_alarm_volume(): """Get alarms and timers from GH.""" async with gdh_session() as session: - googledevices = Alarm(host, loop, session) + googledevices = Assistant(host, loop, session) await googledevices.get_alarm_volume() print("Volume:", googledevices.alarm_volume) diff --git a/googledevices/utils/test_all.py b/googledevices/utils/test_all.py index b56b3ab..e8cbafa 100644 --- a/googledevices/utils/test_all.py +++ b/googledevices/utils/test_all.py @@ -13,23 +13,50 @@ async def test_all(): # pylint: disable=R0915 """Test all.""" print("Testing Cast.") - print("Testing Cast - Alarm.") + print("Testing Cast - Assistant.") async with gdh_session() as session: - print("Testing Cast - Alarm - get_alarms") - test_class = await Cast(TEST_HOST_CAST, LOOP, session).alarm() + print("Testing Cast - Assistant - get_alarms") + test_class = await Cast(TEST_HOST_CAST, LOOP, session).assistant() test = await test_class.get_alarms() print(format_json(test)) async with gdh_session() as session: - print("Testing Cast - Alarm - set_alarm_volume") - test_class = await Cast(TEST_HOST_CAST, LOOP, session).alarm() + print("Testing Cast - Assistant - set_alarm_volume") + test_class = await Cast(TEST_HOST_CAST, LOOP, session).assistant() test = await test_class.set_alarm_volume(0.6) print(format_json(test)) async with gdh_session() as session: - print("Testing Cast - Alarm - get_alarm_volume") - test_class = await Cast(TEST_HOST_CAST, LOOP, session).alarm() + print("Testing Cast - Assistant - get_alarm_volume") + test_class = await Cast(TEST_HOST_CAST, LOOP, session).assistant() test = await test_class.get_alarm_volume() print(format_json(test)) + async with gdh_session() as session: + print("Testing Cast - Assistant - set_night_mode_params") + test_class = await Cast(TEST_HOST_CAST, LOOP, session).assistant() + data = {} + test = await test_class.set_night_mode_params(data) + print(format_json(test)) + async with gdh_session() as session: + print("Testing Cast - Assistant - notifications_enabled") + test_class = await Cast(TEST_HOST_CAST, LOOP, session).assistant() + test = await test_class.notifications_enabled() + print(format_json(test)) + async with gdh_session() as session: + print("Testing Cast - Assistant - set_accessibility") + test_class = await Cast(TEST_HOST_CAST, LOOP, session).assistant() + test = await test_class.set_accessibility() + print(format_json(test)) + async with gdh_session() as session: + print("Testing Cast - Assistant - delete_alarms") + test_class = await Cast(TEST_HOST_CAST, LOOP, session).assistant() + data = [] + test = await test_class.delete_alarms(data) + print(format_json(test)) + async with gdh_session() as session: + print("Testing Cast - Assistant - set_equalizer") + test_class = await Cast(TEST_HOST_CAST, LOOP, session).assistant() + test = await test_class.set_equalizer() + print(format_json(test)) print("Testing Cast - Bluetooth.") @@ -55,6 +82,21 @@ async def test_all(): # pylint: disable=R0915 test_class = await Cast(TEST_HOST_CAST, LOOP, session).bluetooth() test = await test_class.get_scan_result() print(format_json(test)) + async with gdh_session() as session: + print("Testing Cast - Bluetooth - get_paired_devices") + test_class = await Cast(TEST_HOST_CAST, LOOP, session).bluetooth() + test = await test_class.get_paired_devices() + print(format_json(test)) + async with gdh_session() as session: + print("Testing Cast - Bluetooth - pair_with_mac") + test_class = await Cast(TEST_HOST_CAST, LOOP, session).bluetooth() + test = await test_class.pair_with_mac("AA:BB:CC:DD:EE:FF") + print(format_json(test)) + async with gdh_session() as session: + print("Testing Cast - Bluetooth - forget_paired_device") + test_class = await Cast(TEST_HOST_CAST, LOOP, session).bluetooth() + test = await test_class.forget_paired_device("AA:BB:CC:DD:EE:FF") + print(format_json(test)) print("Testing Cast - Info.") @@ -63,6 +105,56 @@ async def test_all(): # pylint: disable=R0915 test_class = await Cast(TEST_HOST_CAST, LOOP, session).info() test = await test_class.get_device_info() print(format_json(test)) + async with gdh_session() as session: + print("Testing Cast - Info - get_offer") + test_class = await Cast(TEST_HOST_CAST, LOOP, session).info() + test = await test_class.get_offer() + print(format_json(test)) + async with gdh_session() as session: + print("Testing Cast - Info - get_timezones") + test_class = await Cast(TEST_HOST_CAST, LOOP, session).info() + test = await test_class.get_timezones() + print(format_json(test)) + async with gdh_session() as session: + print("Testing Cast - Info - get_locales") + test_class = await Cast(TEST_HOST_CAST, LOOP, session).info() + test = await test_class.get_locales() + print(format_json(test)) + async with gdh_session() as session: + print("Testing Cast - Info - speedtest") + test_class = await Cast(TEST_HOST_CAST, LOOP, session).info() + test = await test_class.speedtest() + print(format_json(test)) + async with gdh_session() as session: + print("Testing Cast - Info - get_app_device_id") + test_class = await Cast(TEST_HOST_CAST, LOOP, session).info() + test = await test_class.get_app_device_id() + print(format_json(test)) + + print("Testing Cast - Wifi.") + + async with gdh_session() as session: + print("Testing Cast - Wifi - get_configured_networks") + test_class = await Cast(TEST_HOST_CAST, LOOP, session).wifi() + test = await test_class.get_configured_networks() + print(format_json(test)) + async with gdh_session() as session: + print("Testing Cast - Wifi - scan_for_wifi") + test_class = await Cast(TEST_HOST_CAST, LOOP, session).wifi() + test = await test_class.scan_for_wifi() + print(format_json(test)) + async with gdh_session(): + await gdh_sleep(2) + async with gdh_session() as session: + print("Testing Cast - Wifi - get_wifi_scan_result") + test_class = await Cast(TEST_HOST_CAST, LOOP, session).wifi() + test = await test_class.get_wifi_scan_result() + print(format_json(test)) + async with gdh_session() as session: + print("Testing Cast - Wifi - forget_network") + test_class = await Cast(TEST_HOST_CAST, LOOP, session).wifi() + test = await test_class.forget_network(9) + print(format_json(test)) print("Testing Cast - Settings.")