Add a Local To-do component

.strict-typing

@@ -204,6 +204,7 @@ homeassistant.components.light.*
 homeassistant.components.litejet.*
 homeassistant.components.litterrobot.*
 homeassistant.components.local_ip.*
+homeassistant.components.local_todo.*
 homeassistant.components.lock.*
 homeassistant.components.logbook.*
 homeassistant.components.logger.*

CODEOWNERS

@@ -708,6 +708,8 @@ build.json @home-assistant/supervisor
 /tests/components/local_calendar/ @allenporter
 /homeassistant/components/local_ip/ @issacg
 /tests/components/local_ip/ @issacg
+/homeassistant/components/local_todo/ @allenporter
+/tests/components/local_todo/ @allenporter
 /homeassistant/components/lock/ @home-assistant/core
 /tests/components/lock/ @home-assistant/core
 /homeassistant/components/logbook/ @home-assistant/core

homeassistant/components/local_todo/__init__.py

@@ -0,0 +1,55 @@
+"""The Local To-do integration."""
+from __future__ import annotations
+
+from pathlib import Path
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryNotReady
+from homeassistant.util import slugify
+
+from .const import CONF_STORAGE_KEY, CONF_TODO_LIST_NAME, DOMAIN
+from .store import LocalTodoListStore
+
+PLATFORMS: list[Platform] = [Platform.TODO]
+
+STORAGE_PATH = ".storage/local_todo.{key}.ics"
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up Local To-do from a config entry."""
+
+    hass.data.setdefault(DOMAIN, {})
+
+    path = Path(hass.config.path(STORAGE_PATH.format(key=entry.data[CONF_STORAGE_KEY])))
+    store = LocalTodoListStore(hass, path)
+    try:
+        await store.async_load()
+    except OSError as err:
+        raise ConfigEntryNotReady("Failed to load file {path}: {err}") from err
+
+    hass.data[DOMAIN][entry.entry_id] = store
+
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
+        hass.data[DOMAIN].pop(entry.entry_id)
+
+    return unload_ok
+
+
+async def async_remove_entry(hass: HomeAssistant, entry: ConfigEntry) -> None:
+    """Handle removal of an entry."""
+    key = slugify(entry.data[CONF_TODO_LIST_NAME])
+    path = Path(hass.config.path(STORAGE_PATH.format(key=key)))
+
+    def unlink(path: Path) -> None:
+        path.unlink(missing_ok=True)
+
+    await hass.async_add_executor_job(unlink, path)

homeassistant/components/local_todo/config_flow.py

@@ -0,0 +1,44 @@
+"""Config flow for Local To-do integration."""
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+import voluptuous as vol
+
+from homeassistant import config_entries
+from homeassistant.data_entry_flow import FlowResult
+from homeassistant.util import slugify
+
+from .const import CONF_STORAGE_KEY, CONF_TODO_LIST_NAME, DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+STEP_USER_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_TODO_LIST_NAME): str,
+    }
+)
+
+
+class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for Local To-do."""
+
+    VERSION = 1
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle the initial step."""
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            key = slugify(user_input[CONF_TODO_LIST_NAME])
+            self._async_abort_entries_match({CONF_STORAGE_KEY: key})
+            user_input[CONF_STORAGE_KEY] = key
+            return self.async_create_entry(
+                title=user_input[CONF_TODO_LIST_NAME], data=user_input
+            )
+
+        return self.async_show_form(
+            step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
+        )

homeassistant/components/local_todo/const.py

@@ -0,0 +1,6 @@
+"""Constants for the Local To-do integration."""
+
+DOMAIN = "local_todo"
+
+CONF_TODO_LIST_NAME = "todo_list_name"
+CONF_STORAGE_KEY = "storage_key"

homeassistant/components/local_todo/manifest.json

@@ -0,0 +1,9 @@
+{
+  "domain": "local_todo",
+  "name": "Local To-do",
+  "codeowners": ["@allenporter"],
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/local_todo",
+  "iot_class": "local_polling",
+  "requirements": ["ical==5.1.0"]
+}

homeassistant/components/local_todo/store.py

@@ -0,0 +1,36 @@
+"""Local storage for the Local To-do integration."""
+
+import asyncio
+from pathlib import Path
+
+from homeassistant.core import HomeAssistant
+
+
+class LocalTodoListStore:
+    """Local storage for a single To-do list."""
+
+    def __init__(self, hass: HomeAssistant, path: Path) -> None:
+        """Initialize LocalTodoListStore."""
+        self._hass = hass
+        self._path = path
+        self._lock = asyncio.Lock()
+
+    async def async_load(self) -> str:
+        """Load the calendar from disk."""
+        async with self._lock:
+            return await self._hass.async_add_executor_job(self._load)
+
+    def _load(self) -> str:
+        """Load the calendar from disk."""
+        if not self._path.exists():
+            return ""
+        return self._path.read_text()
+
+    async def async_store(self, ics_content: str) -> None:
+        """Persist the calendar to storage."""
+        async with self._lock:
+            await self._hass.async_add_executor_job(self._store, ics_content)
+
+    def _store(self, ics_content: str) -> None:
+        """Persist the calendar to storage."""
+        self._path.write_text(ics_content)

homeassistant/components/local_todo/strings.json

@@ -0,0 +1,16 @@
+{
+  "title": "Local To-do",
+  "config": {
+    "step": {
+      "user": {
+        "description": "Please choose a name for your new To-do list",
+        "data": {
+          "todo_list_name": "To-do list name"
+        }
+      }
+    },
+    "abort": {
+      "already_configured": "[%key:common::config_flow::abort::already_configured_service%]"
+    }
+  }
+}

homeassistant/components/local_todo/todo.py

@@ -0,0 +1,162 @@
+"""A Local To-do todo platform."""
+
+from collections.abc import Iterable
+import dataclasses
+import logging
+from typing import Any
+
+from ical.calendar import Calendar
+from ical.calendar_stream import IcsCalendarStream
+from ical.store import TodoStore
+from ical.todo import Todo, TodoStatus
+from pydantic import ValidationError
+
+from homeassistant.components.todo import (
+    TodoItem,
+    TodoItemStatus,
+    TodoListEntity,
+    TodoListEntityFeature,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+from .const import CONF_TODO_LIST_NAME, DOMAIN
+from .store import LocalTodoListStore
+
+_LOGGER = logging.getLogger(__name__)
+
+
+PRODID = "-//homeassistant.io//local_todo 1.0//EN"
+
+ICS_TODO_STATUS_MAP = {
+    TodoStatus.IN_PROCESS: TodoItemStatus.NEEDS_ACTION,
+    TodoStatus.NEEDS_ACTION: TodoItemStatus.NEEDS_ACTION,
+    TodoStatus.COMPLETED: TodoItemStatus.COMPLETED,
+    TodoStatus.CANCELLED: TodoItemStatus.COMPLETED,
+}
+ICS_TODO_STATUS_MAP_INV = {
+    TodoItemStatus.COMPLETED: TodoStatus.COMPLETED,
+    TodoItemStatus.NEEDS_ACTION: TodoStatus.NEEDS_ACTION,
+}
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up the local_todo todo platform."""
+
+    store = hass.data[DOMAIN][config_entry.entry_id]
+    ics = await store.async_load()
+    calendar = IcsCalendarStream.calendar_from_ics(ics)
+    calendar.prodid = PRODID
+
+    name = config_entry.data[CONF_TODO_LIST_NAME]
+    entity = LocalTodoListEntity(store, calendar, name, unique_id=config_entry.entry_id)
+    async_add_entities([entity], True)
+
+
+def _todo_dict_factory(obj: Iterable[tuple[str, Any]]) -> dict[str, str]:
+    """Convert TodoItem dataclass items to dictionary of attributes for ical consumption."""
+    result: dict[str, str] = {}
+    for name, value in obj:
+        if name == "status":
+            result[name] = ICS_TODO_STATUS_MAP_INV[value]
+        elif value is not None:
+            result[name] = value
+    return result
+
+
+def _convert_item(item: TodoItem) -> Todo:
+    """Convert a HomeAssistant TodoItem to an ical Todo."""
+    try:
+        return Todo(**dataclasses.asdict(item, dict_factory=_todo_dict_factory))
+    except ValidationError as err:
+        _LOGGER.debug("Error parsing todo input fields: %s (%s)", item, err)
+        raise HomeAssistantError("Error parsing todo input fields") from err
+
+
+class LocalTodoListEntity(TodoListEntity):
+    """A To-do List representation of the Shopping List."""
+
+    _attr_has_entity_name = True
+    _attr_supported_features = (
+        TodoListEntityFeature.CREATE_TODO_ITEM
+        | TodoListEntityFeature.DELETE_TODO_ITEM
+        | TodoListEntityFeature.UPDATE_TODO_ITEM
+        | TodoListEntityFeature.MOVE_TODO_ITEM
+    )
+    _attr_should_poll = False
+
+    def __init__(
+        self,
+        store: LocalTodoListStore,
+        calendar: Calendar,
+        name: str,
+        unique_id: str,
+    ) -> None:
+        """Initialize LocalTodoListEntity."""
+        self._store = store
+        self._calendar = calendar
+        self._attr_name = name.capitalize()
+        self._attr_unique_id = unique_id
+
+    async def async_update(self) -> None:
+        """Update entity state based on the local To-do items."""
+        self._attr_todo_items = [
+            TodoItem(
+                uid=item.uid,
+                summary=item.summary or "",
+                status=ICS_TODO_STATUS_MAP.get(
+                    item.status or TodoStatus.NEEDS_ACTION, TodoItemStatus.NEEDS_ACTION
+                ),
+            )
+            for item in self._calendar.todos
+        ]
+
+    async def async_create_todo_item(self, item: TodoItem) -> None:
+        """Add an item to the To-do list."""
+        todo = _convert_item(item)
+        TodoStore(self._calendar).add(todo)
+        await self._async_save()
+        await self.async_update_ha_state(force_refresh=True)
+
+    async def async_update_todo_item(self, item: TodoItem) -> None:
+        """Update an item to the To-do list."""
+        todo = _convert_item(item)
+        TodoStore(self._calendar).edit(todo.uid, todo)
+        await self._async_save()
+        await self.async_update_ha_state(force_refresh=True)
+
+    async def async_delete_todo_items(self, uids: list[str]) -> None:
+        """Add an item to the To-do list."""
+        store = TodoStore(self._calendar)
+        for uid in uids:
+            store.delete(uid)
+        await self._async_save()
+        await self.async_update_ha_state(force_refresh=True)
+
+    async def async_move_todo_item(self, uid: str, pos: int) -> None:
+        """Re-order an item to the To-do list."""
+        todos = self._calendar.todos
+        found_item: Todo | None = None
+        for idx, itm in enumerate(todos):
+            if itm.uid == uid:
+                found_item = itm
+                todos.pop(idx)
+                break
+        if found_item is None:
+            raise HomeAssistantError(
+                f"Item '{uid}' not found in todo list {self.entity_id}"
+            )
+        todos.insert(pos, found_item)
+        await self._async_save()
+        await self.async_update_ha_state(force_refresh=True)
+
+    async def _async_save(self) -> None:
+        """Persist the todo list to disk."""
+        content = IcsCalendarStream.calendar_to_ics(self._calendar)
+        await self._store.async_store(content)

