Add home maintenance recommender domain

README.md

@@ -31,6 +31,28 @@ If you plan to run the app in a conda or virtual environment, make sure to set u
 
 4. Once running, the app will be accessible at `localhost:8501`.
 
+### Home Maintenance Domain Demo
+
+This repository also includes a home maintenance domain adaptation that recommends household upkeep tasks from a local catalog. It can use AO packages when installed, and it includes a deterministic fallback so reviewers can run it without private packages or paid APIs.
+
+Run the home maintenance demo:
+
+```bash
+streamlit run home_recommender.py
+```
+
+Run the fallback CLI:
+
+```bash
+python home_recommender.py
+```
+
+Run the home maintenance domain tests:
+
+```bash
+python -m unittest tests/test_home_domain.py
+```
+
 
 ### Docker Installation
 
@@ -55,10 +77,11 @@ You're done! Access the app at `localhost:8501` in your browser.
 
 The recommender system works by loading a set of random video links. Once the user hits the Run button, a video will be shown, and the system will suggest whether it recommends the video or not. The user can then provide feedback using "pain" or "pleasure" signals to guide the recommendation process. Based on this feedback, the system adjusts its responses and suggests another video. This cycle continues, allowing for more accurate and personalized recommendations over time.
 
+The home maintenance demo follows the same continuous-feedback pattern with a different domain. It encodes each task into the same eight-bit AO-compatible input shape using category, urgency, effort, and the user's current household goal. User feedback updates fallback rankings immediately and trains an AO Agent when optional AO packages are available.
+
 
 ## Contributing
 
 Fork the repository, make your changes, and submit a pull request for review. 
 
 
-

arch__HomeRecommender.py

@@ -0,0 +1,15 @@
+# -*- coding: utf-8 -*-
+"""AO architecture for the home maintenance recommender domain."""
+
+import ao_arch as ar
+
+
+description = "Home Maintenance Recommender"
+
+# category + urgency + high effort + household goal
+arch_i = [3, 2, 1, 2]
+arch_z = [10]
+arch_c = []
+connector_function = "full_conn"
+
+arch = ar.Arch(arch_i, arch_z, arch_c, connector_function, description)

home_domain.py

