Module sharkiq.sharkiq

Shark IQ Wrapper.

Classes

class OperatingModes (*args, **kwds)

Expand source code

@enum.unique
class OperatingModes(enum.IntEnum):
    """
    Vacuum operation modes.

    Attributes:
        STOP: Stopped.
        PAUSE: Paused.
        START: Started.
        RETURN: Returning.
    """
    STOP = 0
    PAUSE = 1
    START = 2
    RETURN = 3

Vacuum operation modes.

Attributes

STOP

Stopped.

PAUSE

Paused.

START

Started.

RETURN

Returning.

Ancestors

• enum.IntEnum
• builtins.int
• enum.ReprEnum
• enum.Enum

Class variables

var PAUSE

The type of the None singleton.

var RETURN

The type of the None singleton.

var START

The type of the None singleton.

var STOP

The type of the None singleton.

class PowerModes (*args, **kwds)

Expand source code

@enum.unique
class PowerModes(enum.IntEnum):
    """
    Vacuum power modes.

    Attributes:
        ECO: Eco mode.
        NORMAL: Normal mode.
        MAX: Max mode.
    """
    ECO = 1
    NORMAL = 0
    MAX = 2

Vacuum power modes.

Attributes

ECO

Eco mode.

NORMAL

Normal mode.

MAX

Max mode.

Ancestors

• enum.IntEnum
• builtins.int
• enum.ReprEnum
• enum.Enum

Class variables

var ECO

The type of the None singleton.

var MAX

The type of the None singleton.

var NORMAL

The type of the None singleton.

class Properties (*args, **kwds)

Expand source code

@enum.unique
class Properties(enum.Enum):
    """
    Useful properties.
    
    Attributes:
        AREAS_TO_CLEAN: Areas to clean.
        BATTERY_CAPACITY: Battery capacity.
        CHARGING_STATUS: Charging status.
        CLEAN_COMPLETE: Cleaning complete.
        CLEANING_STATISTICS: Cleaning statistics.
        DOCKED_STATUS: Docked status.
        ERROR_CODE: Error code.
        EVACUATING: Evacuating.
        FIND_DEVICE: Find device.
        LOW_LIGHT_MISSION: Low light mission.
        NAV_MODULE_FW_VERSION: Nav module firmware version.
        OPERATING_MODE: Operating mode.
        POWER_MODE: Power mode.
        RECHARGE_RESUME: Recharge resume.
        RECHARGING_TO_RESUME: Recharging to resume.
        ROBOT_FIRMWARE_VERSION: Robot firmware version.
        RSSI: RSSI.
    """
    AREAS_TO_CLEAN = "Areas_To_Clean"
    BATTERY_CAPACITY = "Battery_Capacity"
    CHARGING_STATUS = "Charging_Status"
    CLEAN_COMPLETE = "CleanComplete"
    CLEANING_STATISTICS = "Cleaning_Statistics"
    DOCKED_STATUS = "DockedStatus"
    ERROR_CODE = "Error_Code"
    EVACUATING = "Evacuating"  # Doesn't really work because update frequency on the dock (default 20s) is too slow
    FIND_DEVICE = "Find_Device"
    LOW_LIGHT_MISSION = "LowLightMission"
    NAV_MODULE_FW_VERSION = "Nav_Module_FW_Version"
    OPERATING_MODE = "Operating_Mode"
    POWER_MODE = "Power_Mode"
    RECHARGE_RESUME = "Recharge_Resume"
    RECHARGING_TO_RESUME = "Recharging_To_Resume"
    ROBOT_FIRMWARE_VERSION = "Robot_Firmware_Version"
    ROBOT_ROOM_LIST = "Robot_Room_List"
    RSSI = "RSSI"

Useful properties.

Attributes

AREAS_TO_CLEAN

Areas to clean.

BATTERY_CAPACITY

Battery capacity.

CHARGING_STATUS

Charging status.

CLEAN_COMPLETE

Cleaning complete.

CLEANING_STATISTICS

Cleaning statistics.

DOCKED_STATUS

Docked status.

ERROR_CODE

Error code.

EVACUATING

Evacuating.

FIND_DEVICE

Find device.

LOW_LIGHT_MISSION

Low light mission.

NAV_MODULE_FW_VERSION

Nav module firmware version.

