pyp6xer_update_activity

Instructions

Implementation Reference

• server.py:1750-1783 (handler)

  The pyp6xer_update_activity tool handler function that updates fields on a single activity in the in-memory cache. It accepts task_code, updates dict, cache_key, proj_id, and ctx parameters, then delegates to _apply_activity_update to apply changes.

  @mcp.tool(annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=True, openWorldHint=False))
  def pyp6xer_update_activity(
      task_code: str,
      updates: Annotated[dict, Field(description="Dict of field→value to update on the activity (e.g. {'percent_complete': 50})")],
      cache_key: Annotated[str, Field(description="Cache key identifying the loaded XER file (set when calling pyp6xer_load_file)")] = "default",
      proj_id: Annotated[str | None, Field(description="Project ID or short name; uses first project if omitted")] = None,
      ctx: Context = None,
  ) -> str:
      """Update fields on a single activity in the in-memory cache.
  
      Changes are held in memory until pyp6xer_write_file is called.
  
      Updatable fields:
      - status_code: 'TK_NotStart', 'TK_Active', or 'TK_Complete'
      - phys_complete_pct: physical percent complete (0–100)
      - remain_drtn_hr_cnt: remaining duration in hours
      - act_start_date / act_end_date: actual dates (YYYY-MM-DD)
      - expect_end_date: expected finish (YYYY-MM-DD)
      - target_start_date / target_end_date: baseline dates (YYYY-MM-DD)
  
      Args:
          task_code: Activity ID to update.
          updates:   Dict of {field_name: new_value}.
          cache_key: Cache key of the loaded file.
          proj_id:   Optional project filter.
      """
      entry = _get_cache(ctx, cache_key)
      applied = _apply_activity_update(entry, task_code, proj_id, updates)
      return json.dumps({
          "status": "updated",
          "task_code": task_code,
          "changes": applied,
          "note": "Call pyp6xer_write_file to persist. Omit output_path to overwrite the source file.",
      }, indent=2)

• server.py:1682-1747 (helper)

  The _apply_activity_update helper function that performs the actual field updates on both the in-memory Xer object and the raw_tables dict. It handles field mapping via _UPDATABLE_FIELDS, date format conversion, and returns a dict of changes.

  def _apply_activity_update(entry: dict, task_code: str, proj_id: str | None, updates: dict) -> dict:
      """
      Apply updates to both the in-memory Xer object and the raw_tables dict.
      Returns a dict of {field: {from, to}} describing what changed.
      """
      xer: Xer = entry["xer"]
      raw_tables: dict = entry["raw_tables"]
  
      # Find the in-memory task
      tasks = _get_tasks(xer, proj_id)
      task = next((t for t in tasks if t.task_code == task_code), None)
      if task is None:
          raise ValueError(f"Activity '{task_code}' not found.")
  
      # Find the raw TASK row
      task_rows = raw_tables.get("TASK", {}).get("rows", [])
      raw_row = next((r for r in task_rows if r.get("task_id") == task.uid), None)
  
      applied = {}
      for field, value in updates.items():
          if field not in _UPDATABLE_FIELDS:
              raise ValueError(
                  f"Field '{field}' is not updatable. "
                  f"Updatable fields: {list(_UPDATABLE_FIELDS.keys())}"
              )
          raw_field = _UPDATABLE_FIELDS[field]
  
          # Get old value for change log
          old_raw = raw_row.get(raw_field, "") if raw_row else ""
  
          # Update raw table row
          if raw_row is not None:
              if field in ("act_start_date", "act_end_date", "expect_end_date",
                           "target_start_date", "target_end_date"):
                  # Convert YYYY-MM-DD to XER datetime format if needed
                  if value and ":" not in str(value):
                      value_raw = f"{value} 00:00"
                  else:
                      value_raw = value or ""
                  raw_row[raw_field] = value_raw
              else:
                  raw_row[raw_field] = str(value) if value is not None else ""
  
          # Update in-memory object (best-effort)
          try:
              if field == "status_code":
                  from xerparser.schemas.task import TASK as TaskClass
                  task.status = TaskClass.TaskStatus[value]
              elif field == "phys_complete_pct":
                  task.phys_complete_pct = float(value) / 100.0
              elif field == "remain_drtn_hr_cnt":
                  task.remain_drtn_hr_cnt = float(value)
              elif field in ("act_start_date", "act_end_date", "expect_end_date",
                             "target_start_date", "target_end_date"):
                  from xerparser.src.utils import optional_date
                  if value:
                      dt = datetime.strptime(str(value).strip(), "%Y-%m-%d")
                      setattr(task, field, dt)
                  else:
                      setattr(task, field, None)
          except Exception:
              pass  # Raw table update is authoritative; in-memory is best-effort
  
          applied[field] = {"from": old_raw, "to": str(value)}
  
      return applied

• server.py:1667-1677 (schema)

  The _UPDATABLE_FIELDS mapping dict that defines which field names can be updated and maps them to their XER raw table field names. Also the _XER_DATE_FMT constant for date formatting.

  # Field mapping: friendly name -> XER raw table field name
  _UPDATABLE_FIELDS = {
      "status_code": "status_code",
      "phys_complete_pct": "phys_complete_pct",
      "remain_drtn_hr_cnt": "remain_drtn_hr_cnt",
      "act_start_date": "act_start_date",
      "act_end_date": "act_end_date",
      "expect_end_date": "expect_end_date",
      "target_start_date": "target_start_date",
      "target_end_date": "target_end_date",
  }