@@ -0,0 +1,148 @@
+"""Home maintenance recommendation domain for the AO recommender demo."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Iterable
+
+
+CATEGORY_BITS = {
+    "safety": [0, 0, 0],
+    "energy": [0, 0, 1],
+    "plumbing": [0, 1, 0],
+    "appliance": [0, 1, 1],
+    "cleaning": [1, 0, 0],
+    "seasonal": [1, 0, 1],
+}
+
+URGENCY_BITS = {
+    "routine": [0, 0],
+    "soon": [0, 1],
+    "urgent": [1, 1],
+}
+
+GOAL_BITS = {
+    "safety": [0, 0],
+    "save-money": [0, 1],
+    "prevent-breakdowns": [1, 0],
+    "clean-comfort": [1, 1],
+}
+
+
+@dataclass(frozen=True)
+class HomeTask:
+    """A household upkeep task that can be recommended."""
+
+    name: str
+    category: str
+    urgency: str
+    effort: str
+    cost_tier: str
+    minutes: int
+    contexts: tuple[str, ...]
+    description: str
+
+
+HOME_TASKS: tuple[HomeTask, ...] = (
+    HomeTask("Test smoke and carbon monoxide alarms", "safety", "routine", "low", "free", 15, ("safety", "prevent-breakdowns"), "Confirm alarms work and replace weak batteries before they become an emergency."),
+    HomeTask("Replace HVAC filter", "energy", "soon", "low", "low", 20, ("save-money", "prevent-breakdowns"), "Improve airflow, reduce energy waste, and protect the heating or cooling system."),
+    HomeTask("Flush sediment from the water heater", "plumbing", "soon", "medium", "free", 45, ("prevent-breakdowns", "save-money"), "Drain sediment so the heater runs efficiently and is less likely to fail early."),
+    HomeTask("Clean refrigerator condenser coils", "appliance", "routine", "medium", "free", 35, ("save-money", "prevent-breakdowns"), "Remove dust from the coils so the compressor does not work harder than needed."),
+    HomeTask("Seal drafts around doors and windows", "energy", "soon", "medium", "low", 60, ("save-money", "clean-comfort"), "Use weatherstripping or caulk to reduce drafts and stabilize room temperature."),
+    HomeTask("Inspect under-sink plumbing for leaks", "plumbing", "routine", "low", "free", 20, ("safety", "prevent-breakdowns"), "Catch slow leaks before they damage cabinets, flooring, or walls."),
+    HomeTask("Deep-clean dryer lint path", "safety", "urgent", "medium", "free", 40, ("safety", "save-money"), "Clear lint from the trap, hose, and vent path to reduce fire risk and improve drying time."),
+    HomeTask("Descale shower heads and faucets", "cleaning", "routine", "low", "low", 30, ("clean-comfort", "save-money"), "Remove mineral buildup to improve water flow and keep fixtures looking fresh."),
+    HomeTask("Clean gutters before heavy rain", "seasonal", "urgent", "high", "free", 90, ("prevent-breakdowns", "safety"), "Prevent overflow that can damage fascia, siding, foundations, or basements."),
+    HomeTask("Vacuum bathroom exhaust fan grille", "cleaning", "routine", "low", "free", 15, ("clean-comfort", "prevent-breakdowns"), "Restore ventilation so humidity leaves the room more quickly."),
+    HomeTask("Check appliance hoses for bulges or cracks", "appliance", "soon", "low", "free", 20, ("safety", "prevent-breakdowns"), "Inspect washer, dishwasher, and ice-maker lines for early signs of failure."),
+    HomeTask("Build a seasonal maintenance checklist", "seasonal", "routine", "medium", "free", 45, ("prevent-breakdowns", "clean-comfort"), "Create a simple recurring plan so small tasks do not become expensive repairs."),
+)
+
+
+def encode_home_task(task: HomeTask, goal: str = "prevent-breakdowns") -> list[int]:
+    """Encode a home task plus household goal into the AO eight-bit input shape."""
+
+    if task.category not in CATEGORY_BITS:
+        raise ValueError(f"Unknown category: {task.category}")
+    if task.urgency not in URGENCY_BITS:
+        raise ValueError(f"Unknown urgency: {task.urgency}")
+    if goal not in GOAL_BITS:
+        raise ValueError(f"Unknown goal: {goal}")
+
+    high_effort_bit = [1 if task.effort == "high" else 0]
+    return CATEGORY_BITS[task.category] + URGENCY_BITS[task.urgency] + high_effort_bit + GOAL_BITS[goal]
+
+
+def score_home_task(
+    task: HomeTask,
+    goal: str,
+    minutes_available: int,
+    prefer_low_cost: bool = True,
+    feedback: dict[str, int] | None = None,
+) -> int:
+    """Score a home task with deterministic context preferences and optional feedback."""
+
+    score = 0
+    if goal in task.contexts:
+        score += 35
+    if task.minutes <= minutes_available:
+        score += 20
+    else:
+        score -= (task.minutes - minutes_available) // 5 * 4
+    if task.urgency == "urgent":
+        score += 16
+    elif task.urgency == "soon":
+        score += 10
+    if prefer_low_cost and task.cost_tier in {"free", "low"}:
+        score += 10
+    if goal == "safety" and task.category == "safety":
+        score += 18
+    if goal == "save-money" and task.category in {"energy", "appliance"}:
+        score += 14
+    if goal == "clean-comfort" and task.category == "cleaning":
+        score += 14
+    if feedback:
+        score += feedback.get(task.name, 0) * 12
+    return score
+
+
+def recommend_home_tasks(
+    goal: str = "prevent-breakdowns",
+    minutes_available: int = 45,
+    prefer_low_cost: bool = True,
+    feedback: dict[str, int] | None = None,
+    tasks: Iterable[HomeTask] = HOME_TASKS,
+    limit: int = 5,
+) -> list[tuple[HomeTask, int]]:
+    """Return household upkeep recommendations sorted from strongest to weakest match."""
+
+    if goal not in GOAL_BITS:
+        raise ValueError(f"Unknown goal: {goal}")
+
+    ranked = [
+        (
+            task,
+            score_home_task(
+                task,
+                goal=goal,
+                minutes_available=minutes_available,
+                prefer_low_cost=prefer_low_cost,
+                feedback=feedback,
+            ),
+        )
+        for task in tasks
+    ]
+    ranked.sort(key=lambda item: (item[1], -item[0].minutes, item[0].name), reverse=True)
+    return ranked[:limit]
+
+
+def apply_feedback(
+    feedback: dict[str, int] | None,
+    task_name: str,
+    liked: bool,
+) -> dict[str, int]:
+    """Return updated feedback weights for a home maintenance task."""
+
+    updated = dict(feedback or {})
+    updated[task_name] = updated.get(task_name, 0) + (1 if liked else -1)
+    return updated

home_recommender.py

@@ -0,0 +1,160 @@
+"""Streamlit and CLI demo for a home maintenance recommender domain."""
+
+from __future__ import annotations
+
+from home_domain import (
+    HOME_TASKS,
+    apply_feedback,
+    encode_home_task,
+    recommend_home_tasks,
+)
+
+
+def create_agent():
+    """Create an AO Agent when optional AO packages are installed."""
+
+    try:
+        import ao_core as ao
+        from arch__HomeRecommender import arch
+    except Exception:
+        return None
+
+    agent = ao.Agent(arch, notes="Home Maintenance Agent")
+    for _ in range(4):
+        agent.reset_state()
+        agent.reset_state(training=True)
+    return agent
+
+
+def ao_percentage(agent, binary_input: list[int]) -> int | None:
+    """Return an AO recommendation percentage, or None when AO is unavailable."""
+
+    if agent is None:
+        return None
+
+    agent.reset_state()
+    response = None
+    for _ in range(5):
+        response = agent.next_state(INPUT=binary_input, print_result=False)
+
+    if response is None:
+        return None
+
+    return round(sum(1 for value in response if value == 1) / len(response) * 100)
+
+
+def train_agent(agent, binary_input: list[int], liked: bool) -> None:
+    """Train the optional AO Agent on user feedback."""
+
+    if agent is None:
+        return
+
+    import numpy as np
+
+    label_value = 1 if liked else 0
+    label = np.full(agent.arch.Z__flat.shape, label_value, dtype=np.int8)
+    for _ in range(5 if liked else 10):
+        agent.reset_state()
+        agent.next_state(INPUT=binary_input, LABEL=label, print_result=False, unsequenced=True)
+
+
+def run_cli() -> None:
+    """Print fallback recommendations without requiring Streamlit or AO packages."""
+
+    print("Top home maintenance recommendations:")
+    for task, score in recommend_home_tasks(goal="prevent-breakdowns", minutes_available=45):
+        print(f"- {task.name} ({score})")
+        print(f"  input={encode_home_task(task, 'prevent-breakdowns')}")
+        print(f"  {task.description}")
+
+
+def run_streamlit() -> None:
+    """Run the interactive Streamlit demo."""
+
+    import streamlit as st
+
+    st.set_page_config(
+        page_title="Home Maintenance Recommender by AO Labs",
+        page_icon="misc/ao_favicon.png",
+        layout="wide",
+        initial_sidebar_state="expanded",
+    )
+
+    if "home_feedback" not in st.session_state:
+        st.session_state.home_feedback = {}
+    if "home_agent" not in st.session_state:
+        st.session_state.home_agent = create_agent()
+
+    st.title("Home Maintenance Recommender")
+    st.write("A domain adaptation of the AO recommender for household upkeep and repair prevention.")
+
+    with st.sidebar:
+        goal = st.selectbox(
+            "Household goal",
+            ("safety", "save-money", "prevent-breakdowns", "clean-comfort"),
+            index=2,
+            format_func=lambda value: value.replace("-", " ").title(),
+        )
+        minutes_available = st.slider("Minutes available today", 10, 120, 45, 5)
+        prefer_low_cost = st.checkbox("Prefer free or low-cost tasks", value=True)
+        ao_status = "available" if st.session_state.home_agent is not None else "fallback mode"
+        st.write(f"AO Agent: {ao_status}")
+
+    ranked = recommend_home_tasks(
+        goal=goal,
+        minutes_available=minutes_available,
+        prefer_low_cost=prefer_low_cost,
+        feedback=st.session_state.home_feedback,
+        limit=len(HOME_TASKS),
+    )
+
+    for task, fallback_score in ranked[:5]:
+        binary_input = encode_home_task(task, goal)
+        ao_score = ao_percentage(st.session_state.home_agent, binary_input)
+        display_score = ao_score if ao_score is not None else fallback_score
+
+        st.subheader(task.name)
+        st.write(task.description)
+        st.write(
+            {
+                "category": task.category,
+                "urgency": task.urgency,
+                "effort": task.effort,
+                "cost_tier": task.cost_tier,
+                "minutes": task.minutes,
+                "encoded_input": binary_input,
+                "score": display_score,
+            }
+        )
+
+        left, right = st.columns(2)
+        if left.button("Recommend more like this", key=f"like-{task.name}"):
+            st.session_state.home_feedback = apply_feedback(
+                st.session_state.home_feedback, task.name, liked=True
+            )
+            train_agent(st.session_state.home_agent, binary_input, liked=True)
+            st.rerun()
+        if right.button("Recommend less like this", key=f"less-{task.name}"):
+            st.session_state.home_feedback = apply_feedback(
+                st.session_state.home_feedback, task.name, liked=False
+            )
+            train_agent(st.session_state.home_agent, binary_input, liked=False)
+            st.rerun()
+
+
+def is_streamlit_runtime() -> bool:
+    """Detect whether this script is being executed by Streamlit."""
+
+    try:
+        from streamlit.runtime.scriptrunner import get_script_run_ctx
+    except Exception:
+        return False
+
+    return get_script_run_ctx() is not None
+
+
+if __name__ == "__main__":
+    if is_streamlit_runtime():
+        run_streamlit()
+    else:
+        run_cli()