OPERATING_MODE

Operating mode.

POWER_MODE

Power mode.

RECHARGE_RESUME

Recharge resume.

RECHARGING_TO_RESUME

Recharging to resume.

ROBOT_FIRMWARE_VERSION

Robot firmware version.

RSSI

RSSI.

Ancestors

• enum.Enum

Class variables

var AREAS_TO_CLEAN

The type of the None singleton.

var BATTERY_CAPACITY

The type of the None singleton.

var CHARGING_STATUS

The type of the None singleton.

var CLEANING_STATISTICS

The type of the None singleton.

var CLEAN_COMPLETE

The type of the None singleton.

var DOCKED_STATUS

The type of the None singleton.

var ERROR_CODE

The type of the None singleton.

var EVACUATING

The type of the None singleton.

var FIND_DEVICE

The type of the None singleton.

var LOW_LIGHT_MISSION

The type of the None singleton.

var NAV_MODULE_FW_VERSION

The type of the None singleton.

var OPERATING_MODE

The type of the None singleton.

var POWER_MODE

The type of the None singleton.

var RECHARGE_RESUME

The type of the None singleton.

var RECHARGING_TO_RESUME

The type of the None singleton.

var ROBOT_FIRMWARE_VERSION

The type of the None singleton.

var ROBOT_ROOM_LIST

The type of the None singleton.

var RSSI

The type of the None singleton.

class SharkIqVacuum (ayla_api: AylaApi, device_dct: Dict, europe: bool = False)

Expand source code