homeassistant/generated/config_flows.py

@@ -262,6 +262,7 @@
         "livisi",
         "local_calendar",
         "local_ip",
+        "local_todo",
         "locative",
         "logi_circle",
         "lookin",

homeassistant/generated/integrations.json

@@ -3105,6 +3105,11 @@
       "config_flow": true,
       "iot_class": "local_polling"
     },
+    "local_todo": {
+      "integration_type": "hub",
+      "config_flow": true,
+      "iot_class": "local_polling"
+    },
     "locative": {
       "name": "Locative",
       "integration_type": "hub",
@@ -6825,6 +6830,7 @@
     "islamic_prayer_times",
     "local_calendar",
     "local_ip",
+    "local_todo",
     "min_max",
     "mobile_app",
     "moehlenhoff_alpha2",

mypy.ini

@@ -1801,6 +1801,16 @@ disallow_untyped_defs = true
 warn_return_any = true
 warn_unreachable = true
 
+[mypy-homeassistant.components.local_todo.*]
+check_untyped_defs = true
+disallow_incomplete_defs = true
+disallow_subclassing_any = true
+disallow_untyped_calls = true
+disallow_untyped_decorators = true
+disallow_untyped_defs = true
+warn_return_any = true
+warn_unreachable = true
+
 [mypy-homeassistant.components.lock.*]
 check_untyped_defs = true
 disallow_incomplete_defs = true