Skip to content

Commit

Permalink
replace runner argument with rest_connection in fc actions
Browse files Browse the repository at this point in the history
  • Loading branch information
hahn-th committed May 30, 2024
1 parent 113ce9b commit db1f276
Show file tree
Hide file tree
Showing 2 changed files with 84 additions and 83 deletions.
111 changes: 56 additions & 55 deletions src/homematicip/action/functional_channel_actions.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
from homematicip.action.action import Action
from homematicip.connection.rest_connection import RestConnection
from homematicip.model.enums import AccelerationSensorMode, AccelerationSensorNeutralPosition, \
AccelerationSensorSensitivity, NotificationSoundType, DoorCommand, LockState, RGBColorState, ClimateControlDisplay, \
AcousticAlarmSignal, AcousticAlarmTiming, WaterAlarmTrigger, OpticalSignalBehaviour
Expand All @@ -7,7 +8,7 @@


@Action.allowed_types('BLIND_CHANNEL', 'MULTI_MODE_INPUT_BLIND_CHANNEL')
async def action_set_slats_level(runner: Runner, fc: FunctionalChannel, slats_level: float,
async def action_set_slats_level(rest_connection: RestConnection, fc: FunctionalChannel, slats_level: float,
shutter_level: float = None):
if shutter_level is None:
shutter_level = fc.shutterLevel
Expand All @@ -17,145 +18,145 @@ async def action_set_slats_level(runner: Runner, fc: FunctionalChannel, slats_le
"slatsLevel": slats_level,
"shutterLevel": shutter_level
}
return await runner.rest_connection.async_post("device/control/setSlatsLevel", data)
return await rest_connection.async_post("device/control/setSlatsLevel", data)


@Action.allowed_types('BLIND_CHANNEL', 'MULTI_MODE_INPUT_BLIND_CHANNEL', 'SHUTTER_CHANNEL')
async def action_set_shutter_level(runner: Runner, fc: FunctionalChannel, shutter_level: float):
async def action_set_shutter_level(rest_connection: RestConnection, fc: FunctionalChannel, shutter_level: float):
data = {"channelIndex": fc.index, "deviceId": fc.deviceId, "shutterLevel": shutter_level}
return await runner.rest_connection.async_post("device/control/setShutterLevel", data)
return await rest_connection.async_post("device/control/setShutterLevel", data)


@Action.allowed_types('SWITCH_CHANNEL',
'MULTI_MODE_INPUT_SWITCH_CHANNEL',
'SWITCH_MEASURING_CHANNEL',
'OPTICAL_SIGNAL_CHANNEL')
async def action_set_switch_state(runner: Runner, fc: FunctionalChannel, on: bool):
async def action_set_switch_state(rest_connection: RestConnection, fc: FunctionalChannel, on: bool):
data = {"channelIndex": fc.index, "deviceId": fc.deviceId, "on": on}
return await runner.rest_connection.async_post("device/control/setSwitchState", data)
return await rest_connection.async_post("device/control/setSwitchState", data)


@Action.allowed_types('IMPULSE_OUTPUT_CHANNEL')
async def action_start_impulse(runner: Runner, fc: FunctionalChannel):
async def action_start_impulse(rest_connection: RestConnection, fc: FunctionalChannel):
"""Toggle Wall mounted Garage Door Controller."""
data = {"channelIndex": fc.index, "deviceId": fc.deviceId}
return await runner.rest_connection.async_post("device/control/startImpulse", data)
return await rest_connection.async_post("device/control/startImpulse", data)


@Action.allowed_types("DIMMER_CHANNEL", "MULTI_MODE_INPUT_DIMMER_CHANNEL", "NOTIFICATION_LIGHT_CHANNEL")
async def action_set_dim_level(runner: Runner, fc: FunctionalChannel, dim_level: float):
async def action_set_dim_level(rest_connection: RestConnection, fc: FunctionalChannel, dim_level: float):
data = {"channelIndex": fc.index, "deviceId": fc.deviceId, "dimLevel": dim_level}
return await runner.rest_connection.async_post("device/control/setDimLevel", data)
return await rest_connection.async_post("device/control/setDimLevel", data)


@Action.allowed_types("ACCELERATION_SENSOR_CHANNEL", "TILT_VIBRATION_SENSOR")
async def set_acceleration_sensor_mode_action(runner: Runner, fc: FunctionalChannel, mode: AccelerationSensorMode):
async def set_acceleration_sensor_mode_action(rest_connection: RestConnection, fc: FunctionalChannel, mode: AccelerationSensorMode):
data = {
"channelIndex": fc.index,
"deviceId": fc.deviceId,
"accelerationSensorMode": str(mode),
}
return await runner.rest_connection.async_post("device/configuration/setAccelerationSensorMode", data)
return await rest_connection.async_post("device/configuration/setAccelerationSensorMode", data)


@Action.allowed_types("ACCELERATION_SENSOR_CHANNEL", "TILT_VIBRATION_SENSOR")
async def set_acceleration_sensor_neutral_position(runner: Runner, fc: FunctionalChannel,
async def set_acceleration_sensor_neutral_position(rest_connection: RestConnection, fc: FunctionalChannel,
neutral_position: AccelerationSensorNeutralPosition):
data = {
"channelIndex": fc.index,
"deviceId": fc.deviceId,
"accelerationSensorNeutralPosition": str(neutral_position)
}
return await runner.rest_connection.async_post("device/configuration/setAccelerationSensorNeutralPosition", data)
return await rest_connection.async_post("device/configuration/setAccelerationSensorNeutralPosition", data)


@Action.allowed_types("ACCELERATION_SENSOR_CHANNEL", "TILT_VIBRATION_SENSOR")
async def set_acceleration_sensor_sensitivity(runner: Runner, fc: FunctionalChannel,
async def set_acceleration_sensor_sensitivity(rest_connection: RestConnection, fc: FunctionalChannel,
sensitivity: AccelerationSensorSensitivity):
data = {
"channelIndex": fc.index,
"deviceId": fc.deviceId,
"accelerationSensorSensitivity": str(sensitivity),
}
return await runner.rest_connection.async_post("device/configuration/setAccelerationSensorSensitivity", data)
return await rest_connection.async_post("device/configuration/setAccelerationSensorSensitivity", data)


@Action.allowed_types("ACCELERATION_SENSOR_CHANNEL", "TILT_VIBRATION_SENSOR")
async def set_acceleration_sensor_trigger_angle(runner: Runner, fc: FunctionalChannel, angle: int):
async def set_acceleration_sensor_trigger_angle(rest_connection: RestConnection, fc: FunctionalChannel, angle: int):
data = {
"channelIndex": fc.index,
"deviceId": fc.deviceId,
"accelerationSensorTriggerAngle": angle
}
return await runner.rest_connection.async_post("device/configuration/setAccelerationSensorTriggerAngle", data)
return await rest_connection.async_post("device/configuration/setAccelerationSensorTriggerAngle", data)


@Action.allowed_types("ACCELERATION_SENSOR_CHANNEL", "TILT_VIBRATION_SENSOR")
async def set_acceleration_sensor_event_filter_period(runner: Runner, fc: FunctionalChannel, period: float):
async def set_acceleration_sensor_event_filter_period(rest_connection: RestConnection, fc: FunctionalChannel, period: float):
data = {
"channelIndex": fc.index,
"deviceId": fc.deviceId,
"accelerationSensorEventFilterPeriod": period
}
return await runner.rest_connection.async_post("device/configuration/setAccelerationSensorEventFilterPeriod", data)
return await rest_connection.async_post("device/configuration/setAccelerationSensorEventFilterPeriod", data)


@Action.allowed_types("ACCELERATION_SENSOR_CHANNEL")
async def set_notification_sound_type(runner: Runner, fc: FunctionalChannel, sound_type: NotificationSoundType,
async def set_notification_sound_type(rest_connection: RestConnection, fc: FunctionalChannel, sound_type: NotificationSoundType,
is_high_to_low: bool):
data = {
"channelIndex": fc.index,
"deviceId": fc.deviceId,
"notificationSoundType": str(sound_type),
"isHighToLow": is_high_to_low
}
return await runner.rest_connection.async_post("device/configuration/setNotificationSoundType", data)
return await rest_connection.async_post("device/configuration/setNotificationSoundType", data)


@Action.allowed_types("DEVICE_BASE_FLOOR_HEATING")
async def set_minimum_floor_heating_valve_position(runner: Runner, fc: FunctionalChannel, position: float):
async def set_minimum_floor_heating_valve_position(rest_connection: RestConnection, fc: FunctionalChannel, position: float):
data = {
"channelIndex": fc.index,
"deviceId": fc.deviceId,
"minimumFloorHeatingValvePosition": position
}
return await runner.rest_connection.async_post("device/configuration/setMinimumFloorHeatingValvePosition", data)
return await rest_connection.async_post("device/configuration/setMinimumFloorHeatingValvePosition", data)


@Action.allowed_types("DEVICE_OPERATIONLOCK", "DEVICE_OPERATIONLOCK_WITH_SABOTAGE")
async def set_operation_lock(runner: Runner, fc: FunctionalChannel, operation_lock: bool):
async def set_operation_lock(rest_connection: RestConnection, fc: FunctionalChannel, operation_lock: bool):
data = {
"channelIndex": fc.index,
"deviceId": fc.deviceId,
"operationLock": operation_lock
}
return await runner.rest_connection.async_post("device/configuration/setOperationLock", data)
return await rest_connection.async_post("device/configuration/setOperationLock", data)


@Action.allowed_types("DOOR_CHANNEL")
async def action_send_door_command(runner: Runner, fc: FunctionalChannel, door_command: DoorCommand):
async def action_send_door_command(rest_connection: RestConnection, fc: FunctionalChannel, door_command: DoorCommand):
data = {
"channelIndex": fc.index,
"deviceId": fc.deviceId,
"doorCommand": str(door_command),
}
return await runner.rest_connection.async_post("device/control/sendDoorCommand", data)
return await rest_connection.async_post("device/control/sendDoorCommand", data)


@Action.allowed_types("DOOR_LOCK_CHANNEL")
async def action_set_door_state(runner: Runner, fc: FunctionalChannel, lock_state: LockState, pin: str = None):
async def action_set_door_state(rest_connection: RestConnection, fc: FunctionalChannel, lock_state: LockState, pin: str = None):
data = {
"deviceId": fc.deviceId,
"channelIndex": fc.index,
"authorizationPin": pin,
"targetLockState": str(lock_state),
}
return await runner.rest_connection.async_post("device/control/setLockState", data)
return await rest_connection.async_post("device/control/setLockState", data)


@Action.allowed_types("NOTIFICATION_LIGHT_CHANNEL", "OPTICAL_SIGNAL_CHANNEL")
async def action_set_optical_signal(
runner: Runner,
rest_connection: RestConnection,
fc: FunctionalChannel,
optical_signal_behaviour: OpticalSignalBehaviour,
rgb: RGBColorState,
Expand All @@ -164,7 +165,7 @@ async def action_set_optical_signal(
"""sets the signal type for the leds
Args:
runner(Runner): The executing runner. Contains the connection to the homematicip cloud
rest_connection(RestConnection): The rest connection
fc(FunctionalChannel): The functional channel the action should be executed on
optical_signal_behaviour(OpticalSignalBehaviour): LED signal behaviour
rgb(RGBColorState): Color
Expand All @@ -181,23 +182,23 @@ async def action_set_optical_signal(
"opticalSignalBehaviour": str(optical_signal_behaviour),
"simpleRGBColorState": str(rgb),
}
return await runner.rest_connection.async_post("device/control/setOpticalSignal", data)
return await rest_connection.async_post("device/control/setOpticalSignal", data)


@Action.allowed_types("NOTIFICATION_LIGHT_CHANNEL")
async def action_set_rgb_dim_level(runner: Runner, fc: FunctionalChannel, rgb: RGBColorState, dim_level: float):
async def action_set_rgb_dim_level(rest_connection: RestConnection, fc: FunctionalChannel, rgb: RGBColorState, dim_level: float):
data = {
"channelIndex": fc.index,
"deviceId": fc.deviceId,
"simpleRGBColorState": str(rgb),
"dimLevel": dim_level,
}
return await runner.rest_connection.async_post("device/control/setSimpleRGBColorDimLevel", data)
return await rest_connection.async_post("device/control/setSimpleRGBColorDimLevel", data)


@Action.allowed_types("NOTIFICATION_LIGHT_CHANNEL")
async def action_set_rgb_dim_level_with_time(
runner: Runner,
rest_connection: RestConnection,
fc: FunctionalChannel,
rgb: RGBColorState,
dim_level: float,
Expand All @@ -212,100 +213,100 @@ async def action_set_rgb_dim_level_with_time(
"onTime": on_time,
"rampTime": ramp_time,
}
return await runner.rest_connection.async_post("device/control/setSimpleRGBColorDimLevelWithTime", data)
return await rest_connection.async_post("device/control/setSimpleRGBColorDimLevelWithTime", data)


@Action.allowed_types("SHADING_CHANNEL")
async def action_set_primary_shading_level(runner: Runner, fc: FunctionalChannel, primary_shading_level: float):
async def action_set_primary_shading_level(rest_connection: RestConnection, fc: FunctionalChannel, primary_shading_level: float):
data = {
"channelIndex": fc.index,
"deviceId": fc.deviceId,
"primaryShadingLevel": primary_shading_level,
}
return await runner.rest_connection.async_post("device/control/setPrimaryShadingLevel", data)
return await rest_connection.async_post("device/control/setPrimaryShadingLevel", data)


@Action.allowed_types("SHADING_CHANNEL")
async def action_set_secondary_shading_level(runner: Runner, fc: FunctionalChannel, primary_shading_level: float,
async def action_set_secondary_shading_level(rest_connection: RestConnection, fc: FunctionalChannel, primary_shading_level: float,
secondary_shading_level: float):
data = {
"channelIndex": fc.index,
"deviceId": fc.deviceId,
"primaryShadingLevel": primary_shading_level,
"secondaryShadingLevel": secondary_shading_level,
}
return await runner.rest_connection.async_post("device/control/setSecondaryShadingLevel", data)
return await rest_connection.async_post("device/control/setSecondaryShadingLevel", data)


@Action.allowed_types("SWITCH_MEASURING_CHANNEL")
async def action_reset_energy_counter(runner: Runner, fc: FunctionalChannel):
async def action_reset_energy_counter(rest_connection: RestConnection, fc: FunctionalChannel):
data = {
"channelIndex": fc.index,
"deviceId": fc.deviceId
}
return await runner.rest_connection.async_post("device/control/resetEnergyCounter", data)
return await rest_connection.async_post("device/control/resetEnergyCounter", data)


@Action.allowed_types("WALL_MOUNTED_THERMOSTAT_PRO_CHANNEL")
async def action_set_display(runner: Runner, fc: FunctionalChannel, display: ClimateControlDisplay):
async def action_set_display(rest_connection: RestConnection, fc: FunctionalChannel, display: ClimateControlDisplay):
data = {
"channelIndex": fc.index,
"deviceId": fc.deviceId,
"display": str(display),
}
return await runner.rest_connection.async_post("device/configuration/setClimateControlDisplay", data)
return await rest_connection.async_post("device/configuration/setClimateControlDisplay", data)


@Action.allowed_types("WATER_SENSOR_CHANNEL")
async def action_set_acoustic_alarm_signal(runner: Runner, fc: FunctionalChannel,
async def action_set_acoustic_alarm_signal(rest_connection: RestConnection, fc: FunctionalChannel,
acoustic_alarm_signal: AcousticAlarmSignal):
data = {
"channelIndex": fc.index,
"deviceId": fc.deviceId,
"acousticAlarmSignal": str(acoustic_alarm_signal),
}
return await runner.rest_connection.async_post("device/configuration/setAcousticAlarmSignal", data)
return await rest_connection.async_post("device/configuration/setAcousticAlarmSignal", data)


@Action.allowed_types("WATER_SENSOR_CHANNEL")
async def action_set_acoustic_alarm_timing(runner: Runner, fc: FunctionalChannel,
async def action_set_acoustic_alarm_timing(rest_connection: RestConnection, fc: FunctionalChannel,
acoustic_alarm_timing: AcousticAlarmTiming):
data = {
"channelIndex": fc.index,
"deviceId": fc.deviceId,
"acousticAlarmTiming": str(acoustic_alarm_timing),
}
return await runner.rest_connection.async_post("device/configuration/setAcousticAlarmTiming", data)
return await rest_connection.async_post("device/configuration/setAcousticAlarmTiming", data)


@Action.allowed_types("WATER_SENSOR_CHANNEL")
async def action_set_acoustic_water_alarm_trigger(runner: Runner, fc: FunctionalChannel,
async def action_set_acoustic_water_alarm_trigger(rest_connection: RestConnection, fc: FunctionalChannel,
acoustic_water_alarm_trigger: WaterAlarmTrigger):
data = {
"channelIndex": fc.index,
"deviceId": fc.deviceId,
"acousticWaterAlarmTrigger": str(acoustic_water_alarm_trigger),
}
return await runner.rest_connection.async_post("device/configuration/setAcousticWaterAlarmTrigger", data)
return await rest_connection.async_post("device/configuration/setAcousticWaterAlarmTrigger", data)


@Action.allowed_types("WATER_SENSOR_CHANNEL")
async def action_set_inapp_water_alarm_trigger(runner: Runner, fc: FunctionalChannel,
async def action_set_inapp_water_alarm_trigger(rest_connection: RestConnection, fc: FunctionalChannel,
inapp_water_alarm_trigger: WaterAlarmTrigger):
data = {
"channelIndex": fc.index,
"deviceId": fc.deviceId,
"inAppWaterAlarmTrigger": str(inapp_water_alarm_trigger),
}
return await runner.rest_connection.async_post("device/configuration/setInAppWaterAlarmTrigger", data)
return await rest_connection.async_post("device/configuration/setInAppWaterAlarmTrigger", data)


@Action.allowed_types("WATER_SENSOR_CHANNEL")
async def action_set_siren_water_alarm_trigger(runner: Runner, fc: FunctionalChannel,
async def action_set_siren_water_alarm_trigger(rest_connection: RestConnection, fc: FunctionalChannel,
siren_water_alarm_trigger: WaterAlarmTrigger):
data = {
"channelIndex": fc.index,
"deviceId": fc.deviceId,
"sirenWaterAlarmTrigger": str(siren_water_alarm_trigger),
}
return await runner.rest_connection.async_post("device/configuration/setSirenWaterAlarmTrigger", data)
return await rest_connection.async_post("device/configuration/setSirenWaterAlarmTrigger", data)
Loading

0 comments on commit db1f276

Please sign in to comment.