class SharkIqVacuum:
    """Shark IQ vacuum entity."""

    def __init__(self, ayla_api: "AylaApi", device_dct: Dict, europe: bool = False):
        """
        Initialize a SharkIqVacuum object.

        Args:
            ayla_api: The AylaApi object.
            device_dct: The device dictionary.
            europe: True if the account is registered in Europe.
        """
        self.ayla_api = ayla_api
        self._dsn = device_dct['dsn']
        self._key = device_dct['key']
        self._oem_model_number = device_dct['oem_model']  # type: str
        self._vac_model_number = None  # type: Optional[str]
        self._vac_serial_number = None  # type: Optional[str]
        self.properties_full = defaultdict(dict)  # Using a defaultdict prevents errors before calling `update()`
        self.property_values = SharkPropertiesView(self)
        self._settable_properties = None  # type: Optional[Set]
        self.europe = europe

        # Properties
        self._name = device_dct['product_name']
        self._error = None

    @property
    def oem_model_number(self) -> str:
        """
        The OEM model number.
        
        Returns:
            The OEM model number.
        """
        return self._oem_model_number

    @property
    def vac_model_number(self) -> Optional[str]:
        """
        The vacuum model number.

        Returns:
            The vacuum model number.
        """
        return self._vac_model_number

    @property
    def vac_serial_number(self) -> Optional[str]:
        """
        The vacuum serial number.

        Returns:
            The vacuum serial number.
        """
        return self._vac_serial_number

    @property
    def name(self):
        """
        The vacuum name.

        Returns:
            The vacuum name.
        """
        return self._name

    @property
    def serial_number(self) -> str:
        """
        The vacuum serial number.

        Returns:
            The vacuum serial number.
        """
        return self._dsn

    @property
    def metadata_endpoint(self) -> str:
        """
        Endpoint for device metadata.

        Returns:
            The endpoint for device metadata.
        """
        return f'{EU_DEVICE_URL if self.europe else DEVICE_URL:s}/apiv1/dsns/{self._dsn:s}/data.json'

    def _update_metadata(self, metadata: List[Dict]):
        """
        Update metadata.

        Args:
            metadata: The metadata.
        """
        data = [d['datum'] for d in metadata if d.get('datum', {}).get('key', '') == 'sharkDeviceMobileData']
        if data:
            datum = data[0]
            # I do not know why they don't just use multiple keys for this
            try:
                values = json.loads(datum.get('value'))
            except ValueError:
                values = {}
            self._vac_model_number = values.get('vacModelNumber')
            self._vac_serial_number = values.get('vacSerialNumber')

    def get_metadata(self):
        """Fetch device metadata. Not needed for basic operation."""
        resp = self.ayla_api.request('get', self.metadata_endpoint)
        self._update_metadata(resp.json())

    async def async_get_metadata(self):
        """Fetch device metadata. Not needed for basic operation."""
        async with await self.ayla_api.async_request('get', self.metadata_endpoint) as resp:
            resp_data = await resp.json()
        self._update_metadata(resp_data)

    def set_property_endpoint(self, property_name) -> str:
        """
        Get the API endpoint for a given property.
        
        Args:
            property_name: The property name.
        
        Returns:
            The API endpoint.
        """
        return f'{EU_DEVICE_URL if self.europe else DEVICE_URL:s}/apiv1/dsns/{self._dsn:s}/properties/{property_name:s}/datapoints.json'

    def get_property_value(self, property_name: PropertyName) -> Any:
        """
        Get the value of a property from the properties dictionary.
        
        Args:
            property_name: The property name.
        
        Returns:
            The property value.
        """
        if isinstance(property_name, enum.Enum):
            property_name = property_name.value
        return self.property_values[property_name]

    def set_property_value(self, property_name: PropertyName, value: PropertyValue):
        """
        Update a property.

        Args:
            property_name: The property name.
            value: The property value.
        """
        if isinstance(property_name, enum.Enum):
            property_name = property_name.value
        if isinstance(value, enum.Enum):
            value = value.value
        if self.properties_full.get(property_name, {}).get('read_only'):
            raise SharkIqReadOnlyPropertyError(f'{property_name} is read only')

        end_point = self.set_property_endpoint(f'SET_{property_name}')
        data = {'datapoint': {'value': value}}
        resp = self.ayla_api.request('post', end_point, json=data)
        self.properties_full[property_name].update(resp.json())

    async def async_set_property_value(self, property_name: PropertyName, value: PropertyValue):
        """
        Update a property async.

        Args:
            property_name: The property name.
            value: The property value.
        """
        if isinstance(property_name, enum.Enum):
            property_name = property_name.value
        if isinstance(value, enum.Enum):
            value = value.value

        end_point = self.set_property_endpoint(f'SET_{property_name}')
        data = {'datapoint': {'value': value}}
        async with await self.ayla_api.async_request('post', end_point, json=data) as resp:
            resp_data = await resp.json()
        self.properties_full[property_name].update(resp_data)

    @property
    def update_url(self) -> str:
        """
        API endpoint to fetch updated device information.
        
        Returns:
            The API endpoint.
        """
        return f'{EU_DEVICE_URL if self.europe else DEVICE_URL}/apiv1/dsns/{self.serial_number}/properties.json'

    def update(self, property_list: Optional[Iterable[str]] = None):
        """
        Update the known device state.

        Args:
            property_list: The list of properties to update.
        """
        full_update = property_list is None
        if full_update:
            params = None
        else:
            params = {'names[]': property_list}

        resp = self.ayla_api.request('get', self.update_url, params=params)
        properties = resp.json()
        self._do_update(full_update, properties)

    async def async_update(self, property_list: Optional[Iterable[str]] = None):
        """
        Update the known device state async.
        
        Args:
            property_list: The list of properties to update.
        """
        full_update = property_list is None
        if full_update:
            params = None
        else:
            params = {'names[]': property_list}

        async with await self.ayla_api.async_request('get', self.update_url, params=params) as resp:
            properties = await resp.json()

        self._do_update(full_update, properties)

    def _do_update(self, full_update: bool, properties: List[Dict]):
        """
        Update the internal state from fetched properties.
        
        Args:
            full_update: Whether to update all properties.
            properties: The properties.
        """
        property_names = {p['property']['name'] for p in properties}
        settable_properties = {_clean_property_name(p) for p in property_names if p[:3].upper() == 'SET'}
        readable_properties = {
            _clean_property_name(p['property']['name']): p['property']
            for p in properties if p['property']['name'].upper() != 'SET'
        }

        if full_update or self._settable_properties is None:
            self._settable_properties = settable_properties
        else:
            self._settable_properties = self._settable_properties.union(settable_properties)

        # Update the property map so we can update by name instead of by fickle number
        if full_update:
            # Did a full update, so let's wipe everything
            self.properties_full = defaultdict(dict)
        self.properties_full.update(readable_properties)

    def set_operating_mode(self, mode: OperatingModes):
        """
        Set the operating mode. This is just a convenience wrapper around `set_property_value`.
        
        Args:
            mode: The operating mode.
        """
        self.set_property_value(Properties.OPERATING_MODE, mode)

    async def async_set_operating_mode(self, mode: OperatingModes):
        """
        Set the operating mode. This is just a convenience wrapper around `set_property_value`.
        
        Args:
            mode: The operating mode.
        """
        await self.async_set_property_value(Properties.OPERATING_MODE, mode)

    def find_device(self):
        """Make the device emit an annoying chirp so you can find it."""
        self.set_property_value(Properties.FIND_DEVICE, 1)

    async def async_find_device(self):
        """Make the device emit an annoying chirp so you can find it."""
        await self.async_set_property_value(Properties.FIND_DEVICE, 1)

    @property
    def error_code(self) -> Optional[int]:
        """
        Error code.

        Returns:
            The error code.
        """
        return self.get_property_value(Properties.ERROR_CODE)

    @property
    def error_text(self) -> Optional[str]:
        """
        Error message.
        
        Returns:
            The error message.
        """
        err = self.error_code
        if err:
            return ERROR_MESSAGES.get(err, f'Unknown error ({err})')
        return None

    @staticmethod
    def _get_most_recent_datum(data_list: List[Dict], date_field: str = 'updated_at') -> Dict:
        """
        Get the most recent data point from a list of annoyingly nested values.
        
        Args:
            data_list: The list of data points.
            date_field: The field to use for the date.
            
        Returns:
            The most recent data point.
        """
        datapoints = {
            _parse_datetime(d['datapoint'][date_field]): d['datapoint'] for d in data_list if 'datapoint' in d
        }
        if not datapoints:
            return {}
        latest_datum = datapoints[max(datapoints.keys())]
        return latest_datum

    def _get_file_property_endpoint(self, property_name: PropertyName) -> str:
        """
        Check that property_name is a file property and return its lookup endpoint.
        
        Args:
            property_name: The property name.
            
        Returns:
            The endpoint.
        """
        if isinstance(property_name, enum.Enum):
            property_name = property_name.value

        property_id = self.properties_full[property_name]['key']
        if self.properties_full[property_name].get('base_type') != 'file':
            raise ValueError(f'{property_name} is not a file property')
        return f'{EU_DEVICE_URL if self.europe else DEVICE_URL:s}/apiv1/properties/{property_id:d}/datapoints.json'

    def get_file_property_url(self, property_name: PropertyName) -> Optional[str]:
        """
        File properties are versioned and need a special lookup.
        
        Args:
            property_name: The property name.
            
        Returns:
            The URL.
        """
        try:
            url = self._get_file_property_endpoint(property_name)
        except KeyError:
            return None

        resp = self.ayla_api.request('get', url)
        data_list = resp.json()
        latest_datum = self._get_most_recent_datum(data_list)
        return latest_datum.get('file')

    async def async_get_file_property_url(self, property_name: PropertyName) -> Optional[str]:
        """
        File properties are versioned and need a special lookup.
        
        Args:
            property_name: The property name.
            
        Returns:
            The URL.
        """
        try:
            url = self._get_file_property_endpoint(property_name)
        except KeyError:
            return None

        async with await self.ayla_api.async_request('get', url) as resp:
            data_list = await resp.json()
        latest_datum = self._get_most_recent_datum(data_list)
        return latest_datum.get('file')

    def get_file_property(self, property_name: PropertyName) -> bytes:
        """
        Get the latest file for a file property and return as bytes.
        
        Args:
            property_name: The property name.
            
        Returns:
            The file as bytes.
        """
        # These do not require authentication, so we won't use the ayla_api
        url = self.get_file_property_url(property_name)
        resp = requests.get(url)
        return resp.content

    async def async_get_file_property(self, property_name: PropertyName) -> bytes:
        """
        Get the latest file for a file property and return as bytes.
        
        Args:
            property_name: The property name.
            
        Returns:
            The file as bytes.
        """
        url = await self.async_get_file_property_url(property_name)
        session = self.ayla_api.websession
        async with session.get(url) as resp:
            return await resp.read()

    def _encode_room_list(self, rooms: List[str]):
        """
        Base64 encode the list of rooms to clean.
        
        Args:
            rooms: The list of rooms.
            
        Returns:
            The base64 encoded list of rooms.
        """
        if not rooms:
            # By default, clean all rooms
            return '*'

        room_list = self._get_device_room_list()
        _LOGGER.debug(f'Room list identifier is: {room_list["identifier"]}')

        # Header explained:
        # 0x80: Control character - some mode selection
        # 0x01: Start of Heading Character
        # 0x0B: Use Line Tabulation (entries separated by newlines)
        # 0xca: Control character - purpose unknown
        # 0x02: Start of text (indicates start of room list)
        header = '\x80\x01\x0b\xca\x02'

        # For each room in the list:
        # - Insert a byte representing the length of the room name string
        # - Add the room name
        # - Join with newlines (presumably because of the 0x0B in the header)
        rooms_enc = "\n".join([chr(len(room)) + room for room in rooms])

        # The footer starts with control character 0x1A
        # Then add the length indicator for the room list identifier
        # Then add the room list identifier
        footer = '\x1a' + chr(len(room_list['identifier'])) + room_list['identifier']

        # Now that we've computed the room list and footer and know their lengths, finish building the header
        # This character denotes the length of the remaining input
        header += chr(0
                      + 1  # Add one for a newline following the length specifier
                      + len(rooms_enc)
                      + len(footer)
                      )
        header += '\n'  # This is the newline reference above

        # Finally, join and base64 encode the parts
        return base64.b64encode(
            # First encode the string as latin_1 to get the right endianness
            (header + rooms_enc + footer).encode('latin_1')
            # Then return as a utf8 string for ease of handling
        ).decode('utf8')

    def _get_device_room_list(self):
        """Gets the list of known rooms from the device, including the map identifier"""
        room_list = self.get_property_value(Properties.ROBOT_ROOM_LIST)
        split = room_list.split(':')
        return {
            # The room list is preceded by an identifier, which I believe identifies the list of rooms with the
            # onboard map in the robot
            'identifier': split[0],
            'rooms': split[1:],
        }

    def get_room_list(self) -> List[str]:
        """Gets the list of rooms known by the device"""
        return self._get_device_room_list()['rooms']

    def clean_rooms(self, rooms: List[str]) -> None:
        """
        Clean the given rooms.

        Args:
            rooms: The list of rooms to clean.
        """
        payload = self._encode_room_list(rooms)
        _LOGGER.debug('Room list payload: ' + payload)
        self.set_property_value(Properties.AREAS_TO_CLEAN, payload)
        self.set_operating_mode(OperatingModes.START)

    async def async_clean_rooms(self, rooms: List[str]) -> None:
        """
        Clean the given rooms.

        Args:
            rooms: The list of rooms to clean.
        """
        payload = self._encode_room_list(rooms)
        _LOGGER.debug("Room list payload: " + payload)
        await self.async_set_property_value(Properties.AREAS_TO_CLEAN, payload)
        await self.async_set_operating_mode(OperatingModes.START)

