pyp6xer_compare_snapshots

Instructions

Implementation Reference

• server.py:1590-1660 (handler)

  The main handler function for pyp6xer_compare_snapshots. It loads two XER snapshots by cache_key, compares activities task_code-by-task_code, and reports added, removed, and changed activities (finish date, status, remaining duration, total float, percent complete).

  @mcp.tool(annotations=ToolAnnotations(readOnlyHint=True, destructiveHint=False, idempotentHint=True, openWorldHint=False))
  def pyp6xer_compare_snapshots(
      cache_key_a: Annotated[str, Field(description="Cache key of the base (original) snapshot")],
      cache_key_b: Annotated[str, Field(description="Cache key of the modified snapshot to compare")],
      proj_id: Annotated[str | None, Field(description="Project ID or short name; uses first project if omitted")] = None,
      ctx: Context = None,
  ) -> str:
      """Compare two loaded XER files (snapshots) to identify schedule changes.
  
      Reports added, removed, and changed activities (dates, duration, float, status).
      Useful for analysing schedule updates between periods.
  
      Args:
          cache_key_a: Cache key for the baseline/earlier snapshot.
          cache_key_b: Cache key for the updated/later snapshot.
          proj_id:     Optional project filter (matched by short_name).
      """
      xer_a = _get_xer(ctx, cache_key_a)
      xer_b = _get_xer(ctx, cache_key_b)
  
      def _tasks_by_code(xer, pid):
          tasks = _get_tasks(xer, pid)
          return {t.task_code: t for t in tasks}
  
      # Try to match project by short_name if proj_id provided
      pid_a = proj_id
      pid_b = proj_id
  
      tasks_a = _tasks_by_code(xer_a, pid_a)
      tasks_b = _tasks_by_code(xer_b, pid_b)
  
      codes_a = set(tasks_a.keys())
      codes_b = set(tasks_b.keys())
  
      added = list(codes_b - codes_a)
      removed = list(codes_a - codes_b)
      common = codes_a & codes_b
  
      changes = []
      for code in sorted(common):
          a, b = tasks_a[code], tasks_b[code]
          diffs = {}
          try:
              if a.finish != b.finish:
                  diffs["finish"] = {"from": _fmt_date(a.finish), "to": _fmt_date(b.finish)}
          except Exception:
              pass
          if a.status != b.status:
              diffs["status"] = {"from": a.status.value, "to": b.status.value}
          if a.remaining_duration != b.remaining_duration:
              diffs["remaining_duration_days"] = {"from": a.remaining_duration, "to": b.remaining_duration}
          if a.total_float != b.total_float:
              diffs["total_float_days"] = {"from": a.total_float, "to": b.total_float}
          if abs((a.percent_complete or 0) - (b.percent_complete or 0)) > 0.005:
              diffs["percent_complete"] = {
                  "from": round((a.percent_complete or 0) * 100, 1),
                  "to": round((b.percent_complete or 0) * 100, 1),
              }
          if diffs:
              changes.append({"task_code": code, "name": b.name, "changes": diffs})
  
      return json.dumps({
          "snapshot_a": cache_key_a,
          "snapshot_b": cache_key_b,
          "activities_in_a": len(tasks_a),
          "activities_in_b": len(tasks_b),
          "added": added,
          "removed": removed,
          "changed_count": len(changes),
          "changed": changes,
      }, indent=2)

• server.py:1590-1596 (registration)

  The tool is registered via the @mcp.tool decorator on the pyp6xer_compare_snapshots function. The decorator also sets ToolAnnotations with readOnlyHint=True, destructiveHint=False, idempotentHint=True, openWorldHint=False.

  @mcp.tool(annotations=ToolAnnotations(readOnlyHint=True, destructiveHint=False, idempotentHint=True, openWorldHint=False))
  def pyp6xer_compare_snapshots(
      cache_key_a: Annotated[str, Field(description="Cache key of the base (original) snapshot")],
      cache_key_b: Annotated[str, Field(description="Cache key of the modified snapshot to compare")],
      proj_id: Annotated[str | None, Field(description="Project ID or short name; uses first project if omitted")] = None,
      ctx: Context = None,
  ) -> str:

• server.py:1591-1596 (schema)

  Input schema defined via Pydantic Field annotations on parameters: cache_key_a (str, required), cache_key_b (str, required), proj_id (str | None, optional), ctx (Context).

  def pyp6xer_compare_snapshots(
      cache_key_a: Annotated[str, Field(description="Cache key of the base (original) snapshot")],
      cache_key_b: Annotated[str, Field(description="Cache key of the modified snapshot to compare")],
      proj_id: Annotated[str | None, Field(description="Project ID or short name; uses first project if omitted")] = None,
      ctx: Context = None,
  ) -> str:

• server.py:149-175 (helper)

  Helper functions _get_xer, _get_tasks, and _task_to_dict used by compare_snapshots to retrieve XER data and format activity dictionaries.

  def _get_xer(ctx: Context, cache_key: str) -> Xer:
      return _get_cache(ctx, cache_key)["xer"]
  
  
  def _get_project(xer: Xer, proj_id: str | None):
      if proj_id:
          proj = xer.projects.get(proj_id)
          if proj is None:
              ids = list(xer.projects.keys())
              raise ValueError(
                  f"Project '{proj_id}' not found. Available project IDs: {ids}"
              )
          return proj
      projects = list(xer.projects.values())
      if not projects:
          raise ValueError("No projects found in this XER file.")
      return projects[0]
  
  
  def _get_tasks(xer: Xer, proj_id: str | None):
      """Return tasks for a project (or all tasks if proj_id is None)."""
      if proj_id:
          return _get_project(xer, proj_id).tasks
      return list(xer.tasks.values())
  
  
  def _task_to_dict(task, fields: list[str] | None = None) -> dict:

• server.py:139-139 (helper)

  Helper _get_cache used to retrieve cached XER data by cache_key, called by _get_xer which is used by compare_snapshots.

  def _get_cache(ctx: Context, cache_key: str) -> dict: