new view, gnome extension

dmapp/dmweb/get_period_times.py

@@ -148,6 +148,130 @@ def get_work_period_totals(start, end):
     return combined_rows
 
 
+def get_task_blocks_calendar(start, end, task=None, min_block_seconds=300):
+    """
+    Get task blocks for calendar-style visualization.
+    Groups consecutive switches to the same task into blocks, tracking active/idle time.
+
+    Returns list of blocks:
+    [{
+        'task_id': str,
+        'task_path': str,
+        'start': datetime,
+        'end': datetime,
+        'duration': int (total seconds),
+        'active_seconds': int (Plan/Think/Work time),
+        'idle_seconds': int (Other/Away time),
+        'active_ratio': float (0.0 to 1.0)
+    }, ...]
+    """
+    task_query = {"$in": task.split(",")} if task else {}
+
+    match_query = {"date": {"$gte": start, "$lte": end}}
+    if task_query:
+        match_query["task"] = task_query
+
+    # Get all switches in period, sorted by date
+    raw_switches = list(switches.find(match_query).sort("date", 1))
+
+    if not raw_switches:
+        return []
+
+    blocks = []
+    current_block = None
+
+    for switch in raw_switches:
+        ws = switch["workspace"]
+        task_id = switch.get("task")
+        switch_start = switch["date"].replace(tzinfo=utctz).astimezone(timezone)
+        switch_duration = switch["delta"]
+        switch_end = switch_start + timedelta(seconds=switch_duration)
+
+        is_active = ws in ["Plan", "Think", "Work"]
+
+        # Start new block if task changed
+        if current_block is None or current_block["task_id"] != task_id:
+            if current_block is not None:
+                blocks.append(current_block)
+
+            # Get task path
+            task_path = None
+            if task_id:
+                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
+
+            current_block = {
+                "task_id": task_id,
+                "task_path": task_path or "No Task",
+                "start": switch_start,
+                "end": switch_end,
+                "duration": switch_duration,
+                "active_seconds": switch_duration if is_active else 0,
+                "idle_seconds": 0 if is_active else switch_duration
+            }
+        else:
+            # Extend current block
+            current_block["end"] = switch_end
+            current_block["duration"] += switch_duration
+            if is_active:
+                current_block["active_seconds"] += switch_duration
+            else:
+                current_block["idle_seconds"] += switch_duration
+
+    # Add final block
+    if current_block is not None:
+        blocks.append(current_block)
+
+    # Filter out very short blocks and calculate active ratio
+    filtered_blocks = []
+    for block in blocks:
+        if block["duration"] >= min_block_seconds:
+            block["active_ratio"] = block["active_seconds"] / block["duration"] if block["duration"] > 0 else 0
+            filtered_blocks.append(block)
+
+    return filtered_blocks
+
+
+def get_raw_switches(start, end, task=None):
+    """
+    Get all raw switch documents in the period.
+
+    Returns list of switches:
+    [{
+        'workspace': str,
+        'task_id': str,
+        'task_path': str,
+        'date': datetime,
+        'delta': int (seconds)
+    }, ...]
+    """
+    task_query = {"$in": task.split(",")} if task else {}
+
+    match_query = {"date": {"$gte": start, "$lte": end}}
+    if task_query:
+        match_query["task"] = task_query
+
+    raw_switches = list(switches.find(match_query).sort("date", 1))
+
+    result = []
+    for switch in raw_switches:
+        task_id = switch.get("task")
+        task_path = None
+        if task_id:
+            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
+
+        result.append({
+            "workspace": switch["workspace"],
+            "task_id": task_id,
+            "task_path": task_path or "No Task",
+            "date": switch["date"].replace(tzinfo=utctz).astimezone(timezone),
+            "delta": switch["delta"]
+        })
+
+    return result
+
+
 def get_period_totals(start, end, task=None):
     task_query = {"$in": task.split(",")} if task else {}