Shark IQ vacuum entity.

Initialize a SharkIqVacuum object.

Args

ayla_api

The AylaApi object.

device_dct

The device dictionary.

europe

True if the account is registered in Europe.

Instance variables

prop error_code : int | None

Expand source code

@property
def error_code(self) -> Optional[int]:
    """
    Error code.

    Returns:
        The error code.
    """
    return self.get_property_value(Properties.ERROR_CODE)

Error code.

Returns

The error code.

prop error_text : str | None

Expand source code

@property
def error_text(self) -> Optional[str]:
    """
    Error message.
    
    Returns:
        The error message.
    """
    err = self.error_code
    if err:
        return ERROR_MESSAGES.get(err, f'Unknown error ({err})')
    return None

Error message.

Returns

The error message.

prop metadata_endpoint : str

Expand source code

@property
def metadata_endpoint(self) -> str:
    """
    Endpoint for device metadata.

    Returns:
        The endpoint for device metadata.
    """
    return f'{EU_DEVICE_URL if self.europe else DEVICE_URL:s}/apiv1/dsns/{self._dsn:s}/data.json'

Endpoint for device metadata.

Returns

The endpoint for device metadata.

prop name

Expand source code

@property
def name(self):
    """
    The vacuum name.

    Returns:
        The vacuum name.
    """
    return self._name

