Add plant care 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`.
 
+### Plant Care Domain Demo
+
+This repository also includes a plant care domain adaptation that recommends indoor plant care actions 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 plant care demo:
+
+```bash
+streamlit run plant_recommender.py
+```
+
+Run the fallback CLI:
+
+```bash
+python plant_recommender.py
+```
+
+Run the plant care domain tests:
+
+```bash
+python -m unittest tests/test_plant_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 plant care demo follows the same continuous-feedback pattern with a different domain. It encodes each care action into the same eight-bit AO-compatible input shape using plant type, available light, effort, and the user's current care 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__PlantRecommender.py

@@ -0,0 +1,15 @@
+# -*- coding: utf-8 -*-
+"""AO architecture for the plant care recommender domain."""
+
+import ao_arch as ar
+
+
+description = "Plant Care Recommender System"
+
+# plant_type, light level, high effort flag, care 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)

plant_domain.py

@@ -0,0 +1,150 @@
+"""Plant care recommendation domain for the AO recommender demo."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Iterable
+
+
+PLANT_TYPE_BITS = {
+    "succulent": [0, 0, 0],
+    "tropical": [0, 0, 1],
+    "herb": [0, 1, 0],
+    "flowering": [0, 1, 1],
+    "fern": [1, 0, 0],
+    "tree": [1, 0, 1],
+}
+
+LIGHT_BITS = {
+    "low": [0, 0],
+    "medium": [0, 1],
+    "bright": [1, 1],
+}
+
+GOAL_BITS = {
+    "easy-care": [0, 0],
+    "faster-growth": [0, 1],
+    "fix-symptoms": [1, 0],
+    "harvest": [1, 1],
+}
+
+
+@dataclass(frozen=True)
+class PlantCareAction:
+    """A plant care action that can be recommended."""
+
+    name: str
+    plant_type: str
+    light: str
+    effort: str
+    water_need: str
+    minutes: int
+    contexts: tuple[str, ...]
+    description: str
+
+
+PLANT_ACTIONS: tuple[PlantCareAction, ...] = (
+    PlantCareAction("Check soil moisture before watering", "succulent", "bright", "low", "low", 5, ("easy-care", "fix-symptoms"), "Avoid overwatering by watering only when the top soil is dry."),
+    PlantCareAction("Rotate the pot toward even light", "tropical", "medium", "low", "medium", 5, ("easy-care", "faster-growth"), "Turn the plant so new growth stays balanced instead of leaning toward a window."),
+    PlantCareAction("Trim yellowing leaves", "tropical", "medium", "low", "medium", 10, ("fix-symptoms", "easy-care"), "Remove damaged foliage so the plant directs energy to healthy growth."),
+    PlantCareAction("Prune basil above a leaf pair", "herb", "bright", "low", "medium", 10, ("harvest", "faster-growth"), "Harvest in a way that encourages branching and delays flowering."),
+    PlantCareAction("Mist and inspect fern fronds", "fern", "low", "medium", "high", 15, ("fix-symptoms", "faster-growth"), "Raise local humidity and look for crispy edges or pest pressure."),
+    PlantCareAction("Deadhead spent blooms", "flowering", "bright", "medium", "medium", 20, ("faster-growth", "easy-care"), "Remove fading flowers so the plant can keep producing new buds."),
+    PlantCareAction("Top dress with compost", "tree", "bright", "medium", "medium", 25, ("faster-growth", "easy-care"), "Add a light nutrient boost without disturbing the root ball."),
+    PlantCareAction("Flush mineral salts from soil", "tropical", "bright", "high", "high", 30, ("fix-symptoms", "faster-growth"), "Run water through the pot to remove fertilizer salts that can burn leaf tips."),
+    PlantCareAction("Pinch mint runners back", "herb", "medium", "low", "medium", 10, ("harvest", "easy-care"), "Keep growth compact and encourage fresh edible leaves."),
+    PlantCareAction("Move succulent closer to bright light", "succulent", "bright", "medium", "low", 15, ("fix-symptoms", "faster-growth"), "Reduce stretching by giving the plant stronger indirect light."),
+    PlantCareAction("Check drainage and saucer water", "flowering", "medium", "low", "medium", 8, ("fix-symptoms", "easy-care"), "Prevent root rot by clearing standing water after watering."),
+    PlantCareAction("Stake a leaning indoor tree", "tree", "medium", "medium", "medium", 20, ("fix-symptoms", "easy-care"), "Support the stem while you correct light direction and watering rhythm."),
+)
+
+
+def encode_plant_action(action: PlantCareAction, goal: str = "easy-care") -> list[int]:
+    """Encode a plant care action plus user goal into the AO eight-bit input shape."""
+
+    if action.plant_type not in PLANT_TYPE_BITS:
+        raise ValueError(f"Unknown plant type: {action.plant_type}")
+    if action.light not in LIGHT_BITS:
+        raise ValueError(f"Unknown light level: {action.light}")
+    if goal not in GOAL_BITS:
+        raise ValueError(f"Unknown goal: {goal}")
+
+    high_effort_bit = [1 if action.effort == "high" else 0]
+    return PLANT_TYPE_BITS[action.plant_type] + LIGHT_BITS[action.light] + high_effort_bit + GOAL_BITS[goal]
+
+
+def score_plant_action(
+    action: PlantCareAction,
+    goal: str,
+    minutes_available: int,
+    indoor_light: str = "medium",
+    feedback: dict[str, int] | None = None,
+) -> int:
+    """Score a plant action with deterministic plant-care preferences and feedback."""
+
+    score = 0
+    if goal in action.contexts:
+        score += 35
+    if action.minutes <= minutes_available:
+        score += 20
+    else:
+        score -= (action.minutes - minutes_available) // 5 * 4
+    if action.light == indoor_light:
+        score += 16
+    elif indoor_light == "bright" and action.light == "medium":
+        score += 8
+    if goal == "easy-care" and action.effort == "low":
+        score += 14
+    if goal == "fix-symptoms":
+        score += 12
+    if goal == "harvest" and action.plant_type == "herb":
+        score += 18
+    if action.water_need == "low" and goal == "easy-care":
+        score += 8
+    if feedback:
+        score += feedback.get(action.name, 0) * 12
+    return score
+
+
+def recommend_plant_actions(
+    goal: str = "easy-care",
+    minutes_available: int = 15,
+    indoor_light: str = "medium",
+    feedback: dict[str, int] | None = None,
+    actions: Iterable[PlantCareAction] = PLANT_ACTIONS,
+    limit: int = 5,
+) -> list[tuple[PlantCareAction, int]]:
+    """Return plant care recommendations sorted from strongest to weakest match."""
+
+    if goal not in GOAL_BITS:
+        raise ValueError(f"Unknown goal: {goal}")
+    if indoor_light not in LIGHT_BITS:
+        raise ValueError(f"Unknown light level: {indoor_light}")
+
+    ranked = [
+        (
+            action,
+            score_plant_action(
+                action,
+                goal=goal,
+                minutes_available=minutes_available,
+                indoor_light=indoor_light,
+                feedback=feedback,
+            ),
+        )
+        for action in actions
+    ]
+    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,
+    action_name: str,
+    liked: bool,
+) -> dict[str, int]:
+    """Return updated feedback weights for a plant care action."""
+
+    updated = dict(feedback or {})
+    updated[action_name] = updated.get(action_name, 0) + (1 if liked else -1)
+    return updated

plant_recommender.py

@@ -0,0 +1,165 @@
+"""Streamlit and CLI demo for a plant care recommender domain."""
+
+from __future__ import annotations
+
+from plant_domain import (
+    PLANT_ACTIONS,
+    apply_feedback,
+    encode_plant_action,
+    recommend_plant_actions,
+)
+
+
+def create_agent():
+    """Create an AO Agent when optional AO packages are installed."""
+
+    try:
+        import ao_core as ao
+        from arch__PlantRecommender import arch
+    except Exception:
+        return None
+
+    agent = ao.Agent(arch, notes="Plant Care 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 plant care recommendations:")
+    for action, score in recommend_plant_actions(goal="easy-care", minutes_available=15):
+        print(f"- {action.name} ({score})")
+        print(f"  input={encode_plant_action(action, 'easy-care')}")
+        print(f"  {action.description}")
+
+
+def run_streamlit() -> None:
+    """Run the interactive Streamlit demo."""
+
+    import streamlit as st
+
+    st.set_page_config(
+        page_title="Plant Care Recommender by AO Labs",
+        page_icon="misc/ao_favicon.png",
+        layout="wide",
+        initial_sidebar_state="expanded",
+    )
+
+    if "plant_feedback" not in st.session_state:
+        st.session_state.plant_feedback = {}
+    if "plant_agent" not in st.session_state:
+        st.session_state.plant_agent = create_agent()
+
+    st.title("Plant Care Recommender")
+    st.write("A domain adaptation of the AO recommender for indoor plant care actions.")
+
+    with st.sidebar:
+        goal = st.selectbox(
+            "Care goal",
+            ("easy-care", "faster-growth", "fix-symptoms", "harvest"),
+            index=0,
+            format_func=lambda value: value.replace("-", " ").title(),
+        )
+        indoor_light = st.selectbox(
+            "Available light",
+            ("low", "medium", "bright"),
+            index=1,
+            format_func=str.title,
+        )
+        minutes_available = st.slider("Minutes available today", 5, 45, 15, 5)
+        ao_status = "available" if st.session_state.plant_agent is not None else "fallback mode"
+        st.write(f"AO Agent: {ao_status}")
+
+    ranked = recommend_plant_actions(
+        goal=goal,
+        minutes_available=minutes_available,
+        indoor_light=indoor_light,
+        feedback=st.session_state.plant_feedback,
+        limit=len(PLANT_ACTIONS),
+    )
+
+    for action, fallback_score in ranked[:5]:
+        binary_input = encode_plant_action(action, goal)
+        ao_score = ao_percentage(st.session_state.plant_agent, binary_input)
+        display_score = ao_score if ao_score is not None else fallback_score
+
+        st.subheader(action.name)
+        st.write(action.description)
+        st.write(
+            {
+                "plant_type": action.plant_type,
+                "light": action.light,
+                "effort": action.effort,
+                "water_need": action.water_need,
+                "minutes": action.minutes,
+                "encoded_input": binary_input,
+                "score": display_score,
+            }
+        )
+
+        left, right = st.columns(2)
+        if left.button("Recommend more like this", key=f"like-{action.name}"):
+            st.session_state.plant_feedback = apply_feedback(
+                st.session_state.plant_feedback, action.name, liked=True
+            )
+            train_agent(st.session_state.plant_agent, binary_input, liked=True)
+            st.rerun()
+        if right.button("Recommend less like this", key=f"less-{action.name}"):
+            st.session_state.plant_feedback = apply_feedback(
+                st.session_state.plant_feedback, action.name, liked=False
+            )
+            train_agent(st.session_state.plant_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()