Improve lists in integrations [R-S] (#113233)

* Improve lists in integrations [R-S]

* Fix

* Fix
This commit is contained in:
Joost Lekkerkerker 2024-03-13 21:55:00 +01:00 committed by GitHub
parent e6a692f354
commit 77917506bb
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
60 changed files with 543 additions and 611 deletions

View File

@ -245,10 +245,11 @@ class RachioIro:
_deinit_webhooks(None) _deinit_webhooks(None)
# Choose which events to listen for and get their IDs # Choose which events to listen for and get their IDs
event_types = [] event_types = [
for event_type in self.rachio.notification.get_webhook_event_type()[1]: {"id": event_type[KEY_ID]}
if event_type[KEY_NAME] in LISTEN_EVENT_TYPES: for event_type in self.rachio.notification.get_webhook_event_type()[1]
event_types.append({"id": event_type[KEY_ID]}) if event_type[KEY_NAME] in LISTEN_EVENT_TYPES
]
# Register to listen to these events from the device # Register to listen to these events from the device
url = self.rachio.webhook_url url = self.rachio.webhook_url

View File

@ -169,10 +169,13 @@ def _create_entities(hass: HomeAssistant, config_entry: ConfigEntry) -> list[Ent
schedules = controller.list_schedules() schedules = controller.list_schedules()
flex_schedules = controller.list_flex_schedules() flex_schedules = controller.list_flex_schedules()
current_schedule = controller.current_schedule current_schedule = controller.current_schedule
for zone in zones: entities.extend(
entities.append(RachioZone(person, controller, zone, current_schedule)) RachioZone(person, controller, zone, current_schedule) for zone in zones
for sched in schedules + flex_schedules: )
entities.append(RachioSchedule(person, controller, sched, current_schedule)) entities.extend(
RachioSchedule(person, controller, schedule, current_schedule)
for schedule in schedules + flex_schedules
)
return entities return entities

View File

@ -45,8 +45,10 @@ def setup_platform(
else: else:
# create a sensor for each zone managed by faucet # create a sensor for each zone managed by faucet
for zone in raincloud.controller.faucet.zones: sensors.extend(
sensors.append(RainCloudBinarySensor(zone, sensor_type)) RainCloudBinarySensor(zone, sensor_type)
for zone in raincloud.controller.faucet.zones
)
add_entities(sensors, True) add_entities(sensors, True)

View File

@ -48,8 +48,10 @@ def setup_platform(
sensors.append(RainCloudSensor(raincloud.controller.faucet, sensor_type)) sensors.append(RainCloudSensor(raincloud.controller.faucet, sensor_type))
else: else:
# create a sensor for each zone managed by a faucet # create a sensor for each zone managed by a faucet
for zone in raincloud.controller.faucet.zones: sensors.extend(
sensors.append(RainCloudSensor(zone, sensor_type)) RainCloudSensor(zone, sensor_type)
for zone in raincloud.controller.faucet.zones
)
add_entities(sensors, True) add_entities(sensors, True)

View File

@ -47,13 +47,14 @@ def setup_platform(
raincloud = hass.data[DATA_RAINCLOUD].data raincloud = hass.data[DATA_RAINCLOUD].data
default_watering_timer = config[CONF_WATERING_TIME] default_watering_timer = config[CONF_WATERING_TIME]
sensors = [] add_entities(
for sensor_type in config[CONF_MONITORED_CONDITIONS]: (
# create a sensor for each zone managed by faucet RainCloudSwitch(default_watering_timer, zone, sensor_type)
for zone in raincloud.controller.faucet.zones: for zone in raincloud.controller.faucet.zones
sensors.append(RainCloudSwitch(default_watering_timer, zone, sensor_type)) for sensor_type in config[CONF_MONITORED_CONDITIONS]
),
add_entities(sensors, True) True,
)
class RainCloudSwitch(RainCloudEntity, SwitchEntity): class RainCloudSwitch(RainCloudEntity, SwitchEntity):

View File

@ -512,21 +512,20 @@ def _update_states_table_with_foreign_key_options(
) -> None: ) -> None:
"""Add the options to foreign key constraints.""" """Add the options to foreign key constraints."""
inspector = sqlalchemy.inspect(engine) inspector = sqlalchemy.inspect(engine)
alters = [] alters = [
for foreign_key in inspector.get_foreign_keys(TABLE_STATES): {
if foreign_key["name"] and ( "old_fk": ForeignKeyConstraint((), (), name=foreign_key["name"]),
"columns": foreign_key["constrained_columns"],
}
for foreign_key in inspector.get_foreign_keys(TABLE_STATES)
if foreign_key["name"]
and (
# MySQL/MariaDB will have empty options # MySQL/MariaDB will have empty options
not foreign_key.get("options") not foreign_key.get("options")
or
# Postgres will have ondelete set to None # Postgres will have ondelete set to None
foreign_key.get("options", {}).get("ondelete") is None or foreign_key.get("options", {}).get("ondelete") is None
): )
alters.append( ]
{
"old_fk": ForeignKeyConstraint((), (), name=foreign_key["name"]),
"columns": foreign_key["constrained_columns"],
}
)
if not alters: if not alters:
return return

View File

@ -243,7 +243,6 @@ class ReolinkVODMediaSource(MediaSource):
start = now - dt.timedelta(days=31) start = now - dt.timedelta(days=31)
end = now end = now
children: list[BrowseMediaSource] = []
if _LOGGER.isEnabledFor(logging.DEBUG): if _LOGGER.isEnabledFor(logging.DEBUG):
_LOGGER.debug( _LOGGER.debug(
"Requesting recording days of %s from %s to %s", "Requesting recording days of %s from %s to %s",
@ -254,19 +253,19 @@ class ReolinkVODMediaSource(MediaSource):
statuses, _ = await host.api.request_vod_files( statuses, _ = await host.api.request_vod_files(
channel, start, end, status_only=True, stream=stream channel, start, end, status_only=True, stream=stream
) )
for status in statuses: children: list[BrowseMediaSource] = [
for day in status.days: BrowseMediaSource(
children.append( domain=DOMAIN,
BrowseMediaSource( identifier=f"DAY|{config_entry_id}|{channel}|{stream}|{status.year}|{status.month}|{day}",
domain=DOMAIN, media_class=MediaClass.DIRECTORY,
identifier=f"DAY|{config_entry_id}|{channel}|{stream}|{status.year}|{status.month}|{day}", media_content_type=MediaType.PLAYLIST,
media_class=MediaClass.DIRECTORY, title=f"{status.year}/{status.month}/{day}",
media_content_type=MediaType.PLAYLIST, can_play=False,
title=f"{status.year}/{status.month}/{day}", can_expand=True,
can_play=False, )
can_expand=True, for status in statuses
) for day in status.days
) ]
return BrowseMediaSource( return BrowseMediaSource(
domain=DOMAIN, domain=DOMAIN,

View File

@ -51,21 +51,17 @@ async def async_get_actions(
except ValueError: except ValueError:
return [] return []
actions = [] return [
for action_type in ACTION_TYPES: {
if hasattr(device, action_type): CONF_DEVICE_ID: device_id,
data: dict[int, str] = getattr(device, ACTION_SELECTION[action_type], {}) CONF_DOMAIN: DOMAIN,
for value in data.values(): CONF_TYPE: action_type,
actions.append( CONF_SUBTYPE: value,
{ }
CONF_DEVICE_ID: device_id, for action_type in ACTION_TYPES
CONF_DOMAIN: DOMAIN, if hasattr(device, action_type)
CONF_TYPE: action_type, for value in getattr(device, ACTION_SELECTION[action_type], {}).values()
CONF_SUBTYPE: value, ]
}
)
return actions
def _get_commands( def _get_commands(

View File

@ -51,20 +51,17 @@ async def async_get_triggers(
"""List device triggers for RFXCOM RFXtrx devices.""" """List device triggers for RFXCOM RFXtrx devices."""
device = async_get_device_object(hass, device_id) device = async_get_device_object(hass, device_id)
triggers = [] return [
for conf_type in TRIGGER_TYPES: {
data: dict[int, str] = getattr(device, TRIGGER_SELECTION[conf_type], {}) CONF_PLATFORM: "device",
for command in data.values(): CONF_DEVICE_ID: device_id,
triggers.append( CONF_DOMAIN: DOMAIN,
{ CONF_TYPE: conf_type,
CONF_PLATFORM: "device", CONF_SUBTYPE: command,
CONF_DEVICE_ID: device_id, }
CONF_DOMAIN: DOMAIN, for conf_type in TRIGGER_TYPES
CONF_TYPE: conf_type, for command in getattr(device, TRIGGER_SELECTION[conf_type], {}).values()
CONF_SUBTYPE: command, ]
}
)
return triggers
async def async_validate_trigger_config( async def async_validate_trigger_config(

View File

@ -255,18 +255,15 @@ async def async_setup_entry(
device_id: DeviceTuple, device_id: DeviceTuple,
entity_info: dict[str, Any], entity_info: dict[str, Any],
) -> list[Entity]: ) -> list[Entity]:
entities: list[Entity] = [] return [
for data_type in set(event.values) & set(SENSOR_TYPES_DICT): RfxtrxSensor(
entities.append( event.device,
RfxtrxSensor( device_id,
event.device, SENSOR_TYPES_DICT[data_type],
device_id, event=event if auto else None,
SENSOR_TYPES_DICT[data_type],
event=event if auto else None,
)
) )
for data_type in set(event.values) & set(SENSOR_TYPES_DICT)
return entities ]
await async_setup_platform_entry( await async_setup_platform_entry(
hass, config_entry, async_add_entities, _supported, _constructor hass, config_entry, async_add_entities, _supported, _constructor

View File

@ -34,10 +34,11 @@ async def async_get_config_entry_diagnostics(
) -> dict[str, Any]: ) -> dict[str, Any]:
"""Return diagnostics for a config entry.""" """Return diagnostics for a config entry."""
ring: ring_doorbell.Ring = hass.data[DOMAIN][entry.entry_id]["api"] ring: ring_doorbell.Ring = hass.data[DOMAIN][entry.entry_id]["api"]
devices_raw = [] devices_raw = [
for device_type in ring.devices_data: ring.devices_data[device_type][device_id]
for device_id in ring.devices_data[device_type]: for device_type in ring.devices_data
devices_raw.append(ring.devices_data[device_type][device_id]) for device_id in ring.devices_data[device_type]
]
return async_redact_data( return async_redact_data(
{"device_data": devices_raw}, {"device_data": devices_raw},
TO_REDACT, TO_REDACT,

View File

@ -41,13 +41,12 @@ async def async_setup_entry(
devices_coordinator: RingDataCoordinator = hass.data[DOMAIN][config_entry.entry_id][ devices_coordinator: RingDataCoordinator = hass.data[DOMAIN][config_entry.entry_id][
RING_DEVICES_COORDINATOR RING_DEVICES_COORDINATOR
] ]
lights = []
for device in devices["stickup_cams"]: async_add_entities(
if device.has_capability("light"): RingLight(device, devices_coordinator)
lights.append(RingLight(device, devices_coordinator)) for device in devices["stickup_cams"]
if device.has_capability("light")
async_add_entities(lights) )
class RingLight(RingEntity, LightEntity): class RingLight(RingEntity, LightEntity):

View File

@ -28,12 +28,10 @@ async def async_setup_entry(
coordinator: RingDataCoordinator = hass.data[DOMAIN][config_entry.entry_id][ coordinator: RingDataCoordinator = hass.data[DOMAIN][config_entry.entry_id][
RING_DEVICES_COORDINATOR RING_DEVICES_COORDINATOR
] ]
sirens = []
for device in devices["chimes"]: async_add_entities(
sirens.append(RingChimeSiren(device, coordinator)) RingChimeSiren(device, coordinator) for device in devices["chimes"]
)
async_add_entities(sirens)
class RingChimeSiren(RingEntity, SirenEntity): class RingChimeSiren(RingEntity, SirenEntity):

View File

@ -38,13 +38,12 @@ async def async_setup_entry(
coordinator: RingDataCoordinator = hass.data[DOMAIN][config_entry.entry_id][ coordinator: RingDataCoordinator = hass.data[DOMAIN][config_entry.entry_id][
RING_DEVICES_COORDINATOR RING_DEVICES_COORDINATOR
] ]
switches = []
for device in devices["stickup_cams"]: async_add_entities(
if device.has_capability("siren"): SirenSwitch(device, coordinator)
switches.append(SirenSwitch(device, coordinator)) for device in devices["stickup_cams"]
if device.has_capability("siren")
async_add_entities(switches) )
class BaseRingSwitch(RingEntity, SwitchEntity): class BaseRingSwitch(RingEntity, SwitchEntity):

View File

@ -87,12 +87,10 @@ def build_setup_functions(
product_info: dict[str, HomeDataProduct], product_info: dict[str, HomeDataProduct],
) -> list[Coroutine[Any, Any, RoborockDataUpdateCoordinator | None]]: ) -> list[Coroutine[Any, Any, RoborockDataUpdateCoordinator | None]]:
"""Create a list of setup functions that can later be called asynchronously.""" """Create a list of setup functions that can later be called asynchronously."""
setup_functions = [] return [
for device in device_map.values(): setup_device(hass, user_data, device, product_info[device.product_id])
setup_functions.append( for device in device_map.values()
setup_device(hass, user_data, device, product_info[device.product_id]) ]
)
return setup_functions
async def setup_device( async def setup_device(

View File

@ -115,15 +115,13 @@ async def async_setup_entry(
coordinator: RokuDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id] coordinator: RokuDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
device: RokuDevice = coordinator.data device: RokuDevice = coordinator.data
entities: list[RokuSelectEntity] = [] entities: list[RokuSelectEntity] = [
RokuSelectEntity(
for description in ENTITIES: coordinator=coordinator,
entities.append( description=description,
RokuSelectEntity(
coordinator=coordinator,
description=description,
)
) )
for description in ENTITIES
]
if len(device.channels) > 0: if len(device.channels) > 0:
entities.append( entities.append(

View File

@ -37,11 +37,11 @@ class BraavaJet(IRobotVacuum):
super().__init__(roomba, blid) super().__init__(roomba, blid)
# Initialize fan speed list # Initialize fan speed list
speed_list = [] self._attr_fan_speed_list = [
for behavior in BRAAVA_MOP_BEHAVIORS: f"{behavior}-{spray}"
for spray in BRAAVA_SPRAY_AMOUNT: for behavior in BRAAVA_MOP_BEHAVIORS
speed_list.append(f"{behavior}-{spray}") for spray in BRAAVA_SPRAY_AMOUNT
self._attr_fan_speed_list = speed_list ]
@property @property
def fan_speed(self): def fan_speed(self):

View File

@ -58,9 +58,7 @@ def setup_platform(
russ = russound.Russound(host, port) russ = russound.Russound(host, port)
russ.connect() russ.connect()
sources = [] sources = [source["name"] for source in config[CONF_SOURCES]]
for source in config[CONF_SOURCES]:
sources.append(source["name"])
if russ.is_connected(): if russ.is_connected():
for zone_id, extra in config[CONF_ZONES].items(): for zone_id, extra in config[CONF_ZONES].items():

View File

@ -79,7 +79,7 @@ async def async_setup_platform(
sensor_def = pysaj.Sensors(wifi) sensor_def = pysaj.Sensors(wifi)
# Use all sensors by default # Use all sensors by default
hass_sensors = [] hass_sensors: list[SAJsensor] = []
kwargs = {} kwargs = {}
if wifi: if wifi:
@ -103,11 +103,11 @@ async def async_setup_platform(
if not done: if not done:
raise PlatformNotReady raise PlatformNotReady
for sensor in sensor_def: hass_sensors.extend(
if sensor.enabled: SAJsensor(saj.serialnumber, sensor, inverter_name=config.get(CONF_NAME))
hass_sensors.append( for sensor in sensor_def
SAJsensor(saj.serialnumber, sensor, inverter_name=config.get(CONF_NAME)) if sensor.enabled
) )
async_add_entities(hass_sensors) async_add_entities(hass_sensors)

View File

@ -45,17 +45,15 @@ async def async_setup_entry(
) -> None: ) -> None:
"""Set up binary_sensors based on a config entry.""" """Set up binary_sensors based on a config entry."""
coordinator: SchlageDataUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id] coordinator: SchlageDataUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id]
entities = [] async_add_entities(
for device_id in coordinator.data.locks: SchlageBinarySensor(
for description in _DESCRIPTIONS: coordinator=coordinator,
entities.append( description=description,
SchlageBinarySensor( device_id=device_id,
coordinator=coordinator, )
description=description, for device_id in coordinator.data.locks
device_id=device_id, for description in _DESCRIPTIONS
) )
)
async_add_entities(entities)
class SchlageBinarySensor(SchlageEntity, BinarySensorEntity): class SchlageBinarySensor(SchlageEntity, BinarySensorEntity):

View File

@ -62,17 +62,15 @@ async def async_setup_entry(
) -> None: ) -> None:
"""Set up switches based on a config entry.""" """Set up switches based on a config entry."""
coordinator: SchlageDataUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id] coordinator: SchlageDataUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id]
entities = [] async_add_entities(
for device_id in coordinator.data.locks: SchlageSwitch(
for description in SWITCHES: coordinator=coordinator,
entities.append( description=description,
SchlageSwitch( device_id=device_id,
coordinator=coordinator, )
description=description, for device_id in coordinator.data.locks
device_id=device_id, for description in SWITCHES
) )
)
async_add_entities(entities)
class SchlageSwitch(SchlageEntity, SwitchEntity): class SchlageSwitch(SchlageEntity, SwitchEntity):

View File

@ -175,32 +175,31 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback, async_add_entities: AddEntitiesCallback,
) -> None: ) -> None:
"""Set up entry.""" """Set up entry."""
entities: list[ScreenLogicBinarySensor] = []
coordinator: ScreenlogicDataUpdateCoordinator = hass.data[SL_DOMAIN][ coordinator: ScreenlogicDataUpdateCoordinator = hass.data[SL_DOMAIN][
config_entry.entry_id config_entry.entry_id
] ]
gateway = coordinator.gateway gateway = coordinator.gateway
for core_sensor_description in SUPPORTED_CORE_SENSORS: entities: list[ScreenLogicBinarySensor] = [
ScreenLogicPushBinarySensor(coordinator, core_sensor_description)
for core_sensor_description in SUPPORTED_CORE_SENSORS
if ( if (
gateway.get_data( gateway.get_data(
*core_sensor_description.data_root, core_sensor_description.key *core_sensor_description.data_root, core_sensor_description.key
) )
is not None is not None
): )
entities.append( ]
ScreenLogicPushBinarySensor(coordinator, core_sensor_description)
)
for p_index, p_data in gateway.get_data(DEVICE.PUMP).items(): for p_index, p_data in gateway.get_data(DEVICE.PUMP).items():
if not p_data or not p_data.get(VALUE.DATA): if not p_data or not p_data.get(VALUE.DATA):
continue continue
for proto_pump_sensor_description in SUPPORTED_PUMP_SENSORS: entities.extend(
entities.append( ScreenLogicPumpBinarySensor(
ScreenLogicPumpBinarySensor( coordinator, copy(proto_pump_sensor_description), p_index
coordinator, copy(proto_pump_sensor_description), p_index
)
) )
for proto_pump_sensor_description in SUPPORTED_PUMP_SENSORS
)
chem_sensor_description: ScreenLogicPushBinarySensorDescription chem_sensor_description: ScreenLogicPushBinarySensorDescription
for chem_sensor_description in SUPPORTED_INTELLICHEM_SENSORS: for chem_sensor_description in SUPPORTED_INTELLICHEM_SENSORS:

View File

@ -47,26 +47,23 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback, async_add_entities: AddEntitiesCallback,
) -> None: ) -> None:
"""Set up entry.""" """Set up entry."""
entities = []
coordinator: ScreenlogicDataUpdateCoordinator = hass.data[SL_DOMAIN][ coordinator: ScreenlogicDataUpdateCoordinator = hass.data[SL_DOMAIN][
config_entry.entry_id config_entry.entry_id
] ]
gateway = coordinator.gateway gateway = coordinator.gateway
for body_index in gateway.get_data(DEVICE.BODY): async_add_entities(
entities.append( ScreenLogicClimate(
ScreenLogicClimate( coordinator,
coordinator, ScreenLogicClimateDescription(
ScreenLogicClimateDescription( subscription_code=CODE.STATUS_CHANGED,
subscription_code=CODE.STATUS_CHANGED, data_root=(DEVICE.BODY,),
data_root=(DEVICE.BODY,), key=body_index,
key=body_index, ),
),
)
) )
for body_index in gateway.get_data(DEVICE.BODY)
async_add_entities(entities) )
@dataclass(frozen=True, kw_only=True) @dataclass(frozen=True, kw_only=True)

View File

@ -227,20 +227,21 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback, async_add_entities: AddEntitiesCallback,
) -> None: ) -> None:
"""Set up entry.""" """Set up entry."""
entities: list[ScreenLogicSensor] = []
coordinator: ScreenlogicDataUpdateCoordinator = hass.data[SL_DOMAIN][ coordinator: ScreenlogicDataUpdateCoordinator = hass.data[SL_DOMAIN][
config_entry.entry_id config_entry.entry_id
] ]
gateway = coordinator.gateway gateway = coordinator.gateway
for core_sensor_description in SUPPORTED_CORE_SENSORS: entities: list[ScreenLogicSensor] = [
ScreenLogicPushSensor(coordinator, core_sensor_description)
for core_sensor_description in SUPPORTED_CORE_SENSORS
if ( if (
gateway.get_data( gateway.get_data(
*core_sensor_description.data_root, core_sensor_description.key *core_sensor_description.data_root, core_sensor_description.key
) )
is not None is not None
): )
entities.append(ScreenLogicPushSensor(coordinator, core_sensor_description)) ]
for pump_index, pump_data in gateway.get_data(DEVICE.PUMP).items(): for pump_index, pump_data in gateway.get_data(DEVICE.PUMP).items():
if not pump_data or not pump_data.get(VALUE.DATA): if not pump_data or not pump_data.get(VALUE.DATA):

View File

@ -127,8 +127,10 @@ async def async_setup_entry(
) )
) )
for i in range(len(data.active_voltage)): entities.extend(
entities.append(SenseVoltageSensor(data, i, sense_monitor_id)) SenseVoltageSensor(data, i, sense_monitor_id)
for i in range(len(data.active_voltage))
)
for type_id, typ in TRENDS_SENSOR_TYPES.items(): for type_id, typ in TRENDS_SENSOR_TYPES.items():
for variant_id, variant_name in TREND_SENSOR_VARIANTS: for variant_id, variant_name in TREND_SENSOR_VARIANTS:

View File

@ -230,11 +230,9 @@ class SensiboClimate(SensiboDeviceBaseEntity, ClimateEntity):
@property @property
def hvac_modes(self) -> list[HVACMode]: def hvac_modes(self) -> list[HVACMode]:
"""Return the list of available hvac operation modes.""" """Return the list of available hvac operation modes."""
hvac_modes = []
if TYPE_CHECKING: if TYPE_CHECKING:
assert self.device_data.hvac_modes assert self.device_data.hvac_modes
for mode in self.device_data.hvac_modes: hvac_modes = [SENSIBO_TO_HA[mode] for mode in self.device_data.hvac_modes]
hvac_modes.append(SENSIBO_TO_HA[mode])
return hvac_modes if hvac_modes else [HVACMode.OFF] return hvac_modes if hvac_modes else [HVACMode.OFF]
@property @property

View File

@ -57,15 +57,12 @@ async def async_setup_platform(
discovery_info: DiscoveryInfoType | None = None, discovery_info: DiscoveryInfoType | None = None,
) -> None: ) -> None:
"""Set up the Seven segments OCR platform.""" """Set up the Seven segments OCR platform."""
entities = [] async_add_entities(
for camera in config[CONF_SOURCE]: ImageProcessingSsocr(
entities.append( hass, camera[CONF_ENTITY_ID], config, camera.get(CONF_NAME)
ImageProcessingSsocr(
hass, camera[CONF_ENTITY_ID], config, camera.get(CONF_NAME)
)
) )
for camera in config[CONF_SOURCE]
async_add_entities(entities) )
class ImageProcessingSsocr(ImageProcessingEntity): class ImageProcessingSsocr(ImageProcessingEntity):

View File

@ -188,8 +188,6 @@ def get_block_input_triggers(
if not is_block_momentary_input(device.settings, block, True): if not is_block_momentary_input(device.settings, block, True):
return [] return []
triggers = []
if block.type == "device" or get_number_of_channels(device, block) == 1: if block.type == "device" or get_number_of_channels(device, block) == 1:
subtype = "button" subtype = "button"
else: else:
@ -203,20 +201,12 @@ def get_block_input_triggers(
else: else:
trigger_types = BASIC_INPUTS_EVENTS_TYPES trigger_types = BASIC_INPUTS_EVENTS_TYPES
for trigger_type in trigger_types: return [(trigger_type, subtype) for trigger_type in trigger_types]
triggers.append((trigger_type, subtype))
return triggers
def get_shbtn_input_triggers() -> list[tuple[str, str]]: def get_shbtn_input_triggers() -> list[tuple[str, str]]:
"""Return list of input triggers for SHBTN models.""" """Return list of input triggers for SHBTN models."""
triggers = [] return [(trigger_type, "button") for trigger_type in SHBTN_INPUTS_EVENTS_TYPES]
for trigger_type in SHBTN_INPUTS_EVENTS_TYPES:
triggers.append((trigger_type, "button"))
return triggers
@singleton.singleton("shelly_coap") @singleton.singleton("shelly_coap")

View File

@ -52,10 +52,7 @@ def setup_platform(
auth = sigfox.auth auth = sigfox.auth
devices = sigfox.devices devices = sigfox.devices
sensors = [] add_entities((SigfoxDevice(device, auth, name) for device in devices), True)
for device in devices:
sensors.append(SigfoxDevice(device, auth, name))
add_entities(sensors, True)
def epoch_to_datetime(epoch_time): def epoch_to_datetime(epoch_time):
@ -105,8 +102,7 @@ class SigfoxAPI:
url = urljoin(API_URL, location_url) url = urljoin(API_URL, location_url)
response = requests.get(url, auth=self._auth, timeout=10) response = requests.get(url, auth=self._auth, timeout=10)
devices_data = json.loads(response.text)["data"] devices_data = json.loads(response.text)["data"]
for device in devices_data: devices.extend(device["id"] for device in devices_data)
devices.append(device["id"])
return devices return devices
@property @property

View File

@ -79,8 +79,10 @@ async def async_setup_entry(
if sensor.type in SUPPORTED_BATTERY_SENSOR_TYPES: if sensor.type in SUPPORTED_BATTERY_SENSOR_TYPES:
sensors.append(BatteryBinarySensor(simplisafe, system, sensor)) sensors.append(BatteryBinarySensor(simplisafe, system, sensor))
for lock in system.locks.values(): sensors.extend(
sensors.append(BatteryBinarySensor(simplisafe, system, lock)) BatteryBinarySensor(simplisafe, system, lock)
for lock in system.locks.values()
)
async_add_entities(sensors) async_add_entities(sensors)

View File

@ -34,15 +34,16 @@ async def async_setup_entry(
) -> None: ) -> None:
"""Set up SimpliSafe locks based on a config entry.""" """Set up SimpliSafe locks based on a config entry."""
simplisafe = hass.data[DOMAIN][entry.entry_id] simplisafe = hass.data[DOMAIN][entry.entry_id]
locks = [] locks: list[SimpliSafeLock] = []
for system in simplisafe.systems.values(): for system in simplisafe.systems.values():
if system.version == 2: if system.version == 2:
LOGGER.info("Skipping lock setup for V2 system: %s", system.system_id) LOGGER.info("Skipping lock setup for V2 system: %s", system.system_id)
continue continue
for lock in system.locks.values(): locks.extend(
locks.append(SimpliSafeLock(simplisafe, system, lock)) SimpliSafeLock(simplisafe, system, lock) for lock in system.locks.values()
)
async_add_entities(locks) async_add_entities(locks)

View File

@ -25,16 +25,18 @@ async def async_setup_entry(
) -> None: ) -> None:
"""Set up SimpliSafe freeze sensors based on a config entry.""" """Set up SimpliSafe freeze sensors based on a config entry."""
simplisafe = hass.data[DOMAIN][entry.entry_id] simplisafe = hass.data[DOMAIN][entry.entry_id]
sensors = [] sensors: list[SimplisafeFreezeSensor] = []
for system in simplisafe.systems.values(): for system in simplisafe.systems.values():
if system.version == 2: if system.version == 2:
LOGGER.info("Skipping sensor setup for V2 system: %s", system.system_id) LOGGER.info("Skipping sensor setup for V2 system: %s", system.system_id)
continue continue
for sensor in system.sensors.values(): sensors.extend(
if sensor.type == DeviceTypes.TEMPERATURE: SimplisafeFreezeSensor(simplisafe, system, sensor)
sensors.append(SimplisafeFreezeSensor(simplisafe, system, sensor)) for sensor in system.sensors.values()
if sensor.type == DeviceTypes.TEMPERATURE
)
async_add_entities(sensors) async_add_entities(sensors)

View File

@ -143,35 +143,35 @@ async def async_setup_entry(
"""Set up the SleepIQ bed sensors.""" """Set up the SleepIQ bed sensors."""
data: SleepIQData = hass.data[DOMAIN][entry.entry_id] data: SleepIQData = hass.data[DOMAIN][entry.entry_id]
entities = [] entities: list[SleepIQNumberEntity] = []
for bed in data.client.beds.values(): for bed in data.client.beds.values():
for sleeper in bed.sleepers: entities.extend(
entities.append( SleepIQNumberEntity(
SleepIQNumberEntity( data.data_coordinator,
data.data_coordinator, bed,
bed, sleeper,
sleeper, NUMBER_DESCRIPTIONS[FIRMNESS],
NUMBER_DESCRIPTIONS[FIRMNESS],
)
) )
for actuator in bed.foundation.actuators: for sleeper in bed.sleepers
entities.append( )
SleepIQNumberEntity( entities.extend(
data.data_coordinator, SleepIQNumberEntity(
bed, data.data_coordinator,
actuator, bed,
NUMBER_DESCRIPTIONS[ACTUATOR], actuator,
) NUMBER_DESCRIPTIONS[ACTUATOR],
) )
for foot_warmer in bed.foundation.foot_warmers: for actuator in bed.foundation.actuators
entities.append( )
SleepIQNumberEntity( entities.extend(
data.data_coordinator, SleepIQNumberEntity(
bed, data.data_coordinator,
foot_warmer, bed,
NUMBER_DESCRIPTIONS[FOOT_WARMING_TIMER], foot_warmer,
) NUMBER_DESCRIPTIONS[FOOT_WARMING_TIMER],
) )
for foot_warmer in bed.foundation.foot_warmers
)
async_add_entities(entities) async_add_entities(entities)

View File

@ -844,19 +844,16 @@ async def async_setup_entry(
if TYPE_CHECKING: if TYPE_CHECKING:
assert config_entry.unique_id assert config_entry.unique_id
entities = [] async_add_entities(
for sensor in used_sensors: SMAsensor(
entities.append( coordinator,
SMAsensor( config_entry.unique_id,
coordinator, SENSOR_ENTITIES.get(sensor.name),
config_entry.unique_id, device_info,
SENSOR_ENTITIES.get(sensor.name), sensor,
device_info,
sensor,
)
) )
for sensor in used_sensors
async_add_entities(entities) )
class SMAsensor(CoordinatorEntity, SensorEntity): class SMAsensor(CoordinatorEntity, SensorEntity):

View File

@ -36,18 +36,18 @@ async def async_setup_entry(
) )
) )
elif actuator.type == "INFINITY_OUTPUT_MODULE": elif actuator.type == "INFINITY_OUTPUT_MODULE":
for option in actuator.state_options: entities.extend(
entities.append( SmappeeActuator(
SmappeeActuator( smappee_base,
smappee_base, service_location,
service_location, actuator.name,
actuator.name, actuator_id,
actuator_id, actuator.type,
actuator.type, actuator.serialnumber,
actuator.serialnumber, actuator_state_option=option,
actuator_state_option=option,
)
) )
for option in actuator.state_options
)
async_add_entities(entities, True) async_add_entities(entities, True)

View File

@ -59,9 +59,9 @@ def get_capabilities(capabilities: Sequence[str]) -> Sequence[str] | None:
supported = [Capability.switch] supported = [Capability.switch]
for capability in optional: supported.extend(
if capability in capabilities: capability for capability in optional if capability in capabilities
supported.append(capability) )
return supported return supported

View File

@ -85,15 +85,14 @@ async def async_setup_entry(
network_coordinator = sms_data[NETWORK_COORDINATOR] network_coordinator = sms_data[NETWORK_COORDINATOR]
gateway = sms_data[GATEWAY] gateway = sms_data[GATEWAY]
unique_id = str(await gateway.get_imei_async()) unique_id = str(await gateway.get_imei_async())
entities = [] entities = [
for description in SIGNAL_SENSORS: DeviceSensor(signal_coordinator, description, unique_id, gateway)
entities.append( for description in SIGNAL_SENSORS
DeviceSensor(signal_coordinator, description, unique_id, gateway) ]
) entities.extend(
for description in NETWORK_SENSORS: DeviceSensor(network_coordinator, description, unique_id, gateway)
entities.append( for description in NETWORK_SENSORS
DeviceSensor(network_coordinator, description, unique_id, gateway) )
)
async_add_entities(entities, True) async_add_entities(entities, True)

View File

@ -38,10 +38,9 @@ class SpeedTestDataCoordinator(DataUpdateCoordinator[dict[str, Any]]):
def update_servers(self) -> None: def update_servers(self) -> None:
"""Update list of test servers.""" """Update list of test servers."""
test_servers = self.api.get_servers() test_servers = self.api.get_servers()
test_servers_list = [] test_servers_list = [
for servers in test_servers.values(): server for servers in test_servers.values() for server in servers
for server in servers: ]
test_servers_list.append(server)
for server in sorted( for server in sorted(
test_servers_list, test_servers_list,
key=lambda server: ( key=lambda server: (

View File

@ -395,11 +395,11 @@ class SqueezeBoxEntity(MediaPlayerEntity):
player_ids = { player_ids = {
p.unique_id: p.entity_id for p in self.hass.data[DOMAIN][KNOWN_PLAYERS] p.unique_id: p.entity_id for p in self.hass.data[DOMAIN][KNOWN_PLAYERS]
} }
sync_group = [] return [
for player in self._player.sync_group: player_ids[player]
if player in player_ids: for player in self._player.sync_group
sync_group.append(player_ids[player]) if player in player_ids
return sync_group ]
@property @property
def sync_group(self): def sync_group(self):
@ -550,8 +550,7 @@ class SqueezeBoxEntity(MediaPlayerEntity):
""" """
all_params = [command] all_params = [command]
if parameters: if parameters:
for parameter in parameters: all_params.extend(parameters)
all_params.append(parameter)
await self._player.async_query(*all_params) await self._player.async_query(*all_params)
async def async_call_query(self, command, parameters=None): async def async_call_query(self, command, parameters=None):
@ -562,8 +561,7 @@ class SqueezeBoxEntity(MediaPlayerEntity):
""" """
all_params = [command] all_params = [command]
if parameters: if parameters:
for parameter in parameters: all_params.extend(parameters)
all_params.append(parameter)
self._query_result = await self._player.async_query(*all_params) self._query_result = await self._player.async_query(*all_params)
_LOGGER.debug("call_query got result %s", self._query_result) _LOGGER.debug("call_query got result %s", self._query_result)

View File

@ -24,12 +24,12 @@ async def async_setup_entry(
) -> None: ) -> None:
"""Set up the StarLine button.""" """Set up the StarLine button."""
account: StarlineAccount = hass.data[DOMAIN][entry.entry_id] account: StarlineAccount = hass.data[DOMAIN][entry.entry_id]
entities = [] async_add_entities(
for device in account.api.devices.values(): StarlineButton(account, device, description)
if device.support_state: for device in account.api.devices.values()
for description in BUTTON_TYPES: if device.support_state
entities.append(StarlineButton(account, device, description)) for description in BUTTON_TYPES
async_add_entities(entities) )
class StarlineButton(StarlineEntity, ButtonEntity): class StarlineButton(StarlineEntity, ButtonEntity):

View File

@ -16,11 +16,11 @@ async def async_setup_entry(
) -> None: ) -> None:
"""Set up StarLine entry.""" """Set up StarLine entry."""
account: StarlineAccount = hass.data[DOMAIN][entry.entry_id] account: StarlineAccount = hass.data[DOMAIN][entry.entry_id]
entities = [] async_add_entities(
for device in account.api.devices.values(): StarlineDeviceTracker(account, device)
if device.support_position: for device in account.api.devices.values()
entities.append(StarlineDeviceTracker(account, device)) if device.support_position
async_add_entities(entities) )
class StarlineDeviceTracker(StarlineEntity, TrackerEntity, RestoreEntity): class StarlineDeviceTracker(StarlineEntity, TrackerEntity, RestoreEntity):

View File

@ -54,18 +54,18 @@ def setup_platform(
) -> None: ) -> None:
"""Set up the Sterling Bank sensor platform.""" """Set up the Sterling Bank sensor platform."""
sensors = [] sensors: list[StarlingBalanceSensor] = []
for account in config[CONF_ACCOUNTS]: for account in config[CONF_ACCOUNTS]:
try: try:
starling_account = StarlingAccount( starling_account = StarlingAccount(
account[CONF_ACCESS_TOKEN], sandbox=account[CONF_SANDBOX] account[CONF_ACCESS_TOKEN], sandbox=account[CONF_SANDBOX]
) )
for balance_type in account[CONF_BALANCE_TYPES]: sensors.extend(
sensors.append( StarlingBalanceSensor(
StarlingBalanceSensor( starling_account, account[CONF_NAME], balance_type
starling_account, account[CONF_NAME], balance_type
)
) )
for balance_type in account[CONF_BALANCE_TYPES]
)
except requests.exceptions.HTTPError as error: except requests.exceptions.HTTPError as error:
_LOGGER.error( _LOGGER.error(
"Unable to set up Starling account '%s': %s", account[CONF_NAME], error "Unable to set up Starling account '%s': %s", account[CONF_NAME], error

View File

@ -36,11 +36,11 @@ async def async_setup_entry(
entry: dict = hass.data[DOMAIN][config_entry.entry_id] entry: dict = hass.data[DOMAIN][config_entry.entry_id]
coordinator: DataUpdateCoordinator = entry[ENTRY_COORDINATOR] coordinator: DataUpdateCoordinator = entry[ENTRY_COORDINATOR]
vehicle_info: dict = entry[ENTRY_VEHICLES] vehicle_info: dict = entry[ENTRY_VEHICLES]
entities: list[SubaruDeviceTracker] = [] async_add_entities(
for vehicle in vehicle_info.values(): SubaruDeviceTracker(vehicle, coordinator)
if vehicle[VEHICLE_HAS_REMOTE_SERVICE]: for vehicle in vehicle_info.values()
entities.append(SubaruDeviceTracker(vehicle, coordinator)) if vehicle[VEHICLE_HAS_REMOTE_SERVICE]
async_add_entities(entities) )
class SubaruDeviceTracker( class SubaruDeviceTracker(

View File

@ -23,25 +23,18 @@ async def async_setup_entry(
) -> None: ) -> None:
"""Set up Sure PetCare locks on a config entry.""" """Set up Sure PetCare locks on a config entry."""
entities: list[SurePetcareLock] = []
coordinator: SurePetcareDataCoordinator = hass.data[DOMAIN][entry.entry_id] coordinator: SurePetcareDataCoordinator = hass.data[DOMAIN][entry.entry_id]
for surepy_entity in coordinator.data.values(): async_add_entities(
if surepy_entity.type not in [ SurePetcareLock(surepy_entity.id, coordinator, lock_state)
EntityType.CAT_FLAP, for surepy_entity in coordinator.data.values()
EntityType.PET_FLAP, if surepy_entity.type in [EntityType.CAT_FLAP, EntityType.PET_FLAP]
]:
continue
for lock_state in ( for lock_state in (
LockState.LOCKED_IN, LockState.LOCKED_IN,
LockState.LOCKED_OUT, LockState.LOCKED_OUT,
LockState.LOCKED_ALL, LockState.LOCKED_ALL,
): )
entities.append(SurePetcareLock(surepy_entity.id, coordinator, lock_state)) )
async_add_entities(entities)
class SurePetcareLock(SurePetcareEntity, LockEntity): class SurePetcareLock(SurePetcareEntity, LockEntity):

View File

@ -73,12 +73,13 @@ def setup_platform(
_LOGGER.error("The station doesn't exists: %s", station) _LOGGER.error("The station doesn't exists: %s", station)
return return
entities = [] add_entities(
(
for condition in monitored_conditions: SwissHydrologicalDataSensor(hydro_data, station, condition)
entities.append(SwissHydrologicalDataSensor(hydro_data, station, condition)) for condition in monitored_conditions
),
add_entities(entities, True) True,
)
class SwissHydrologicalDataSensor(SensorEntity): class SwissHydrologicalDataSensor(SensorEntity):

View File

@ -62,15 +62,15 @@ async def async_setup_entry(
SyncThruMainSensor(coordinator, name), SyncThruMainSensor(coordinator, name),
SyncThruActiveAlertSensor(coordinator, name), SyncThruActiveAlertSensor(coordinator, name),
] ]
entities.extend(SyncThruTonerSensor(coordinator, name, key) for key in supp_toner)
for key in supp_toner: entities.extend(SyncThruDrumSensor(coordinator, name, key) for key in supp_drum)
entities.append(SyncThruTonerSensor(coordinator, name, key)) entities.extend(
for key in supp_drum: SyncThruInputTraySensor(coordinator, name, key) for key in supp_tray
entities.append(SyncThruDrumSensor(coordinator, name, key)) )
for key in supp_tray: entities.extend(
entities.append(SyncThruInputTraySensor(coordinator, name, key)) SyncThruOutputTraySensor(coordinator, name, int_key)
for int_key in supp_output_tray: for int_key in supp_output_tray
entities.append(SyncThruOutputTraySensor(coordinator, name, int_key)) )
async_add_entities(entities) async_add_entities(entities)

View File

@ -91,20 +91,18 @@ class SynologyPhotosMediaSource(MediaSource):
) -> list[BrowseMediaSource]: ) -> list[BrowseMediaSource]:
"""Handle browsing different diskstations.""" """Handle browsing different diskstations."""
if not item.identifier: if not item.identifier:
ret = [] return [
for entry in self.entries: BrowseMediaSource(
ret.append( domain=DOMAIN,
BrowseMediaSource( identifier=entry.unique_id,
domain=DOMAIN, media_class=MediaClass.DIRECTORY,
identifier=entry.unique_id, media_content_type=MediaClass.IMAGE,
media_class=MediaClass.DIRECTORY, title=f"{entry.title} - {entry.unique_id}",
media_content_type=MediaClass.IMAGE, can_play=False,
title=f"{entry.title} - {entry.unique_id}", can_expand=True,
can_play=False,
can_expand=True,
)
) )
return ret for entry in self.entries
]
identifier = SynologyPhotosMediaSourceIdentifier(item.identifier) identifier = SynologyPhotosMediaSourceIdentifier(item.identifier)
diskstation: SynologyDSMData = self.hass.data[DOMAIN][identifier.unique_id] diskstation: SynologyDSMData = self.hass.data[DOMAIN][identifier.unique_id]

View File

@ -50,23 +50,20 @@ async def async_setup_entry(
"""Set up System Bridge binary sensor based on a config entry.""" """Set up System Bridge binary sensor based on a config entry."""
coordinator: SystemBridgeDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id] coordinator: SystemBridgeDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
entities = [] entities = [
for description in BASE_BINARY_SENSOR_TYPES: SystemBridgeBinarySensor(coordinator, description, entry.data[CONF_PORT])
entities.append( for description in BASE_BINARY_SENSOR_TYPES
SystemBridgeBinarySensor(coordinator, description, entry.data[CONF_PORT]) ]
)
if ( if (
coordinator.data.battery coordinator.data.battery
and coordinator.data.battery.percentage and coordinator.data.battery.percentage
and coordinator.data.battery.percentage > -1 and coordinator.data.battery.percentage > -1
): ):
for description in BATTERY_BINARY_SENSOR_TYPES: entities.extend(
entities.append( SystemBridgeBinarySensor(coordinator, description, entry.data[CONF_PORT])
SystemBridgeBinarySensor( for description in BATTERY_BINARY_SENSOR_TYPES
coordinator, description, entry.data[CONF_PORT] )
)
)
async_add_entities(entities) async_add_entities(entities)

View File

@ -88,22 +88,21 @@ class SystemBridgeSource(MediaSource):
def _build_bridges(self) -> BrowseMediaSource: def _build_bridges(self) -> BrowseMediaSource:
"""Build bridges for System Bridge media.""" """Build bridges for System Bridge media."""
children = [] children = [
for entry in self.hass.config_entries.async_entries(DOMAIN): BrowseMediaSource(
if entry.entry_id is not None: domain=DOMAIN,
children.append( identifier=entry.entry_id,
BrowseMediaSource( media_class=MediaClass.DIRECTORY,
domain=DOMAIN, media_content_type="",
identifier=entry.entry_id, title=entry.title,
media_class=MediaClass.DIRECTORY, can_play=False,
media_content_type="", can_expand=True,
title=entry.title, children=[],
can_play=False, children_media_class=MediaClass.DIRECTORY,
can_expand=True, )
children=[], for entry in self.hass.config_entries.async_entries(DOMAIN)
children_media_class=MediaClass.DIRECTORY, if entry.entry_id is not None
) ]
)
return BrowseMediaSource( return BrowseMediaSource(
domain=DOMAIN, domain=DOMAIN,

View File

@ -364,45 +364,44 @@ async def async_setup_entry(
"""Set up System Bridge sensor based on a config entry.""" """Set up System Bridge sensor based on a config entry."""
coordinator: SystemBridgeDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id] coordinator: SystemBridgeDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
entities = [] entities = [
for description in BASE_SENSOR_TYPES: SystemBridgeSensor(coordinator, description, entry.data[CONF_PORT])
entities.append( for description in BASE_SENSOR_TYPES
SystemBridgeSensor(coordinator, description, entry.data[CONF_PORT]) ]
)
for index_device, device in enumerate(coordinator.data.disks.devices): for index_device, device in enumerate(coordinator.data.disks.devices):
if device.partitions is None: if device.partitions is None:
continue continue
for index_partition, partition in enumerate(device.partitions): entities.extend(
entities.append( SystemBridgeSensor(
SystemBridgeSensor( coordinator,
coordinator, SystemBridgeSensorEntityDescription(
SystemBridgeSensorEntityDescription( key=f"filesystem_{partition.mount_point.replace(':', '')}",
key=f"filesystem_{partition.mount_point.replace(':', '')}", name=f"{partition.mount_point} space used",
name=f"{partition.mount_point} space used", state_class=SensorStateClass.MEASUREMENT,
state_class=SensorStateClass.MEASUREMENT, native_unit_of_measurement=PERCENTAGE,
native_unit_of_measurement=PERCENTAGE, icon="mdi:harddisk",
icon="mdi:harddisk", value=(
value=( lambda data,
lambda data, dk=index_device,
dk=index_device, pk=index_partition: partition_usage(data, dk, pk)
pk=index_partition: partition_usage(data, dk, pk)
),
), ),
entry.data[CONF_PORT], ),
) entry.data[CONF_PORT],
) )
for index_partition, partition in enumerate(device.partitions)
)
if ( if (
coordinator.data.battery coordinator.data.battery
and coordinator.data.battery.percentage and coordinator.data.battery.percentage
and coordinator.data.battery.percentage > -1 and coordinator.data.battery.percentage > -1
): ):
for description in BATTERY_SENSOR_TYPES: entities.extend(
entities.append( SystemBridgeSensor(coordinator, description, entry.data[CONF_PORT])
SystemBridgeSensor(coordinator, description, entry.data[CONF_PORT]) for description in BATTERY_SENSOR_TYPES
) )
entities.append( entities.append(
SystemBridgeSensor( SystemBridgeSensor(
@ -466,127 +465,128 @@ async def async_setup_entry(
] ]
for index, gpu in enumerate(coordinator.data.gpus): for index, gpu in enumerate(coordinator.data.gpus):
entities = [ entities.extend(
*entities, [
SystemBridgeSensor( SystemBridgeSensor(
coordinator, coordinator,
SystemBridgeSensorEntityDescription( SystemBridgeSensorEntityDescription(
key=f"gpu_{gpu.id}_core_clock_speed", key=f"gpu_{gpu.id}_core_clock_speed",
name=f"{gpu.name} clock speed", name=f"{gpu.name} clock speed",
entity_registry_enabled_default=False, entity_registry_enabled_default=False,
state_class=SensorStateClass.MEASUREMENT, state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfFrequency.MEGAHERTZ, native_unit_of_measurement=UnitOfFrequency.MEGAHERTZ,
device_class=SensorDeviceClass.FREQUENCY, device_class=SensorDeviceClass.FREQUENCY,
icon="mdi:speedometer", icon="mdi:speedometer",
value=lambda data, k=index: gpu_core_clock_speed(data, k), value=lambda data, k=index: gpu_core_clock_speed(data, k),
),
entry.data[CONF_PORT],
), ),
entry.data[CONF_PORT], SystemBridgeSensor(
), coordinator,
SystemBridgeSensor( SystemBridgeSensorEntityDescription(
coordinator, key=f"gpu_{gpu.id}_memory_clock_speed",
SystemBridgeSensorEntityDescription( name=f"{gpu.name} memory clock speed",
key=f"gpu_{gpu.id}_memory_clock_speed", entity_registry_enabled_default=False,
name=f"{gpu.name} memory clock speed", state_class=SensorStateClass.MEASUREMENT,
entity_registry_enabled_default=False, native_unit_of_measurement=UnitOfFrequency.MEGAHERTZ,
state_class=SensorStateClass.MEASUREMENT, device_class=SensorDeviceClass.FREQUENCY,
native_unit_of_measurement=UnitOfFrequency.MEGAHERTZ, icon="mdi:speedometer",
device_class=SensorDeviceClass.FREQUENCY, value=lambda data, k=index: gpu_memory_clock_speed(data, k),
icon="mdi:speedometer", ),
value=lambda data, k=index: gpu_memory_clock_speed(data, k), entry.data[CONF_PORT],
), ),
entry.data[CONF_PORT], SystemBridgeSensor(
), coordinator,
SystemBridgeSensor( SystemBridgeSensorEntityDescription(
coordinator, key=f"gpu_{gpu.id}_memory_free",
SystemBridgeSensorEntityDescription( name=f"{gpu.name} memory free",
key=f"gpu_{gpu.id}_memory_free", state_class=SensorStateClass.MEASUREMENT,
name=f"{gpu.name} memory free", native_unit_of_measurement=UnitOfInformation.MEGABYTES,
state_class=SensorStateClass.MEASUREMENT, device_class=SensorDeviceClass.DATA_SIZE,
native_unit_of_measurement=UnitOfInformation.MEGABYTES, icon="mdi:memory",
device_class=SensorDeviceClass.DATA_SIZE, value=lambda data, k=index: gpu_memory_free(data, k),
icon="mdi:memory", ),
value=lambda data, k=index: gpu_memory_free(data, k), entry.data[CONF_PORT],
), ),
entry.data[CONF_PORT], SystemBridgeSensor(
), coordinator,
SystemBridgeSensor( SystemBridgeSensorEntityDescription(
coordinator, key=f"gpu_{gpu.id}_memory_used_percentage",
SystemBridgeSensorEntityDescription( name=f"{gpu.name} memory used %",
key=f"gpu_{gpu.id}_memory_used_percentage", state_class=SensorStateClass.MEASUREMENT,
name=f"{gpu.name} memory used %", native_unit_of_measurement=PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT, icon="mdi:memory",
native_unit_of_measurement=PERCENTAGE, value=lambda data, k=index: gpu_memory_used_percentage(data, k),
icon="mdi:memory", ),
value=lambda data, k=index: gpu_memory_used_percentage(data, k), entry.data[CONF_PORT],
), ),
entry.data[CONF_PORT], SystemBridgeSensor(
), coordinator,
SystemBridgeSensor( SystemBridgeSensorEntityDescription(
coordinator, key=f"gpu_{gpu.id}_memory_used",
SystemBridgeSensorEntityDescription( name=f"{gpu.name} memory used",
key=f"gpu_{gpu.id}_memory_used", entity_registry_enabled_default=False,
name=f"{gpu.name} memory used", state_class=SensorStateClass.MEASUREMENT,
entity_registry_enabled_default=False, native_unit_of_measurement=UnitOfInformation.MEGABYTES,
state_class=SensorStateClass.MEASUREMENT, device_class=SensorDeviceClass.DATA_SIZE,
native_unit_of_measurement=UnitOfInformation.MEGABYTES, icon="mdi:memory",
device_class=SensorDeviceClass.DATA_SIZE, value=lambda data, k=index: gpu_memory_used(data, k),
icon="mdi:memory", ),
value=lambda data, k=index: gpu_memory_used(data, k), entry.data[CONF_PORT],
), ),
entry.data[CONF_PORT], SystemBridgeSensor(
), coordinator,
SystemBridgeSensor( SystemBridgeSensorEntityDescription(
coordinator, key=f"gpu_{gpu.id}_fan_speed",
SystemBridgeSensorEntityDescription( name=f"{gpu.name} fan speed",
key=f"gpu_{gpu.id}_fan_speed", entity_registry_enabled_default=False,
name=f"{gpu.name} fan speed", state_class=SensorStateClass.MEASUREMENT,
entity_registry_enabled_default=False, native_unit_of_measurement=REVOLUTIONS_PER_MINUTE,
state_class=SensorStateClass.MEASUREMENT, icon="mdi:fan",
native_unit_of_measurement=REVOLUTIONS_PER_MINUTE, value=lambda data, k=index: gpu_fan_speed(data, k),
icon="mdi:fan", ),
value=lambda data, k=index: gpu_fan_speed(data, k), entry.data[CONF_PORT],
), ),
entry.data[CONF_PORT], SystemBridgeSensor(
), coordinator,
SystemBridgeSensor( SystemBridgeSensorEntityDescription(
coordinator, key=f"gpu_{gpu.id}_power_usage",
SystemBridgeSensorEntityDescription( name=f"{gpu.name} power usage",
key=f"gpu_{gpu.id}_power_usage", entity_registry_enabled_default=False,
name=f"{gpu.name} power usage", device_class=SensorDeviceClass.POWER,
entity_registry_enabled_default=False, state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.POWER, native_unit_of_measurement=UnitOfPower.WATT,
state_class=SensorStateClass.MEASUREMENT, value=lambda data, k=index: gpu_power_usage(data, k),
native_unit_of_measurement=UnitOfPower.WATT, ),
value=lambda data, k=index: gpu_power_usage(data, k), entry.data[CONF_PORT],
), ),
entry.data[CONF_PORT], SystemBridgeSensor(
), coordinator,
SystemBridgeSensor( SystemBridgeSensorEntityDescription(
coordinator, key=f"gpu_{gpu.id}_temperature",
SystemBridgeSensorEntityDescription( name=f"{gpu.name} temperature",
key=f"gpu_{gpu.id}_temperature", entity_registry_enabled_default=False,
name=f"{gpu.name} temperature", device_class=SensorDeviceClass.TEMPERATURE,
entity_registry_enabled_default=False, state_class=SensorStateClass.MEASUREMENT,
device_class=SensorDeviceClass.TEMPERATURE, native_unit_of_measurement=UnitOfTemperature.CELSIUS,
state_class=SensorStateClass.MEASUREMENT, value=lambda data, k=index: gpu_temperature(data, k),
native_unit_of_measurement=UnitOfTemperature.CELSIUS, ),
value=lambda data, k=index: gpu_temperature(data, k), entry.data[CONF_PORT],
), ),
entry.data[CONF_PORT], SystemBridgeSensor(
), coordinator,
SystemBridgeSensor( SystemBridgeSensorEntityDescription(
coordinator, key=f"gpu_{gpu.id}_usage_percentage",
SystemBridgeSensorEntityDescription( name=f"{gpu.name} usage %",
key=f"gpu_{gpu.id}_usage_percentage", state_class=SensorStateClass.MEASUREMENT,
name=f"{gpu.name} usage %", native_unit_of_measurement=PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT, icon="mdi:percent",
native_unit_of_measurement=PERCENTAGE, value=lambda data, k=index: gpu_usage_percentage(data, k),
icon="mdi:percent", ),
value=lambda data, k=index: gpu_usage_percentage(data, k), entry.data[CONF_PORT],
), ),
entry.data[CONF_PORT], ]
), )
]
if coordinator.data.cpu.per_cpu is not None: if coordinator.data.cpu.per_cpu is not None:
for cpu in coordinator.data.cpu.per_cpu: for cpu in coordinator.data.cpu.per_cpu:

View File

@ -92,21 +92,20 @@ async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None: ) -> None:
"""Set up System Montor binary sensors based on a config entry.""" """Set up System Montor binary sensors based on a config entry."""
entities: list[SystemMonitorSensor] = []
coordinator: SystemMonitorCoordinator = hass.data[DOMAIN_COORDINATOR] coordinator: SystemMonitorCoordinator = hass.data[DOMAIN_COORDINATOR]
for sensor_description in SENSOR_TYPES: async_add_entities(
_entry = entry.options.get(BINARY_SENSOR_DOMAIN, {}) SystemMonitorSensor(
for argument in _entry.get(CONF_PROCESS, []): coordinator,
entities.append( sensor_description,
SystemMonitorSensor( entry.entry_id,
coordinator, argument,
sensor_description, )
entry.entry_id, for sensor_description in SENSOR_TYPES
argument, for argument in entry.options.get(BINARY_SENSOR_DOMAIN, {}).get(
) CONF_PROCESS, []
) )
async_add_entities(entities) )
class SystemMonitorSensor( class SystemMonitorSensor(

View File

@ -658,9 +658,7 @@ def test_get_significant_states_only(
return hass.states.get(entity_id) return hass.states.get(entity_id)
start = dt_util.utcnow() - timedelta(minutes=4) start = dt_util.utcnow() - timedelta(minutes=4)
points = [] points = [start + timedelta(minutes=i) for i in range(1, 4)]
for i in range(1, 4):
points.append(start + timedelta(minutes=i))
states = [] states = []
with freeze_time(start) as freezer: with freeze_time(start) as freezer:

View File

@ -517,9 +517,7 @@ def test_get_significant_states_only(
return hass.states.get(entity_id) return hass.states.get(entity_id)
start = dt_util.utcnow() - timedelta(minutes=4) start = dt_util.utcnow() - timedelta(minutes=4)
points = [] points = [start + timedelta(minutes=i) for i in range(1, 4)]
for i in range(1, 4):
points.append(start + timedelta(minutes=i))
states = [] states = []
with freeze_time(start) as freezer: with freeze_time(start) as freezer:

View File

@ -507,9 +507,7 @@ def test_get_significant_states_only(
return hass.states.get(entity_id) return hass.states.get(entity_id)
start = dt_util.utcnow() - timedelta(minutes=4) start = dt_util.utcnow() - timedelta(minutes=4)
points = [] points = [start + timedelta(minutes=i) for i in range(1, 4)]
for i in range(1, 4):
points.append(start + timedelta(minutes=i))
states = [] states = []
with freeze_time(start) as freezer: with freeze_time(start) as freezer:

View File

@ -914,18 +914,17 @@ async def test_stats_timestamp_conversion_is_reentrant(
def _get_all_short_term_stats() -> list[dict[str, Any]]: def _get_all_short_term_stats() -> list[dict[str, Any]]:
with session_scope(hass=hass) as session: with session_scope(hass=hass) as session:
results = [] results = [
for result in ( {
session.query(old_db_schema.StatisticsShortTerm) field.name: getattr(result, field.name)
.where(old_db_schema.StatisticsShortTerm.metadata_id == 1000) for field in old_db_schema.StatisticsShortTerm.__table__.c
.all() }
): for result in (
results.append( session.query(old_db_schema.StatisticsShortTerm)
{ .where(old_db_schema.StatisticsShortTerm.metadata_id == 1000)
field.name: getattr(result, field.name) .all()
for field in old_db_schema.StatisticsShortTerm.__table__.c
}
) )
]
return sorted(results, key=lambda row: row["start_ts"]) return sorted(results, key=lambda row: row["start_ts"])
# Do not optimize this block, its intentionally written to interleave # Do not optimize this block, its intentionally written to interleave
@ -1099,14 +1098,12 @@ async def test_stats_timestamp_with_one_by_one(
def _get_all_stats(table: old_db_schema.StatisticsBase) -> list[dict[str, Any]]: def _get_all_stats(table: old_db_schema.StatisticsBase) -> list[dict[str, Any]]:
"""Get all stats from a table.""" """Get all stats from a table."""
with session_scope(hass=hass) as session: with session_scope(hass=hass) as session:
results = [] results = [
for result in session.query(table).where(table.metadata_id == 1000).all(): {field.name: getattr(result, field.name) for field in table.__table__.c}
results.append( for result in session.query(table)
{ .where(table.metadata_id == 1000)
field.name: getattr(result, field.name) .all()
for field in table.__table__.c ]
}
)
return sorted(results, key=lambda row: row["start_ts"]) return sorted(results, key=lambda row: row["start_ts"])
def _insert_and_do_migration(): def _insert_and_do_migration():
@ -1326,14 +1323,12 @@ async def test_stats_timestamp_with_one_by_one_removes_duplicates(
def _get_all_stats(table: old_db_schema.StatisticsBase) -> list[dict[str, Any]]: def _get_all_stats(table: old_db_schema.StatisticsBase) -> list[dict[str, Any]]:
"""Get all stats from a table.""" """Get all stats from a table."""
with session_scope(hass=hass) as session: with session_scope(hass=hass) as session:
results = [] results = [
for result in session.query(table).where(table.metadata_id == 1000).all(): {field.name: getattr(result, field.name) for field in table.__table__.c}
results.append( for result in session.query(table)
{ .where(table.metadata_id == 1000)
field.name: getattr(result, field.name) .all()
for field in table.__table__.c ]
}
)
return sorted(results, key=lambda row: row["start_ts"]) return sorted(results, key=lambda row: row["start_ts"])
def _insert_and_do_migration(): def _insert_and_do_migration():

View File

@ -474,19 +474,18 @@ async def test_dhcp_ip_update(
const.DOMAIN, context={"source": config_entries.SOURCE_DHCP}, data=dhcp_data const.DOMAIN, context={"source": config_entries.SOURCE_DHCP}, data=dhcp_data
) )
expected_calls = [] expected_calls = [
for host in host_call_list: call(
expected_calls.append( host,
call( TEST_USERNAME,
host, TEST_PASSWORD,
TEST_USERNAME, port=TEST_PORT,
TEST_PASSWORD, use_https=TEST_USE_HTTPS,
port=TEST_PORT, protocol=DEFAULT_PROTOCOL,
use_https=TEST_USE_HTTPS, timeout=DEFAULT_TIMEOUT,
protocol=DEFAULT_PROTOCOL,
timeout=DEFAULT_TIMEOUT,
)
) )
for host in host_call_list
]
assert reolink_connect_class.call_args_list == expected_calls assert reolink_connect_class.call_args_list == expected_calls

View File

@ -1292,17 +1292,16 @@ async def test_state_characteristics(hass: HomeAssistant) -> None:
"unit": "%", "unit": "%",
}, },
) )
sensors_config = [] sensors_config = [
for characteristic in characteristics: {
sensors_config.append( "platform": "statistics",
{ "name": f"test_{characteristic['source_sensor_domain']}_{characteristic['name']}",
"platform": "statistics", "entity_id": f"{characteristic['source_sensor_domain']}.test_monitored",
"name": f"test_{characteristic['source_sensor_domain']}_{characteristic['name']}", "state_characteristic": characteristic["name"],
"entity_id": f"{characteristic['source_sensor_domain']}.test_monitored", "max_age": {"minutes": 8}, # 9 values spaces by one minute
"state_characteristic": characteristic["name"], }
"max_age": {"minutes": 8}, # 9 values spaces by one minute for characteristic in characteristics
} ]
)
with freeze_time(current_time) as freezer: with freeze_time(current_time) as freezer:
assert await async_setup_component( assert await async_setup_component(

View File

@ -69,10 +69,10 @@ class MockedInterface(dict):
def GetFriendList(self, steamid: str) -> dict: def GetFriendList(self, steamid: str) -> dict:
"""Get friend list.""" """Get friend list."""
fake_friends = [{"steamid": ACCOUNT_2}] fake_friends = [{"steamid": ACCOUNT_2}]
for _i in range(0, 4): fake_friends.extend(
fake_friends.append( {"steamid": "".join(random.choices(string.digits, k=len(ACCOUNT_1)))}
{"steamid": "".join(random.choices(string.digits, k=len(ACCOUNT_1)))} for _ in range(0, 4)
) )
return {"friendslist": {"friends": fake_friends}} return {"friendslist": {"friends": fake_friends}}
def GetPlayerSummaries(self, steamids: str | list[str]) -> dict: def GetPlayerSummaries(self, steamids: str | list[str]) -> dict:

View File

@ -399,9 +399,7 @@ async def test_hls_max_segments(
# Only NUM_PLAYLIST_SEGMENTS are returned in the playlist. # Only NUM_PLAYLIST_SEGMENTS are returned in the playlist.
start = MAX_SEGMENTS + 1 - NUM_PLAYLIST_SEGMENTS start = MAX_SEGMENTS + 1 - NUM_PLAYLIST_SEGMENTS
segments = [] segments = [make_segment(sequence) for sequence in range(start, MAX_SEGMENTS + 1)]
for sequence in range(start, MAX_SEGMENTS + 1):
segments.append(make_segment(sequence))
assert await resp.text() == make_playlist(sequence=start, segments=segments) assert await resp.text() == make_playlist(sequence=start, segments=segments)
# Fetch the actual segments with a fake byte payload # Fetch the actual segments with a fake byte payload
@ -497,9 +495,7 @@ async def test_hls_max_segments_discontinuity(
# EXT-X-DISCONTINUITY tag to be omitted and EXT-X-DISCONTINUITY-SEQUENCE # EXT-X-DISCONTINUITY tag to be omitted and EXT-X-DISCONTINUITY-SEQUENCE
# returned instead. # returned instead.
start = MAX_SEGMENTS + 1 - NUM_PLAYLIST_SEGMENTS start = MAX_SEGMENTS + 1 - NUM_PLAYLIST_SEGMENTS
segments = [] segments = [make_segment(sequence) for sequence in range(start, MAX_SEGMENTS + 1)]
for sequence in range(start, MAX_SEGMENTS + 1):
segments.append(make_segment(sequence))
assert await resp.text() == make_playlist( assert await resp.text() == make_playlist(
sequence=start, sequence=start,
discontinuity_sequence=1, discontinuity_sequence=1,

View File

@ -96,10 +96,10 @@ def make_segment_with_parts(
response = [] response = []
if discontinuity: if discontinuity:
response.append("#EXT-X-DISCONTINUITY") response.append("#EXT-X-DISCONTINUITY")
for i in range(num_parts): response.extend(
response.append( f'#EXT-X-PART:DURATION={TEST_PART_DURATION:.3f},URI="./segment/{segment}.{i}.m4s"{",INDEPENDENT=YES" if i%independent_period==0 else ""}'
f'#EXT-X-PART:DURATION={TEST_PART_DURATION:.3f},URI="./segment/{segment}.{i}.m4s"{",INDEPENDENT=YES" if i%independent_period==0 else ""}' for i in range(num_parts)
) )
response.extend( response.extend(
[ [
"#EXT-X-PROGRAM-DATE-TIME:" "#EXT-X-PROGRAM-DATE-TIME:"