The vacuum name.

Returns

The vacuum name.

prop oem_model_number : str

Expand source code

@property
def oem_model_number(self) -> str:
    """
    The OEM model number.
    
    Returns:
        The OEM model number.
    """
    return self._oem_model_number

The OEM model number.

Returns

The OEM model number.

prop serial_number : str

Expand source code

@property
def serial_number(self) -> str:
    """
    The vacuum serial number.

    Returns:
        The vacuum serial number.
    """
    return self._dsn

The vacuum serial number.

Returns

The vacuum serial number.

prop update_url : str

Expand source code

@property
def update_url(self) -> str:
    """
    API endpoint to fetch updated device information.
    
    Returns:
        The API endpoint.
    """
    return f'{EU_DEVICE_URL if self.europe else DEVICE_URL}/apiv1/dsns/{self.serial_number}/properties.json'

API endpoint to fetch updated device information.

Returns

The API endpoint.

prop vac_model_number : str | None

Expand source code

@property
def vac_model_number(self) -> Optional[str]:
    """
    The vacuum model number.

    Returns:
        The vacuum model number.
    """
    return self._vac_model_number

The vacuum model number.

Returns

The vacuum model number.

prop vac_serial_number : str | None

Expand source code

@property
def vac_serial_number(self) -> Optional[str]:
    """
    The vacuum serial number.

    Returns:
        The vacuum serial number.
    """
    return self._vac_serial_number

The vacuum serial number.

Returns

The vacuum serial number.

Methods

async def async_clean_rooms(self, rooms: List[str]) ‑> None

Expand source code

async def async_clean_rooms(self, rooms: List[str]) -> None:
    """
    Clean the given rooms.

    Args:
        rooms: The list of rooms to clean.
    """
    payload = self._encode_room_list(rooms)
    _LOGGER.debug("Room list payload: " + payload)
    await self.async_set_property_value(Properties.AREAS_TO_CLEAN, payload)
    await self.async_set_operating_mode(OperatingModes.START)

Clean the given rooms.

Args

rooms

The list of rooms to clean.

async def async_find_device(self)

Expand source code

async def async_find_device(self):
    """Make the device emit an annoying chirp so you can find it."""
    await self.async_set_property_value(Properties.FIND_DEVICE, 1)

Make the device emit an annoying chirp so you can find it.

async def async_get_file_property(self, property_name: str | enum.Enum) ‑> bytes

Expand source code

async def async_get_file_property(self, property_name: PropertyName) -> bytes:
    """
    Get the latest file for a file property and return as bytes.
    
    Args:
        property_name: The property name.
        
    Returns:
        The file as bytes.
    """
    url = await self.async_get_file_property_url(property_name)
    session = self.ayla_api.websession
    async with session.get(url) as resp:
        return await resp.read()

Get the latest file for a file property and return as bytes.

Args

property_name

The property name.

Returns

The file as bytes.

async def async_get_file_property_url(self, property_name: str | enum.Enum) ‑> str | None

Expand source code

async def async_get_file_property_url(self, property_name: PropertyName) -> Optional[str]:
    """
    File properties are versioned and need a special lookup.
    
    Args:
        property_name: The property name.
        
    Returns:
        The URL.
    """
    try:
        url = self._get_file_property_endpoint(property_name)
    except KeyError:
        return None

    async with await self.ayla_api.async_request('get', url) as resp:
        data_list = await resp.json()
    latest_datum = self._get_most_recent_datum(data_list)
    return latest_datum.get('file')

File properties are versioned and need a special lookup.

Args

property_name

The property name.

Returns

The URL.

async def async_get_metadata(self)

Expand source code

async def async_get_metadata(self):
    """Fetch device metadata. Not needed for basic operation."""
    async with await self.ayla_api.async_request('get', self.metadata_endpoint) as resp:
        resp_data = await resp.json()
    self._update_metadata(resp_data)

Fetch device metadata. Not needed for basic operation.

async def async_set_operating_mode(self,
mode: OperatingModes)

Expand source code

async def async_set_operating_mode(self, mode: OperatingModes):
    """
    Set the operating mode. This is just a convenience wrapper around `set_property_value`.
    
    Args:
        mode: The operating mode.
    """
    await self.async_set_property_value(Properties.OPERATING_MODE, mode)

Set the operating mode. This is just a convenience wrapper around set_property_value.

