claude unstested update for moving from wp to project -> wp to task

dmapp/config.py

@@ -1,6 +0,0 @@
-DESKTOPS = ("Work",
-            "Browse",
-            "Write",
-            "Learn",
-            "Idle",
-            "Self")

dmapp/dmcore/config.json

@@ -0,0 +1,10 @@
+{
+  "work_desktop_tasks": {
+    "2": null,
+    "5": null,
+    "6": null,
+    "7": null
+  },
+  "timezone": "America/Argentina/Buenos_Aires",
+  "task_file": "/home/mariano/LETRAS/adm/task/main"
+}

dmapp/dmcore/config.py

@@ -2,6 +2,7 @@
 import logging
 
 from pymongo import MongoClient
+from zoneinfo import ZoneInfo
 
 # Logging configuration
 logging.basicConfig(
@@ -9,7 +10,7 @@ logging.basicConfig(
     format="%(asctime)s %(levelname)-8s [%(filename)s:%(lineno)d] %(message)s",
     datefmt="%Y-%m-%d %H:%M:%S",
 )
-# 2) Get your module’s logger and bump it to DEBUG
+# 2) Get your module's logger and bump it to DEBUG
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.DEBUG)
 
@@ -25,3 +26,9 @@ db = client.deskmeter
 switches = db.switch
 states = db.state
 tasks = db.task
+
+# Application configuration
+desktops = ("Plan", "Think", "Work", "Other", "Away", "Work", "Work", "Work")
+unlabeled = "Away"
+timezone = ZoneInfo("America/Argentina/Buenos_Aires")
+task_file = "/home/mariano/LETRAS/adm/task/main"

dmapp/dmcore/main.py

@@ -1,4 +1,5 @@
 import datetime
+import json
 import os
 import subprocess
 import time
@@ -10,13 +11,92 @@ from config import logger, switches
 from zoneinfo import ZoneInfo
 
 desktops = ("Plan", "Think", "Work", "Other", "Away", "Work", "Work", "Work")
-work_desktops = {2: "snk", 5: "dlt", 6: "vhs", 7: "own"}
-
 unlabeled = "Away"
+config_file = "config.json"
+
+
+def load_config():
+    """Load configuration from JSON file"""
+    with open(config_file, "r") as f:
+        return json.load(f)
+
+
+def reload_config_if_changed():
+    """Check if config file changed and reload to state if needed"""
+    current_config_mtime = state.retrieve("current").get("config_mtime")
+    config_mtime = os.path.getmtime(config_file)
+
+    if current_config_mtime != config_mtime:
+        cfg = load_config()
+        work_desktop_tasks = {int(k): v for k, v in cfg["work_desktop_tasks"].items()}
+        state.sync_desktop_tasks(work_desktop_tasks)
+        state.save("current", config_mtime=config_mtime)
+        logger.info(f"Config reloaded: {work_desktop_tasks}")
 
 
 def now():
-    return datetime.datetime.now(ZoneInfo("America/Argentina/Buenos_Aires"))
+    cfg = load_config()
+    return datetime.datetime.now(ZoneInfo(cfg["timezone"]))
+
+
+def handle_task_file_changes(current_task):
+    """Check if task file changed and update task if needed"""
+    current_mtime = state.retrieve("current").get("filetime")
+    file_mtime = task.get_file_mtime(None)
+
+    if current_mtime != file_mtime:
+        task_id = task.read_and_extract(None)
+        logger.debug(f"task_id:{task_id}")
+        task.file_to_db(None)
+        if task_id != current_task:
+            state.save("current", task=task_id)
+            current_task = task_id
+
+    return current_task
+
+
+def update_workspace_state():
+    """Update current workspace in state"""
+    current_workspace = active_workspace()
+    state.save("current", workspace=current_workspace)
+    return current_workspace
+
+
+def enforce_desktop_task(current_workspace, work_desktop_tasks, current_task):
+    """Enforce assigned task for work desktops"""
+    if current_workspace in work_desktop_tasks and work_desktop_tasks[current_workspace]:
+        assigned_task = work_desktop_tasks[current_workspace]
+
+        if current_task != assigned_task:
+            current_task = assigned_task
+            state.save("current", task=current_task)
+            task.db_to_file_as_is(None)
+
+    return current_task
+
+
+def track_workspace_switch(current_workspace, current_task, last_switch_time):
+    """Update or create switch record"""
+    last_doc = switches.find_one(sort=[("_id", -1)])
+
+    if (
+        last_doc["workspace"] == desktop(current_workspace)
+        and last_doc["task"] == current_task
+    ):
+        delta = round((now() - last_switch_time).total_seconds())
+        switches.update_one(
+            {"_id": last_doc["_id"]}, {"$set": {"delta": delta, "task": current_task}}
+        )
+        return last_switch_time
+    else:
+        switch = {
+            "workspace": desktop(current_workspace),
+            "date": now(),
+            "delta": 0,
+            "task": current_task,
+        }
+        switches.insert_one(switch)
+        return now()
 
 
 def active_workspace():
@@ -40,7 +120,6 @@ def desktop(workspace_index):
 
 
 task.read_and_extract(None)
-state.init_work_state(work_desktops)
 
 
 current_workspace = active_workspace()
@@ -57,65 +136,23 @@ switch = {
 switches.insert_one(switch)
 
 while True:
-    current_mtime = state.retrieve("current").get("filetime")
+    # Check if config changed and reload
-    file_mtime = task.get_file_mtime(None)
+    reload_config_if_changed()
 
-    # First handle file changes
+    # Load work_desktop_tasks from state
-    if current_mtime != file_mtime:
+    work_desktop_tasks = state.retrieve_desktop_state()
-        task_id = task.read_and_extract(None)
-        logger.debug(f"task_id:{task_id}")
-        task.file_to_db(None)
-        if task_id != current_task:  # Only update state if different
-            state.save("current", task=task_id)
-            current_task = task_id
 
+    # Handle task file changes
+    current_task = handle_task_file_changes(current_task)
+
+    # Update current task and workspace
     current_task = state.retrieve("current").get("task")
-    current_workspace = active_workspace()
+    current_workspace = update_workspace_state()
-    state.save("current", workspace=current_workspace)
 
-    last_doc = switches.find_one(sort=[("_id", -1)])
+    # Enforce desktop task assignments
+    current_task = enforce_desktop_task(current_workspace, work_desktop_tasks, current_task)
 
-    # work workflow
+    # Track workspace switches
-    if current_workspace in work_desktops.keys():
+    last_switch_time = track_workspace_switch(current_workspace, current_task, last_switch_time)
-        work_states = state.retrieve_work_state()
-        current_work_task = work_states[work_desktops[current_workspace]]
-
-        # Get all task IDs under current workspace path
-        workspace_tasks = task.get_tasks_tree(
-            f"work/{work_desktops[current_workspace]}"
-        )
-        work_task_ids = {t["task_id"] for t in workspace_tasks if "task_id" in t}
-
-        # if current_task in work_task_ids and current_task != current_work_task:
-        if current_task not in work_task_ids:
-            # Enforce work task if current task is not in workspace
-            if current_task != current_work_task:
-                current_task = current_work_task
-                state.save("current", task=current_task)
-                task.db_to_file_as_is(None)
-
-        elif current_task != current_work_task:
-            # Update work state when switching to a different valid task
-            state.update_work_state(work_desktops[current_workspace], current_task)
-
-    # regular flow
-    if (
-        last_doc["workspace"] == desktop(current_workspace)
-        and last_doc["task"] == current_task
-    ):
-        delta = round((now() - last_switch_time).total_seconds())
-        switches.update_one(
-            {"_id": last_doc["_id"]}, {"$set": {"delta": delta, "task": current_task}}
-        )
-    else:
-        current_workspace = active_workspace()
-        switch = {
-            "workspace": desktop(current_workspace),
-            "date": now(),
-            "delta": 0,
-            "task": current_task,
-        }
-        switches.insert_one(switch)
-        last_switch_time = now()
 
     time.sleep(2)

dmapp/dmcore/state.py

@@ -7,10 +7,11 @@ def save(
     task: str | None = None,
     workspace: str | None = None,
     filetime: str | None = None,
+    config_mtime: float | None = None,
 ) -> None:
     """
     Upsert a document with _id=doc_id, setting any of the provided fields.
-    Leave fields you don’t pass unchanged.
+    Leave fields you don't pass unchanged.
     """
     updates: dict = {}
     if task is not None:
@@ -19,6 +20,8 @@ def save(
         updates["workspace"] = workspace
     if filetime is not None:
         updates["filetime"] = filetime
+    if config_mtime is not None:
+        updates["config_mtime"] = config_mtime
 
     if updates:
         states.update_one(
@@ -30,40 +33,36 @@ def save(
 
 def retrieve(doc_id: str) -> dict[str, str | None]:
     """
-    Fetches the document with _id=doc_id and returns its 'task' and 'workspace'.
+    Fetches the document with _id=doc_id and returns its fields.
-    If the document doesn’t exist, both will be None.
+    If the document doesn't exist, all fields will be None.
     """
     doc = states.find_one({"_id": doc_id})
     return {
         "task": doc.get("task") if doc else None,
         "workspace": doc.get("workspace") if doc else None,
         "filetime": doc.get("filetime") if doc else None,
+        "config_mtime": doc.get("config_mtime") if doc else None,
     }
 
 
-# just
+def sync_desktop_tasks(work_desktop_tasks: dict):
-def update_work_state(work: str, task_id: str):
     """
-    update work state
+    Sync work_desktop_tasks from config file to state
     """
-    states.update_one({"_id": "work"}, {"$set": {work: task_id}})
+    update_dict = {str(k): v for k, v in work_desktop_tasks.items()}
-
+    states.update_one(
-
+        {"_id": "work_desktop_tasks"},
-def init_work_state(wd: dict):
+        {"$set": update_dict},
-    """
+        upsert=True,
-    init work states with default values
-    """
-    if not states.find_one({"_id": "work"}):
-        states.insert_one(
-            {
-                "_id": "work",
-                **{
-                    wd[k]: tasks.find_one({"path": f"work/{wd[k]}"})["task_id"]
-                    for k in wd
-                },
-            }
     )
 
 
-def retrieve_work_state():
+def retrieve_desktop_state():
-    return states.find_one({"_id": "work"})
+    """
+    Get work_desktop_tasks mapping from state
+    """
+    doc = states.find_one({"_id": "work_desktop_tasks"})
+    if not doc:
+        return {}
+    # Convert string keys to int and exclude _id
+    return {int(k): v for k, v in doc.items() if k != "_id"}

dmapp/dmweb/config.json

@@ -0,0 +1,7 @@
+{
+  "timezone": "America/Argentina/Buenos_Aires",
+  "task_file": "/home/mariano/LETRAS/adm/task/main",
+  "mongodb_host": "localhost",
+  "mongodb_port": 27017,
+  "mongodb_db": "deskmeter"
+}

dmapp/dmweb/dm.py

@@ -2,7 +2,7 @@ from datetime import datetime, timedelta
 
 from flask import Blueprint, render_template
 
-from .get_period_times import get_period_totals, task_or_none, timezone, get_work_project_tasks, get_work_period_totals
+from .get_period_times import get_period_totals, task_or_none, timezone, get_work_period_totals, get_current_task_info, convert_seconds, get_task_time_seconds
 
 dmbp = Blueprint("deskmeter", __name__, url_prefix="/", template_folder="templates")
 
@@ -24,9 +24,18 @@ def index(task=None):
     end = datetime.today().replace(hour=23, minute=59, second=59, tzinfo=timezone)
 
     rows = get_period_totals(start, end, task)
+
+    # Get current task info
+    current_task_id, current_task_path = get_current_task_info()
+    current_task_time = None
+    if current_task_id:
+        total_seconds = get_task_time_seconds(start, end, current_task_id)
+        if total_seconds > 0:
+            current_task_time = convert_seconds(total_seconds)
+
     print(rows)
 
-    return render_template("main.html", rows=rows)
+    return render_template("main.html", rows=rows, current_task_path=current_task_path, current_task_time=current_task_time)
 
 
 @dmbp.route("/day/<int:month>/<int:day>")

dmapp/dmweb/dmcal.py

@@ -12,7 +12,6 @@ from .get_period_times import (
     task_file,
     task_or_none,
     timezone,
-    get_work_project_tasks,
     get_work_period_totals,
 )
 

dmapp/dmweb/get_period_times.py

@@ -17,6 +17,52 @@ tasks = db.task
 task_file = "/home/mariano/LETRAS/adm/task/main"
 
 
+def get_current_task_info():
+    """Get current task ID and path from state and tasks collection"""
+    states = db.state
+    current_doc = states.find_one({"_id": "current"})
+
+    if not current_doc or "task" not in current_doc:
+        return None, None
+
+    task_id = current_doc["task"]
+    task_doc = tasks.find_one({"task_id": task_id})
+
+    if task_doc and "path" in task_doc:
+        return task_id, task_doc["path"]
+
+    return task_id, None
+
+
+def get_task_time_seconds(start, end, task_id, workspaces=None):
+    """Get total seconds for a task within a time period using MongoDB aggregation."""
+    if workspaces is None:
+        workspaces = ["Plan", "Think", "Work"]
+
+    pipeline = [
+        {
+            "$match": {
+                "date": {"$gte": start, "$lte": end},
+                "task": task_id,
+                "workspace": {"$in": workspaces}
+            }
+        },
+        {
+            "$group": {
+                "_id": None,
+                "total_seconds": {"$sum": "$delta"}
+            }
+        }
+    ]
+
+    result = list(switches.aggregate(pipeline))
+
+    if result and len(result) > 0:
+        return result[0]["total_seconds"]
+
+    return 0
+
+
 def task_or_none(task=None):
     if not task:
         task = read_and_extract(task_file)
@@ -62,66 +108,43 @@ def read_and_extract(file_path):
     return None
 
 
-def get_work_projects():
-    """Get dict of work projects with their task IDs."""
-    work_tasks = list(
-        tasks.find(
-            {"path": {"$regex": "^work/"}, "task_id": {"$exists": True}},
-            {"path": 1, "task_id": 1, "_id": 0},
-        )
-    )
-    
-    projects = {}
-    for task in work_tasks:
-        # Extract project name from path like "work/cal" -> "cal"
-        path_parts = task["path"].split("/")
-        if len(path_parts) >= 2:
-            project_name = path_parts[1]
-            if project_name not in projects:
-                projects[project_name] = []
-            projects[project_name].append(task["task_id"])
-    
-    return projects
-
-
-def get_work_project_tasks():
-    """Get comma-separated string of all task IDs under work/* paths."""
-    projects = get_work_projects()
-    all_task_ids = []
-    for task_ids in projects.values():
-        all_task_ids.extend(task_ids)
-    return ",".join(all_task_ids) if all_task_ids else None
-
-
 def get_work_period_totals(start, end):
-    """Get period totals grouped by work project."""
+    """Get period totals grouped by task with full path."""
-    projects = get_work_projects()
+    # Get all tasks with time in the period
+    pipeline = [
+        {
+            "$match": {
+                "date": {"$gte": start, "$lte": end},
+                "workspace": {"$in": ["Plan", "Think", "Work"]},
+                "task": {"$exists": True, "$ne": None}
+            }
+        },
+        {
+            "$group": {
+                "_id": "$task",
+                "total_seconds": {"$sum": "$delta"}
+            }
+        }
+    ]
+
+    results = list(switches.aggregate(pipeline))
     combined_rows = []
 
-    for project_name, task_ids in projects.items():
+    for result in results:
-        if not task_ids:
+        task_id = result["_id"]
-            continue
+        total_seconds = result["total_seconds"]
-            
-        task_string = ",".join(task_ids)
-        rows = get_period_totals(start, end, task_string)
-        
-        # Sum up all time for this project (looking for "Work" workspace)
-        total_seconds = 0
-        for row in rows:
-            if row["ws"] == "Work":
-                # Convert time string back to seconds to sum properly
-                time_parts = row["total"].split(":")
-                if len(time_parts) == 3:
-                    hours, minutes, seconds = map(int, time_parts)
-                    total_seconds += hours * 3600 + minutes * 60 + seconds
 
         if total_seconds > 0:
+            # Get task path from tasks collection
+            task_doc = tasks.find_one({"task_id": task_id})
+            task_path = task_doc["path"] if task_doc and "path" in task_doc else task_id
+
             combined_rows.append({
-                "ws": project_name, 
+                "ws": task_path,
                 "total": convert_seconds(total_seconds)
             })
 
-    # Sort by project name for consistency
+    # Sort by path for consistency
     combined_rows.sort(key=lambda x: x["ws"])
     return combined_rows
 

dmapp/dmweb/templates/main.html

@@ -39,6 +39,12 @@
 
 
         {% block content %}
+        {% if current_task_path and current_task_time %}
+        <div style="font-size: 48pt; margin-bottom: 40px; text-align: center;">
+            <div style="color: #333;">{{ current_task_path }}</div>
+            <div style="color: #666; font-size: 36pt;">{{ current_task_time }}</div>
+        </div>
+        {% endif %}
         <table>
         {%  for row in rows %}
             {% if row["ws"] in ['Away', 'Other'] %}