fastybird_fb_mqtt_connector/connector.py
def write_property(  # pylint: disable=too-many-branches
        self,
        property_item: Union[DevicePropertyEntity, ChannelPropertyEntity],
        data: Dict,
    ) -> None:
        """Write device or channel property value to device"""
        if self.__stopped:
            self.__logger.warning("Connector is stopped, value can't be written")

            return

        if isinstance(property_item, (DeviceDynamicPropertyEntity, ChannelDynamicPropertyEntity)):
            property_record: Union[DevicePropertyRecord, ChannelPropertyRecord, None] = None

            if isinstance(property_item, DeviceDynamicPropertyEntity):
                property_record = self.__devices_properties_registry.get_by_id(property_id=property_item.id)

            elif isinstance(property_item, ChannelDynamicPropertyEntity):
                property_record = self.__channels_properties_registry.get_by_id(property_id=property_item.id)

            if property_record is None:
                return

            if property_item.data_type is not None:
                value_to_write = normalize_value(
                    data_type=property_item.data_type,
                    value=data.get("expected_value", None),
                    value_format=property_item.format,
                    value_invalid=property_item.invalid,
                )

            else:
                value_to_write = data.get("expected_value", None)

            if (
                isinstance(value_to_write, (str, int, float, bool, datetime, ButtonPayload, SwitchPayload))
                or value_to_write is None
            ):
                if (
                    isinstance(value_to_write, SwitchPayload)
                    and property_record.data_type == DataType.SWITCH
                    and value_to_write == SwitchPayload.TOGGLE
                ):
                    if property_record.actual_value == SwitchPayload.ON:
                        value_to_write = SwitchPayload.OFF

                    else:
                        value_to_write = SwitchPayload.ON

                if isinstance(property_item, DeviceDynamicPropertyEntity) and isinstance(
                    property_record, DevicePropertyRecord
                ):
                    self.__devices_properties_registry.set_expected_value(
                        device_property=property_record,
                        value=value_to_write,
                    )

                if isinstance(property_item, ChannelDynamicPropertyEntity) and isinstance(
                    property_record, ChannelPropertyRecord
                ):
                    self.__channels_properties_registry.set_expected_value(
                        channel_property=property_record,
                        value=value_to_write,
                    )

                return
Write device or channel property value to device
Write device or channel property value to device
Write device or channel property value to device
def write_property(  
        self,
        property_item: Union[DevicePropertyEntity, ChannelPropertyEntity],
        data: Dict,
    ) -> None:
        if self.__stopped:
            self.__logger.warning("Connector is stopped, value can't be written")
            return
        if isinstance(property_item, (DeviceDynamicPropertyEntity, ChannelDynamicPropertyEntity)):
            property_record: Union[DevicePropertyRecord, ChannelPropertyRecord, None] = None
            if isinstance(property_item, DeviceDynamicPropertyEntity):
                property_record = self.__devices_properties_registry.get_by_id(property_id=property_item.id)
            elif isinstance(property_item, ChannelDynamicPropertyEntity):
                property_record = self.__channels_properties_registry.get_by_id(property_id=property_item.id)
            if property_record is None:
                return
            if property_item.data_type is not None:
                value_to_write = normalize_value(
                    data_type=property_item.data_type,
                    value=data.get("expected_value", None),
                    value_format=property_item.format,
                    value_invalid=property_item.invalid,
                )
            else:
                value_to_write = data.get("expected_value", None)
            if (
                isinstance(value_to_write, (str, int, float, bool, datetime, ButtonPayload, SwitchPayload))
                or value_to_write is None
            ):
                if (
                    isinstance(value_to_write, SwitchPayload)
                    and property_record.data_type == DataType.SWITCH
                    and value_to_write == SwitchPayload.TOGGLE
                ):
                    if property_record.actual_value == SwitchPayload.ON:
                        value_to_write = SwitchPayload.OFF
                    else:
                        value_to_write = SwitchPayload.ON
                if isinstance(property_item, DeviceDynamicPropertyEntity) and isinstance(
                    property_record, DevicePropertyRecord
                ):
                    self.__devices_properties_registry.set_expected_value(
                        device_property=property_record,
                        value=value_to_write,
                    )
                if isinstance(property_item, ChannelDynamicPropertyEntity) and isinstance(
                    property_record, ChannelPropertyRecord
                ):
                    self.__channels_properties_registry.set_expected_value(
                        channel_property=property_record,
                        value=value_to_write,
                    )
                return
def write_property ( self , property_item : Union [ DevicePropertyEntity , ChannelPropertyEntity ] , data : Dict , ) -> None : if self . __stopped : self . __logger . warning ( "Connector is stopped, value can't be written" ) return if isinstance ( property_item , ( DeviceDynamicPropertyEntity , ChannelDynamicPropertyEntity ) ) : property_record : Union [ DevicePropertyRecord , ChannelPropertyRecord , None ] = None if isinstance ( property_item , DeviceDynamicPropertyEntity ) : property_record = self . __devices_properties_registry . get_by_id ( property_id = property_item . id ) elif isinstance ( property_item , ChannelDynamicPropertyEntity ) : property_record = self . __channels_properties_registry . get_by_id ( property_id = property_item . id ) if property_record is None : return if property_item . data_type is not None : value_to_write = normalize_value ( data_type = property_item . data_type , value = data . get ( "expected_value" , None ) , value_format = property_item . format , value_invalid = property_item . invalid , ) else : value_to_write = data . get ( "expected_value" , None ) if ( isinstance ( value_to_write , ( str , int , float , bool , datetime , ButtonPayload , SwitchPayload ) ) or value_to_write is None ) : if ( isinstance ( value_to_write , SwitchPayload ) and property_record . data_type == DataType . SWITCH and value_to_write == SwitchPayload . TOGGLE ) : if property_record . actual_value == SwitchPayload . ON : value_to_write = SwitchPayload . OFF else : value_to_write = SwitchPayload . ON if isinstance ( property_item , DeviceDynamicPropertyEntity ) and isinstance ( property_record , DevicePropertyRecord ) : self . __devices_properties_registry . set_expected_value ( device_property = property_record , value = value_to_write , ) if isinstance ( property_item , ChannelDynamicPropertyEntity ) and isinstance ( property_record , ChannelPropertyRecord ) : self . __channels_properties_registry . set_expected_value ( channel_property = property_record , value = value_to_write , ) return
Write device or channel property value to device
Write device or channel property value to device
# pylint: disable=too-many-branches """Write device or channel property value to device"""
self
property_item
Union[DevicePropertyEntity, ChannelPropertyEntity]
data
Dict