Args

mode

The operating mode.

async def async_set_property_value(self, property_name: str | enum.Enum, value: str | int | enum.Enum)

Expand source code

async def async_set_property_value(self, property_name: PropertyName, value: PropertyValue):
    """
    Update a property async.

    Args:
        property_name: The property name.
        value: The property value.
    """
    if isinstance(property_name, enum.Enum):
        property_name = property_name.value
    if isinstance(value, enum.Enum):
        value = value.value

    end_point = self.set_property_endpoint(f'SET_{property_name}')
    data = {'datapoint': {'value': value}}
    async with await self.ayla_api.async_request('post', end_point, json=data) as resp:
        resp_data = await resp.json()
    self.properties_full[property_name].update(resp_data)

Update a property async.

Args

property_name

The property name.

value

The property value.

async def async_update(self, property_list: Iterable[str] | None = None)

Expand source code

async def async_update(self, property_list: Optional[Iterable[str]] = None):
    """
    Update the known device state async.
    
    Args:
        property_list: The list of properties to update.
    """
    full_update = property_list is None
    if full_update:
        params = None
    else:
        params = {'names[]': property_list}

    async with await self.ayla_api.async_request('get', self.update_url, params=params) as resp:
        properties = await resp.json()

    self._do_update(full_update, properties)

Update the known device state async.

Args

property_list

The list of properties to update.

def clean_rooms(self, rooms: List[str]) ‑> None

Expand source code

def clean_rooms(self, rooms: List[str]) -> None:
    """
    Clean the given rooms.

    Args:
        rooms: The list of rooms to clean.
    """
    payload = self._encode_room_list(rooms)
    _LOGGER.debug('Room list payload: ' + payload)
    self.set_property_value(Properties.AREAS_TO_CLEAN, payload)
    self.set_operating_mode(OperatingModes.START)

Clean the given rooms.

Args

rooms

The list of rooms to clean.

def find_device(self)

Expand source code

def find_device(self):
    """Make the device emit an annoying chirp so you can find it."""
    self.set_property_value(Properties.FIND_DEVICE, 1)

Make the device emit an annoying chirp so you can find it.

def get_file_property(self, property_name: str | enum.Enum) ‑> bytes

Expand source code

def get_file_property(self, property_name: PropertyName) -> bytes:
    """
    Get the latest file for a file property and return as bytes.
    
    Args:
        property_name: The property name.
        
    Returns:
        The file as bytes.
    """
    # These do not require authentication, so we won't use the ayla_api
    url = self.get_file_property_url(property_name)
    resp = requests.get(url)
    return resp.content

Get the latest file for a file property and return as bytes.

Args

property_name

The property name.

Returns

The file as bytes.

def get_file_property_url(self, property_name: str | enum.Enum) ‑> str | None

Expand source code

def get_file_property_url(self, property_name: PropertyName) -> Optional[str]:
    """
    File properties are versioned and need a special lookup.
    
    Args:
        property_name: The property name.
        
    Returns:
        The URL.
    """
    try:
        url = self._get_file_property_endpoint(property_name)
    except KeyError:
        return None

    resp = self.ayla_api.request('get', url)
    data_list = resp.json()
    latest_datum = self._get_most_recent_datum(data_list)
    return latest_datum.get('file')

File properties are versioned and need a special lookup.

Args

property_name

The property name.

Returns

The URL.

def get_metadata(self)

Expand source code

def get_metadata(self):
    """Fetch device metadata. Not needed for basic operation."""
    resp = self.ayla_api.request('get', self.metadata_endpoint)
    self._update_metadata(resp.json())

Fetch device metadata. Not needed for basic operation.

def get_property_value(self, property_name: str | enum.Enum) ‑> Any

Expand source code

def get_property_value(self, property_name: PropertyName) -> Any:
    """
    Get the value of a property from the properties dictionary.
    
    Args:
        property_name: The property name.
    
    Returns:
        The property value.
    """
    if isinstance(property_name, enum.Enum):
        property_name = property_name.value
    return self.property_values[property_name]

Get the value of a property from the properties dictionary.

Args

property_name

The property name.

Returns

The property value.

def get_room_list(self) ‑> List[str]

Expand source code

def get_room_list(self) -> List[str]:
    """Gets the list of rooms known by the device"""
    return self._get_device_room_list()['rooms']

