The_miro
/
SmartestHome
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
SmartestHome/chores/check.py

618 lines
28 KiB
Python
Raw Permalink Blame History

"""chores — camera-verified household task distribution, from
docs/project-plan.md Phase 20.
THE PRINCIPLE, stated once because it drives every design choice below: **"I don't
care who does it, as long as it gets done."** This is not an assignment system that
picks one fair person and waits on them — it's a nudge system that keeps redirecting
to whoever is actually around until the chore is done. Fairness is tracked for
comparison, not enforced by the algorithm — see "The tally is passive" below.
Runs every ~2 hours (systemd timer, `RandomizedDelaySec=1800` gives the "+/-30 min
in case something else is running" jitter) and does three things, in order:
1. **Trash-day-eve**: reads the same personal collection-date ICS feed
`trash-calendar` already reads (`WASTE_ICS_URL`, own env file, own read — this
script never writes to the calendar, `trash-calendar` owns that). If pickup is
tomorrow and no open "trash" chore exists yet, creates one.
2. **Camera checks** (optional, off unless `CAMERA_WATCHPOINTS` is configured — no
camera hardware has been chosen yet, see docs/project-plan.md §1.18): for each
configured watch point (a Frigate camera + optional PTZ preset), grabs a
snapshot via Frigate's own API and asks an Ollama vision model whether it shows
a full bin / dirty dishes. A "needs attention" result opens a chore if one isn't
already open; a "clear" result auto-closes one if it was.
3. **Nudging**: for every open chore, ASAP, not "wait for a schedule" — the first
run after a chore opens nudges whoever `identity` reports home right now (using
`identity`'s own room field to prefer someone actually near the relevant spot,
e.g. the kitchen for dishes, when that data is available). If the chore is
still open `NEGLECT_THRESHOLD_HOURS` after the last nudge, the household
calendar is checked for anyone currently marked busy (a lightweight, honest-
about-its-limits check — see `_household_currently_busy()`), and if nobody's in
a flagged busy window, the next available person — **preferring someone
different from who was last nudged**, "the next person that walks by" — gets
nudged instead. There is no fixed "assignee" who owns a chore; `last_nudged`
just remembers who to avoid re-nagging immediately and who to give tally credit
to if the chore resolves shortly after.
`identity`'s `chore_exempt` flag (set via `POST /people/<id>/chore-settings`, see
identity/README.md) takes a person out of the nudge rotation entirely — a frequent
guest who isn't a household member doesn't owe chores. **`litter` is the one
exception** (`_EXEMPTIONS_DONT_APPLY` below): everyone, exempt or not, still gets
told to put trash they left out into the bin — that isn't "doing a chore," it's
cleaning up after yourself.
## Assignment is a preference, not a lock
`identity`'s per-person `chore_assignments` (set in its admin panel, read off the same
`/presence` call as everything else here) says who *owes* a given chore type. When
someone assigned to a chore is home, they get nudged instead of whoever happens to be
nearest — that's the whole point of assigning it.
**But an assignee who isn't home doesn't block the chore.** The stated principle above
is "I don't care who does it, as long as it gets done," so if nobody assigned is
around, the nudge falls through to the ordinary whoever's-here rotation rather than
waiting. Set `CHORE_ASSIGNMENT_STRICT=true` if you'd rather it wait for the assignee —
that's the honest opposite reading of the same feature, and which one a household
wants isn't something this file can decide for it.
**`litter` ignores assignment entirely** (`_ASSIGNMENTS_DONT_APPLY`), for the same
reason it ignores exemptions: it goes to whoever left the mess, and "cleaning up after
yourself" was never a task anyone could be assigned in the first place.
`identity`'s `chore_reminder_style` free-text field (same endpoint) is passed to an
LLM that **phrases** the ntfy message in that person's preferred tone ("be
assertive," "be gentle, give me a few minutes of grace") — see `_compose_message()`.
This is wording only. It never touches who gets nudged or when, which stays
presence/calendar-driven per the principle above; if `OLLAMA_TEXT_MODEL` is unset,
the style is empty, or the call fails for any reason, the plain deterministic
template is used verbatim — same "never build against a guess" fallback discipline
as the vision checks above.
## The tally is passive
`chore_events`' `nudged` rows are the only record kept of who got asked about what.
**Nothing in the nudge logic above reads this tally to decide who to ask next** —
that decision is presence/calendar-driven only, exactly per the stated principle.
The tally exists purely so a household member can look at the numbers later and
judge fairness for themselves ("comparison for fairness's sake," not an automated
fairness algorithm). Attribution when a chore closes is a **best-effort heuristic**
(whoever was nudged most recently before it resolved) — nobody here actually
confirms who did it, see "What's not built."
NEVER auto-completes a chore from the nudge step itself — only a fresh camera check
(finding the watch point clear) or a manual close (not built here, see README.md)
ever marks one done. A nudge firing is not the same as the chore being done;
conflating the two would let a repeatedly-nudged, never-actually-done chore quietly
disappear.
"""
from __future__ import annotations
import json
import logging
import os
import sqlite3
import sys
import time
import urllib.request
from datetime import date, datetime, timedelta, timezone
from pathlib import Path
LOG = logging.getLogger("chores")
DB_PATH = Path(os.environ.get("CHORES_DB_PATH", "/data/chores.db"))
IDENTITY_URL = os.environ.get("IDENTITY_URL", "").rstrip("/")
IDENTITY_TOKEN = os.environ.get("IDENTITY_TOKEN", "")
FRIGATE_URL = os.environ.get("FRIGATE_URL", "").rstrip("/")
OLLAMA_HOST = os.environ.get("OLLAMA_HOST", "http://llm-host:11434").rstrip("/")
OLLAMA_VISION_MODEL = os.environ.get("OLLAMA_VISION_MODEL", "llava")
# Optional — phrases (never decides) reminders per-person, see _compose_message().
# Unset means every reminder uses the plain template, same as before this existed.
OLLAMA_TEXT_MODEL = os.environ.get("OLLAMA_TEXT_MODEL", "").strip()
NTFY_URL = os.environ.get("NTFY_URL", "").rstrip("/")
NTFY_TOPIC = os.environ.get("NTFY_TOPIC", "")
NEGLECT_THRESHOLD_HOURS = float(os.environ.get("NEGLECT_THRESHOLD_HOURS", "4"))
WASTE_ICS_URL = os.environ.get("WASTE_ICS_URL", "").strip()
# Household calendar busy-check (see _household_currently_busy()) — reuses
# digest-engine's own CALDAV_* credential names for the same "one Nextcloud app
# password" reasoning as trash-calendar, but this is a THIRD, independent read of
# it (never a write) — a lightweight, honestly-scoped check, not per-person
# availability, see README.md's limitation note.
CALDAV_URL = os.environ.get("CALDAV_URL", "").strip()
CALDAV_USERNAME = os.environ.get("CALDAV_USERNAME", "").strip()
CALDAV_PASSWORD = os.environ.get("CALDAV_PASSWORD", "")
CALDAV_VERIFY_TLS = os.environ.get("CALDAV_VERIFY_TLS", "true").strip().lower() == "true"
CALDAV_QUIET_KEYWORDS = [
k.strip().lower() for k in os.environ.get("CALDAV_QUIET_KEYWORDS", "busy,meeting,call,movie,sleep").split(",") if k.strip()
]
_CHORE_PROMPTS = {
"trash": None, # never camera-checked — trash-day-eve driven only, see module docstring
"bin_full": "Look at this photo of a trash/recycling bin. Answer with exactly one word: FULL, PARTIAL, or EMPTY.",
"dishes": "Look at this photo of a kitchen sink/counter area. Answer with exactly one word: DIRTY or CLEAN.",
"litter": (
"Look at this photo of a household area. Is there any trash/garbage/litter left out "
"that does not belong there (not properly disposed of in a bin)? Answer with exactly "
"one word: YES or NO."
),
}
# Which watch points get "who was just seen here" culprit-attribution treatment
# (see _likely_culprit()) instead of the general "whoever's around" nudge — litter
# is specifically about telling whoever left it, not just whoever's nearby now.
_ATTRIBUTE_TO_RECENT_VIEWER = {"litter"}
# Chore types where identity's chore_exempt flag does NOT apply — everyone still
# gets told to clean up litter they left out, exempt household member or not (see
# module docstring). Currently the same set as _ATTRIBUTE_TO_RECENT_VIEWER, but
# they mean different things — one is about attribution, this is about eligibility
# — so they're kept as separate names rather than reusing one for both purposes.
_EXEMPTIONS_DONT_APPLY = {"litter"}
# Chore types that can't be assigned to anyone — see the module docstring's
# "Assignment is a preference, not a lock". Third set with the same one member as the
# two above, and kept separate for the third distinct reason: attribution, then
# eligibility, now assignability. If they ever diverge (a chore that's assignable but
# exempt-proof, say) collapsing them now would be the thing that made that painful.
_ASSIGNMENTS_DONT_APPLY = {"litter"}
# Whether an assigned person who ISN'T home blocks the chore from falling through to
# whoever is. Default false — "as long as it gets done" is the house rule; true makes
# assignment binding instead. See the module docstring.
CHORE_ASSIGNMENT_STRICT = os.environ.get("CHORE_ASSIGNMENT_STRICT", "false").strip().lower() == "true"
def _now() -> str:
return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
def _db() -> sqlite3.Connection:
DB_PATH.parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
return conn
def init_db(conn: sqlite3.Connection) -> None:
conn.executescript(
"""
CREATE TABLE IF NOT EXISTS chores (
id INTEGER PRIMARY KEY,
type TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'open',
assigned_to TEXT,
created_at TEXT NOT NULL,
assigned_at TEXT,
done_at TEXT,
reminder_count INTEGER NOT NULL DEFAULT 0,
last_reminder_at TEXT
);
CREATE TABLE IF NOT EXISTS chore_events (
id INTEGER PRIMARY KEY,
chore_id INTEGER,
event TEXT NOT NULL,
detail TEXT,
created_at TEXT NOT NULL
);
"""
)
def _log_event(conn, chore_id, event, detail="") -> None:
conn.execute(
"INSERT INTO chore_events (chore_id, event, detail, created_at) VALUES (?, ?, ?, ?)",
(chore_id, event, detail, _now()),
)
def _open_chore(conn, chore_type: str):
return conn.execute("SELECT * FROM chores WHERE type = ? AND status = 'open'", (chore_type,)).fetchone()
def _create_chore(conn, chore_type: str) -> int:
cur = conn.execute(
"INSERT INTO chores (type, status, created_at) VALUES (?, 'open', ?)", (chore_type, _now())
)
assert cur.lastrowid is not None
_log_event(conn, cur.lastrowid, "created")
LOG.info("chores: created a new %r chore", chore_type)
return cur.lastrowid
def _close_chore(conn, chore_row) -> None:
conn.execute("UPDATE chores SET status = 'done', done_at = ? WHERE id = ?", (_now(), chore_row["id"]))
_log_event(conn, chore_row["id"], "auto_closed")
LOG.info("chores: %r chore #%d auto-closed (camera check came back clear)", chore_row["type"], chore_row["id"])
# --- 1. Trash-day-eve ------------------------------------------------------------
def check_trash_day(conn) -> None:
if not WASTE_ICS_URL:
return
from icalendar import Calendar as ICalendar
try:
req = urllib.request.Request(WASTE_ICS_URL, headers={"User-Agent": "smartesthome-chores/1"})
with urllib.request.urlopen(req, timeout=30) as resp:
raw = resp.read()
tomorrow = date.today() + timedelta(days=1)
pickup_tomorrow = False
for component in ICalendar.from_ical(raw).walk("VEVENT"):
dtstart = component.get("dtstart")
if dtstart is None:
continue
start = dtstart.dt
start_date = start.date() if isinstance(start, datetime) else start
if start_date == tomorrow:
pickup_tomorrow = True
break
except Exception:
LOG.warning("chores: could not check WASTE_ICS_URL for trash-day-eve", exc_info=True)
return
if pickup_tomorrow and _open_chore(conn, "trash") is None:
_create_chore(conn, "trash")
# --- 2. Camera checks --------------------------------------------------------------
def _watchpoints() -> list[tuple[str, str, str | None]]:
"""CAMERA_WATCHPOINTS format: "type:camera[:preset],type:camera[:preset],...".
type must be a key in _CHORE_PROMPTS other than "trash" (bin_full, dishes).
"""
raw = os.environ.get("CAMERA_WATCHPOINTS", "").strip()
if not raw:
return []
points = []
for entry in raw.split(","):
parts = [p.strip() for p in entry.split(":")]
if len(parts) < 2 or parts[0] not in _CHORE_PROMPTS or parts[0] == "trash":
LOG.warning("chores: ignoring malformed CAMERA_WATCHPOINTS entry %r", entry)
continue
points.append((parts[0], parts[1], parts[2] if len(parts) > 2 else None))
return points
def _frigate_snapshot(camera: str, preset: str | None) -> bytes | None:
# VERIFY: Frigate's PTZ-move-to-preset API shape is assumed, not confirmed
# against a real Frigate PTZ camera — see README.md. A failure here just means
# the snapshot is taken from wherever the camera already was, not a hard error.
if preset:
try:
move_req = urllib.request.Request(
f"{FRIGATE_URL}/api/{camera}/ptz/move/{preset}", method="POST"
)
urllib.request.urlopen(move_req, timeout=10).close()
time.sleep(3) # give the camera time to physically move before snapshotting
except Exception:
LOG.warning("chores: could not move camera %r to preset %r (continuing anyway)", camera, preset, exc_info=True)
try:
with urllib.request.urlopen(f"{FRIGATE_URL}/api/{camera}/latest.jpg", timeout=15) as resp:
return resp.read()
except Exception:
LOG.warning("chores: could not fetch snapshot for camera %r", camera, exc_info=True)
return None
def _ask_vision(image: bytes, prompt: str) -> str | None:
import base64
payload = {
"model": OLLAMA_VISION_MODEL,
"prompt": prompt,
"images": [base64.b64encode(image).decode("ascii")],
"stream": False,
"options": {"temperature": 0.1},
}
try:
req = urllib.request.Request(
f"{OLLAMA_HOST}/api/generate", data=json.dumps(payload).encode(), headers={"Content-Type": "application/json"}
)
with urllib.request.urlopen(req, timeout=90) as resp:
result = json.loads(resp.read())
return (result.get("response") or "").strip().upper()
except Exception:
LOG.warning("chores: Ollama vision call failed", exc_info=True)
return None
def check_cameras(conn) -> None:
for chore_type, camera, preset in _watchpoints():
if not FRIGATE_URL:
LOG.warning("chores: CAMERA_WATCHPOINTS configured but FRIGATE_URL is unset, skipping")
return
image = _frigate_snapshot(camera, preset)
if image is None:
continue
answer = _ask_vision(image, _CHORE_PROMPTS[chore_type])
if answer is None:
continue
needs_attention = ("FULL" in answer and "PARTIAL" not in answer) or "DIRTY" in answer or (
chore_type == "litter" and "YES" in answer
)
existing = _open_chore(conn, chore_type)
if needs_attention and existing is None:
_create_chore(conn, chore_type)
elif not needs_attention and existing is not None:
_close_chore(conn, existing)
# --- 3. Nudging — presence/calendar-driven, "whoever's around," see module docstring
def _presence() -> list[dict]:
"""Each dict: {"name", "speak_name", "nickname", "home", "room",
"face_seen_recently", "chore_exempt", "chore_reminder_style", "chore_assignments"}
— the raw shape identity's own /presence returns. Empty list (never raises past
this point) if identity is unreachable — nudging just waits for the next run.
Everything this module needs about a person arrives in this one call, including
who's assigned what; there's no second lookup per person. (`_anyone_assigned()`
does make one extra call, but only in strict mode and only about a chore type, not
a person.)
"""
if not IDENTITY_URL or not IDENTITY_TOKEN:
return []
try:
req = urllib.request.Request(f"{IDENTITY_URL}/presence")
req.add_header("Authorization", f"Bearer {IDENTITY_TOKEN}")
with urllib.request.urlopen(req, timeout=10) as resp:
data = json.loads(resp.read())
return [p for p in data.get("people", []) if p.get("home")]
except Exception:
LOG.warning("chores: could not reach identity for /presence", exc_info=True)
return []
def _household_currently_busy() -> bool:
"""A deliberately lightweight check, not per-person availability: is there a
CURRENTLY ACTIVE event on the one shared household calendar whose summary
contains one of CALDAV_QUIET_KEYWORDS? If so, nudges wait for the next run
rather than interrupting a meeting/movie/whoever's asleep. This is a real
limitation, not a bug: it can't tell that only ONE person is busy and nudge
someone else anyway — everyone's nudges pause together. See README.md.
Fails open (returns False, i.e. "not busy") on any error — a broken calendar
check must never be the reason chores stop getting nudged at all.
"""
if not (CALDAV_URL and CALDAV_USERNAME and CALDAV_PASSWORD) or not CALDAV_QUIET_KEYWORDS:
return False
try:
import caldav as caldav_lib
now = datetime.now(timezone.utc)
with caldav_lib.DAVClient(
url=CALDAV_URL, username=CALDAV_USERNAME, password=CALDAV_PASSWORD, ssl_verify_cert=CALDAV_VERIFY_TLS
) as client:
for calendar in client.principal().calendars():
for item in calendar.search(start=now, end=now + timedelta(minutes=1), event=True, expand=True):
summary = str(item.icalendar_component.get("summary", "")).lower()
if any(keyword in summary for keyword in CALDAV_QUIET_KEYWORDS):
return True
except Exception:
LOG.warning("chores: calendar busy-check failed, proceeding as not-busy", exc_info=True)
return False
return False
def _anyone_assigned(home: list[dict], chore_type: str) -> bool:
"""Is this chore type assigned to ANYONE — including people who aren't home?
Only consulted in CHORE_ASSIGNMENT_STRICT mode, and only once the cheaper check
(is an assignee home?) has already come back empty, which is why it's allowed to
cost an extra request. The distinction it draws matters: strict mode should wait
for an absent assignee, but must not wait forever on a chore nobody was ever
assigned — that would silently stop unassigned chores from being nudged at all.
Fails OPEN (returns False, i.e. "nobody's assigned, go ahead and nudge whoever's
around") if identity can't be reached — same rule as the calendar busy-check: a
broken lookup must never be the reason chores stop getting done.
"""
if any(chore_type in (p.get("chore_assignments") or []) for p in home):
return True
if not IDENTITY_URL or not IDENTITY_TOKEN:
return False
try:
req = urllib.request.Request(f"{IDENTITY_URL}/chore-assignments")
req.add_header("Authorization", f"Bearer {IDENTITY_TOKEN}")
with urllib.request.urlopen(req, timeout=10) as resp:
data = json.loads(resp.read())
return bool((data.get("assignments") or {}).get(chore_type))
except Exception:
LOG.warning("chores: could not reach identity for /chore-assignments", exc_info=True)
return False
def _likely_culprit(candidates: list[dict]) -> dict | None:
"""For litter-type chores: prefer whoever was MOST RECENTLY seen by camera face
recognition — a best-effort "who was just here" guess, not a certainty. Falls
back to None (caller then treats it like any other chore) if nobody's
face_seen_recently.
"""
seen = [p for p in candidates if p.get("face_seen_recently")]
return seen[0] if seen else None
def nudge_open_chores(conn) -> None:
open_chores = conn.execute("SELECT * FROM chores WHERE status = 'open'").fetchall()
if not open_chores:
return
home = _presence()
if not home:
LOG.info("chores: %d open chore(s), but nobody is home yet — nothing to do", len(open_chores))
return
if _household_currently_busy():
LOG.info("chores: household calendar shows a busy/quiet window right now — skipping nudges this run")
return
for chore in open_chores:
last_nudged = chore["assigned_to"]
last_nudged_at = chore["assigned_at"]
if last_nudged_at is not None:
elapsed_hours = (datetime.now(timezone.utc) - datetime.fromisoformat(last_nudged_at.replace("Z", "+00:00"))).total_seconds() / 3600
if elapsed_hours < NEGLECT_THRESHOLD_HOURS:
continue # not neglected yet — leave whoever was last nudged alone for now
# chore_exempt people are out of the rotation entirely, EXCEPT litter — see
# _EXEMPTIONS_DONT_APPLY's module-level comment.
eligible = home if chore["type"] in _EXEMPTIONS_DONT_APPLY else [p for p in home if not p.get("chore_exempt")]
if not eligible:
LOG.info(
"chores: %r chore #%d open, but everyone home right now is chore_exempt — skipping",
chore["type"], chore["id"],
)
continue
# Whoever's actually been assigned this chore type in identity's admin panel,
# and is home right now — see the module docstring's "Assignment is a
# preference, not a lock" for why an empty list here doesn't stop the nudge.
assigned = (
[]
if chore["type"] in _ASSIGNMENTS_DONT_APPLY
else [p for p in eligible if chore["type"] in (p.get("chore_assignments") or [])]
)
if assigned:
eligible = assigned
elif CHORE_ASSIGNMENT_STRICT and _anyone_assigned(home, chore["type"]):
LOG.info(
"chores: %r chore #%d is assigned, but no assignee is home and "
"CHORE_ASSIGNMENT_STRICT is on — waiting rather than redirecting",
chore["type"], chore["id"],
)
continue
# "The next person that walks by": prefer someone home right now who ISN'T
# who we last nudged (a real redirect, not the same person nagged again) —
# falls back to re-nudging the same person if they're genuinely the only
# one home. Litter gets a different preference first: whoever the camera
# most recently saw, since the point there is telling the actual culprit.
target = None
if chore["type"] in _ATTRIBUTE_TO_RECENT_VIEWER:
target = _likely_culprit(eligible)
if target is None:
different = [p for p in eligible if p["name"] != last_nudged]
target = (different or eligible)[0]
# identity's speak_name is ALWAYS the person's real name, never a nickname
# they've been given — see identity/server.py's module docstring. Reminders go
# out as text and get read aloud by whatever's showing them, so this is one of
# the consumers that rule exists for. Falls back to `name` for an identity
# older than the nickname feature.
name = target.get("speak_name") or target["name"]
conn.execute(
"UPDATE chores SET assigned_to = ?, assigned_at = ?, reminder_count = reminder_count + 1, "
"last_reminder_at = ? WHERE id = ?",
(name, _now(), _now(), chore["id"]),
)
redirected = last_nudged is not None and last_nudged != name
_log_event(conn, chore["id"], "nudged", name)
redirect_note = f" (redirected from {last_nudged})" if redirected and last_nudged else ""
LOG.info("chores: nudged %s about %r chore #%d%s", name, chore["type"], chore["id"], redirect_note)
_notify(
name,
chore["type"],
redirected,
is_culprit=chore["type"] in _ATTRIBUTE_TO_RECENT_VIEWER,
reminder_style=target.get("chore_reminder_style"),
)
def _default_message(name: str, chore_type: str, redirected: bool, is_culprit: bool) -> str:
if is_culprit:
return f"{name}, looks like something was left out — could you put it in the bin?"
if redirected:
return f"{name}, this one's still open — could you take care of: {chore_type}?"
return f"{name}, could you take care of: {chore_type}?"
def _compose_message(name: str, chore_type: str, redirected: bool, is_culprit: bool, reminder_style: str | None) -> str:
"""Wording only — see module docstring's OLLAMA_TEXT_MODEL paragraph. Falls back
to _default_message() verbatim whenever there's no model configured, no style
set for this person, or the call fails/returns nothing — never blocks a nudge
from going out over an LLM hiccup.
"""
fallback = _default_message(name, chore_type, redirected, is_culprit)
if not OLLAMA_TEXT_MODEL or not reminder_style:
return fallback
prompt = (
f"Write ONE short household chore reminder (max 2 sentences) addressed to {name}. "
f"The chore is: {chore_type}. "
f"Follow {name}'s own stated preference for how they like to be reminded: \"{reminder_style}\". "
+ ("Something was left out and may belong to them specifically — ask them to put it away, don't accuse them outright. "
if is_culprit else "")
+ ("They were already asked about this once before and it's still not done. " if redirected else "")
+ "Reply with ONLY the message text itself — no preamble, no quotation marks."
)
try:
payload = {"model": OLLAMA_TEXT_MODEL, "prompt": prompt, "stream": False, "options": {"temperature": 0.4}}
req = urllib.request.Request(
f"{OLLAMA_HOST}/api/generate", data=json.dumps(payload).encode(), headers={"Content-Type": "application/json"}
)
with urllib.request.urlopen(req, timeout=30) as resp:
result = json.loads(resp.read())
text = (result.get("response") or "").strip()
return text if text else fallback
except Exception:
LOG.warning("chores: Ollama message-phrasing call failed, using plain template", exc_info=True)
return fallback
def _notify(name: str, chore_type: str, redirected: bool, is_culprit: bool, reminder_style: str | None = None) -> None:
if not NTFY_URL or not NTFY_TOPIC:
return
message = _compose_message(name, chore_type, redirected, is_culprit, reminder_style)
try:
req = urllib.request.Request(f"{NTFY_URL}/{NTFY_TOPIC}", data=message.encode("utf-8"), method="POST")
urllib.request.urlopen(req, timeout=10).close()
except Exception:
LOG.warning("chores: ntfy notification failed", exc_info=True)
def print_tally(conn) -> None:
"""Read-only reporting, per the module docstring — never feeds back into
nudge_open_chores()'s choice of who to nudge next. Logged, not served
anywhere yet — see README.md's "What's not built."
"""
rows = conn.execute(
"SELECT detail AS name, COUNT(*) AS n FROM chore_events "
"WHERE event = 'nudged' AND created_at >= ? GROUP BY detail ORDER BY n DESC",
((datetime.now(timezone.utc) - timedelta(days=30)).isoformat(),),
).fetchall()
if rows:
LOG.info("chores: 30-day nudge tally (for comparison only, not used to pick who's next): %s",
", ".join(f"{r['name']}: {r['n']}" for r in rows))
def main() -> int:
logging.basicConfig(
level=os.environ.get("LOG_LEVEL", "INFO").upper(),
format="%(asctime)s %(levelname)-7s %(name)s: %(message)s",
)
conn = _db()
try:
init_db(conn)
check_trash_day(conn)
check_cameras(conn)
conn.commit()
nudge_open_chores(conn)
conn.commit()
print_tally(conn)
finally:
conn.close()
return 0
if __name__ == "__main__":
sys.exit(main())
Reference in New Issue View Git Blame Copy Permalink