Skip to content

Fix ruff formatting violations in custom_components #32

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
Copilot wants to merge 1 commit into from
Closed

Fix ruff formatting violations in custom_components #32

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion custom_components/ha_daily_counter/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@

CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)


async def async_setup(hass: HomeAssistant, config: dict[str, Any]) -> bool:
"""Set up the integration from configuration.yaml (not used)."""
return True
Expand Down Expand Up @@ -95,7 +96,9 @@ async def handle_set_counter(call: ServiceCall) -> None:
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
# Llamada a la versión plural de unload si existe; usar cast para evitar chequeos estáticos.
unloaded = await cast(Any, hass.config_entries).async_forward_entry_unloads(entry, ["sensor"])
unloaded = await cast(Any, hass.config_entries).async_forward_entry_unloads(
entry, ["sensor"]
)
return all(unloaded)


Expand Down
138 changes: 79 additions & 59 deletions custom_components/ha_daily_counter/config_flow.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,10 +72,10 @@ async def async_step_user(
if user_input is not None:
try:
self._name = user_input[CONF_NAME]

# Store domain filter for next steps
self._domain_filter = user_input.get("domain_filter")

trigger_entity = user_input[ATTR_TRIGGER_ENTITY]
trigger_state = user_input[ATTR_TRIGGER_STATE]

Expand All @@ -98,7 +98,7 @@ async def async_step_user(

# Si se pidió agregar otro, vamos al paso de añadir más triggers
return await self.async_step_another_trigger()

except Exception as err:
_LOGGER.error(
"Error in user step: %s",
Expand All @@ -115,16 +115,16 @@ async def async_step_user(
data_schema = vol.Schema(
{
vol.Required(CONF_NAME): str,
vol.Required("domain_filter", default=domain_filter): SelectSelector(
vol.Required(
"domain_filter", default=domain_filter
): SelectSelector(
SelectSelectorConfig(
options=DOMAIN_OPTIONS,
mode=SelectSelectorMode.DROPDOWN,
)
),
vol.Required(ATTR_TRIGGER_ENTITY): EntitySelector(
EntitySelectorConfig(
domain=[domain_filter]
)
EntitySelectorConfig(domain=[domain_filter])
),
vol.Required(ATTR_TRIGGER_STATE): str,
vol.Optional("add_another", default=False): bool,
Expand All @@ -147,7 +147,7 @@ async def async_step_user(
vol.Required(ATTR_TRIGGER_STATE): str,
}
)

return self.async_show_form(
step_id="user",
data_schema=data_schema,
Expand All @@ -170,11 +170,11 @@ async def async_step_another_trigger(
try:
# Store text filter if provided
self._text_filter = user_input.get("text_filter", "")

# Guardamos la lógica seleccionada solo en el primer trigger adicional
if len(self._triggers) == 1 and "logic" in user_input:
self._logic = user_input.get("logic", "OR")

# If entity is selected, add to triggers
if ATTR_TRIGGER_ENTITY in user_input:
trigger_entity = user_input[ATTR_TRIGGER_ENTITY]
Expand All @@ -195,7 +195,7 @@ async def async_step_another_trigger(

# Si se pidió agregar otro, repetimos este mismo paso
return await self.async_step_another_trigger()

except Exception as err:
_LOGGER.error(
"Error processing another_trigger input: %s",
Expand All @@ -211,10 +211,11 @@ async def async_step_another_trigger(
all_entities = [
e.entity_id
for e in self.hass.states.async_all()
if self._available_domain and e.entity_id.startswith(self._available_domain)
if self._available_domain
and e.entity_id.startswith(self._available_domain)
]
available_entities = [e for e in all_entities if e not in excluded_entities]

# Apply text filter if provided
text_filter = self._text_filter.lower()
if text_filter:
Expand Down Expand Up @@ -252,16 +253,20 @@ async def async_step_another_trigger(

# Handle case when no entities are available
if not available_entities:
_LOGGER.warning("No available entities found for domain %s", self._available_domain)
_LOGGER.warning(
"No available entities found for domain %s", self._available_domain
)
# Instead of showing an error, go back to user step to start over
return await self.async_step_finish()

# Usamos SelectSelector con opciones construidas desde available_entities
select_options = [SelectOptionDict(value=e, label=e) for e in available_entities]
select_options = [
SelectOptionDict(value=e, label=e) for e in available_entities
]

# Determinar si mostrar el selector de lógica (solo en el primer trigger adicional)
is_first_additional = len(self._triggers) == 1

# Construir el esquema del formulario dinámicamente
schema_dict = {
vol.Optional("text_filter", default=self._text_filter): TextSelector(
Expand All @@ -277,16 +282,16 @@ async def async_step_another_trigger(
),
vol.Required(ATTR_TRIGGER_STATE): str,
}

# Agregar selector de lógica solo si es el primer trigger adicional
if is_first_additional:
schema_dict[vol.Optional("logic", default="OR")] = vol.In(LOGIC_OPTIONS)

# Agregar checkbox add_another al final
schema_dict[vol.Optional("add_another", default=False)] = bool

data_schema = vol.Schema(schema_dict)

except Exception as err:
_LOGGER.error(
"Error building another_trigger form schema: %s",
Expand All @@ -298,7 +303,7 @@ async def async_step_another_trigger(
entity_config = EntitySelectorConfig()
if self._available_domain:
entity_config = EntitySelectorConfig(domain=[self._available_domain])

data_schema = vol.Schema(
{
vol.Required(ATTR_TRIGGER_ENTITY): EntitySelector(entity_config),
Expand All @@ -309,14 +314,16 @@ async def async_step_another_trigger(
return self.async_show_form(
step_id="another_trigger",
data_schema=data_schema,
description_placeholders={"previous_triggers": ", ".join(prev_friendly)} if prev_friendly else {},
description_placeholders={"previous_triggers": ", ".join(prev_friendly)}
if prev_friendly
else {},
errors=errors,
)

async def async_step_finish(self) -> FlowResult:
"""Finaliza el flujo y crea la entry con los triggers y la lógica seleccionada en el primer paso."""
_LOGGER.debug("Config flow step 'finish' started")

try:
title = self._name or "HA Daily Counter"
data = {
Expand All @@ -332,7 +339,7 @@ async def async_step_finish(self) -> FlowResult:
)

return self.async_create_entry(title=title, data=data)

except Exception as err:
_LOGGER.error(
"Error creating config entry: %s",
Expand All @@ -353,12 +360,14 @@ def __init__(self) -> None:
self._selected_edit_index: int | None = None
self._editing_counter: dict[str, Any] = {}

async def async_step_init(self, user_input: dict[str, Any] | None = None) -> FlowResult:
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Initial step: add, edit, or delete a counter."""
# Initialize counters from config_entry on first call
if not self._counters:
self._counters = list(self.config_entry.options.get("counters", []))

if user_input is not None:
if user_input["action"] == "add":
return await self.async_step_user()
Expand All @@ -378,15 +387,17 @@ async def async_step_init(self, user_input: dict[str, Any] | None = None) -> Flo
SelectOptionDict(value="add", label="Add counter"),
SelectOptionDict(value="edit", label="Edit counter"),
SelectOptionDict(value="delete", label="Delete counter"),
SelectOptionDict(value="finish", label="Finish setup")
SelectOptionDict(value="finish", label="Finish setup"),
],
mode=SelectSelectorMode.DROPDOWN
mode=SelectSelectorMode.DROPDOWN,
)
)
},
)

async def async_step_user(self, user_input: dict[str, Any] | None = None) -> FlowResult:
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Step to collect the name of the new counter."""
if user_input is not None:
self._new_counter["name"] = user_input["name"]
Expand All @@ -397,20 +408,22 @@ async def async_step_user(self, user_input: dict[str, Any] | None = None) -> Flo
data_schema={"name": str},
)

async def async_step_trigger_entity(self, user_input: dict[str, Any] | None = None) -> FlowResult:
async def async_step_trigger_entity(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Step to collect the entity that triggers the counter."""
if user_input is not None:
self._new_counter["trigger_entity"] = user_input["trigger_entity"]
return await self.async_step_trigger_state()

return self.async_show_form(
step_id="trigger_entity",
data_schema={
"trigger_entity": EntitySelector(EntitySelectorConfig())
},
data_schema={"trigger_entity": EntitySelector(EntitySelectorConfig())},
)

async def async_step_trigger_state(self, user_input: dict[str, Any] | None = None) -> FlowResult:
async def async_step_trigger_state(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Step to collect the trigger state."""
if user_input is not None:
self._new_counter["trigger_state"] = user_input["trigger_state"]
Expand All @@ -421,12 +434,12 @@ async def async_step_trigger_state(self, user_input: dict[str, Any] | None = Non

return self.async_show_form(
step_id="trigger_state",
data_schema={
"trigger_state": str
},
data_schema={"trigger_state": str},
)

async def async_step_select_edit(self, user_input: dict[str, Any] | None = None) -> FlowResult:
async def async_step_select_edit(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Step to select a counter to edit."""
if not self._counters:
return await self.async_step_init()
Expand All @@ -451,13 +464,15 @@ async def async_step_select_edit(self, user_input: dict[str, Any] | None = None)
SelectOptionDict(value=c["name"], label=c["name"])
for c in self._counters
],
mode=SelectSelectorMode.DROPDOWN
mode=SelectSelectorMode.DROPDOWN,
)
)
},
)

async def async_step_edit_trigger_entity(self, user_input: dict[str, Any] | None = None) -> FlowResult:
async def async_step_edit_trigger_entity(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Step to edit the trigger entity."""
if user_input is not None:
self._editing_counter["trigger_entity"] = user_input["trigger_entity"]
Expand All @@ -468,46 +483,47 @@ async def async_step_edit_trigger_entity(self, user_input: dict[str, Any] | None

return self.async_show_form(
step_id="edit_trigger_entity",
data_schema={
"trigger_entity": EntitySelector(
EntitySelectorConfig()
)
},
data_schema={"trigger_entity": EntitySelector(EntitySelectorConfig())},
description_placeholders={
"current_value": current_entity,
"counter_name": self._editing_counter.get("name", "")
}
"counter_name": self._editing_counter.get("name", ""),
},
)

async def async_step_edit_trigger_state(self, user_input: dict[str, Any] | None = None) -> FlowResult:
async def async_step_edit_trigger_state(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Step to edit the trigger state."""
if user_input is not None:
self._editing_counter["trigger_state"] = user_input["trigger_state"]
# Update the counter in the list
if self._selected_edit_index is not None and 0 <= self._selected_edit_index < len(self._counters):
if (
self._selected_edit_index is not None
and 0 <= self._selected_edit_index < len(self._counters)
):
self._counters[self._selected_edit_index] = self._editing_counter

# Reset editing state
self._selected_edit_index = None
self._editing_counter = {}

return await self.async_step_init()

# Get current trigger state value
current_state = self._editing_counter.get("trigger_state", "")

return self.async_show_form(
step_id="edit_trigger_state",
data_schema={
"trigger_state": str
},
data_schema={"trigger_state": str},
description_placeholders={
"current_value": current_state,
"counter_name": self._editing_counter.get("name", "")
}
"counter_name": self._editing_counter.get("name", ""),
},
)

async def async_step_select_delete(self, user_input: dict[str, Any] | None = None) -> FlowResult:
async def async_step_select_delete(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Step to select a counter to delete."""
if not self._counters:
return await self.async_step_init()
Expand All @@ -525,16 +541,20 @@ async def async_step_select_delete(self, user_input: dict[str, Any] | None = Non
SelectOptionDict(value=c["name"], label=c["name"])
for c in self._counters
],
mode=SelectSelectorMode.DROPDOWN
mode=SelectSelectorMode.DROPDOWN,
)
)
},
)

async def async_step_confirm_delete(self, user_input: dict[str, Any] | None = None) -> FlowResult:
async def async_step_confirm_delete(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Confirm and delete the selected counter."""
if user_input is not None and user_input.get("confirm_delete"):
self._counters = [c for c in self._counters if c["name"] != self._selected_delete_name]
self._counters = [
c for c in self._counters if c["name"] != self._selected_delete_name
]

return await self.async_step_init()

Expand Down
Loading
Loading