Gets the list of rooms known by the device

def set_operating_mode(self,
mode: OperatingModes)

Expand source code

def set_operating_mode(self, mode: OperatingModes):
    """
    Set the operating mode. This is just a convenience wrapper around `set_property_value`.
    
    Args:
        mode: The operating mode.
    """
    self.set_property_value(Properties.OPERATING_MODE, mode)

Set the operating mode. This is just a convenience wrapper around set_property_value.

Args

mode

The operating mode.

def set_property_endpoint(self, property_name) ‑> str

Expand source code

def set_property_endpoint(self, property_name) -> str:
    """
    Get the API endpoint for a given property.
    
    Args:
        property_name: The property name.
    
    Returns:
        The API endpoint.
    """
    return f'{EU_DEVICE_URL if self.europe else DEVICE_URL:s}/apiv1/dsns/{self._dsn:s}/properties/{property_name:s}/datapoints.json'

Get the API endpoint for a given property.

Args

property_name

The property name.

Returns

The API endpoint.

def set_property_value(self, property_name: str | enum.Enum, value: str | int | enum.Enum)

Expand source code

def set_property_value(self, property_name: PropertyName, value: PropertyValue):
    """
    Update a property.

    Args:
        property_name: The property name.
        value: The property value.
    """
    if isinstance(property_name, enum.Enum):
        property_name = property_name.value
    if isinstance(value, enum.Enum):
        value = value.value
    if self.properties_full.get(property_name, {}).get('read_only'):
        raise SharkIqReadOnlyPropertyError(f'{property_name} is read only')

    end_point = self.set_property_endpoint(f'SET_{property_name}')
    data = {'datapoint': {'value': value}}
    resp = self.ayla_api.request('post', end_point, json=data)
    self.properties_full[property_name].update(resp.json())

Update a property.

Args

property_name

The property name.

value

The property value.

def update(self, property_list: Iterable[str] | None = None)

Expand source code

def update(self, property_list: Optional[Iterable[str]] = None):
    """
    Update the known device state.

    Args:
        property_list: The list of properties to update.
    """
    full_update = property_list is None
    if full_update:
        params = None
    else:
        params = {'names[]': property_list}

    resp = self.ayla_api.request('get', self.update_url, params=params)
    properties = resp.json()
    self._do_update(full_update, properties)

Update the known device state.

Args

property_list

The list of properties to update.

class SharkPropertiesView (shark: SharkIqVacuum)

Expand source code

class SharkPropertiesView(abc.Mapping):
    """Convenience API for shark iq properties"""

    @staticmethod
    def _cast_value(value, value_type):
        """
        Cast property value to the appropriate type.

        Args:
            value: The value to cast.
            value_type: The type to cast to.

        Returns:
            The cast value.
        """
        if value is None:
            return None
        type_map = {
            'boolean': bool,
            'decimal': float,
            'integer': int,
            'string': str,
        }
        return type_map.get(value_type, lambda x: x)(value)

    def __init__(self, shark: SharkIqVacuum):
        """
        Initialize the shark properties view.
        
        Args:
            shark: The shark iq vacuum.
        """
        self._shark = shark

    def __getitem__(self, key):
        """
        Get a property value.

        Args:
            key: The property name.

        Returns:
            The property value.
        """
        value = self._shark.properties_full[key].get('value')
        value_type = self._shark.properties_full[key].get('base_type')
        try:
            return self._cast_value(value, value_type)
        except (TypeError, ValueError) as exc:
            # If we failed to convert the type, just return the raw value
            _LOGGER.warning('Error converting property type (value: %r, type: %r)', value, value_type, exc_info=exc)
            return value

    def __iter__(self):
        """Iterate over the properties."""
        for k in self._shark.properties_full.keys():
            yield k

    def __len__(self) -> int:
        """Return the number of properties."""
        return self._shark.properties_full.__len__()

    def __str__(self) -> str:
        """Return a string representation of the properties."""
        return pformat(dict(self))

Convenience API for shark iq properties

Initialize the shark properties view.

Args

shark

The shark iq vacuum.

Ancestors

• collections.abc.Mapping
• collections.abc.Collection
• collections.abc.Sized
• collections.abc.Iterable
• collections.abc.Container