pyp6xer_schedule_quality

Instructions

Implementation Reference

• server.py:822-827 (registration)

  Tool registration using @mcp.tool decorator with readOnlyHint=True annotation

  @mcp.tool(annotations=ToolAnnotations(readOnlyHint=True, destructiveHint=False, idempotentHint=True, openWorldHint=False))
  def pyp6xer_schedule_quality(
      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:

• server.py:823-900 (handler)

  Implements DCMA-style schedule quality checks: open ends, lags/leads, hard constraints, negative float, resource gaps, and milestone checks with DCMA threshold pass/fail scoring

  def pyp6xer_schedule_quality(
      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:
      """Run DCMA-style schedule quality checks.
  
      Checks include:
      - Missing predecessors / successors (open ends)
      - Activities with lags or leads
      - Activities with hard constraints
      - Negative total float
      - Activities with no resources (optional warning)
      - Milestone checks
      """
      xer = _get_xer(ctx, cache_key)
      tasks = _get_tasks(xer, proj_id)
  
      # Exclude LOE, WBS summary, and completed
      workable = [
          t for t in tasks
          if not t.type.is_loe
          and not t.type.is_wbs
          and not t.status.is_completed
      ]
      n = len(workable)
      if n == 0:
          return json.dumps({"error": "No open activities to analyse."})
  
      no_pred = [t.task_code for t in workable if not t.predecessors]
      no_succ = [t.task_code for t in workable if not t.successors]
      with_lags = [
          t.task_code for t in workable
          if any(lnk.lag > 0 for lnk in t.predecessors)
      ]
      with_leads = [
          t.task_code for t in workable
          if any(lnk.lag < 0 for lnk in t.predecessors)
      ]
      constrained = [
          t.task_code for t in workable
          if t.cstr_type and t.cstr_type not in ("CS_ALAP", "CS_MSOA")
      ]
      neg_float = [t.task_code for t in workable if t.is_critical and t.total_float is not None and t.total_float < 0]
      no_resources = [t.task_code for t in workable if not t.resources and not t.type.is_milestone]
  
      def _pct(lst):
          return round(len(lst) / n * 100, 1)
  
      checks = {
          "activities_analysed": n,
          "missing_predecessors": {"count": len(no_pred), "pct": _pct(no_pred), "activities": no_pred[:20]},
          "missing_successors": {"count": len(no_succ), "pct": _pct(no_succ), "activities": no_succ[:20]},
          "activities_with_lags": {"count": len(with_lags), "pct": _pct(with_lags)},
          "activities_with_leads": {"count": len(with_leads), "pct": _pct(with_leads), "activities": with_leads[:20]},
          "hard_constraints": {"count": len(constrained), "pct": _pct(constrained), "activities": constrained[:20]},
          "negative_float": {"count": len(neg_float), "pct": _pct(neg_float), "activities": neg_float},
          "no_resources_assigned": {"count": len(no_resources), "pct": _pct(no_resources)},
      }
  
      # DCMA thresholds (pass if below)
      thresholds = {
          "missing_predecessors": 5.0,
          "missing_successors": 5.0,
          "activities_with_lags": 5.0,
          "activities_with_leads": 0.0,
          "hard_constraints": 5.0,
          "negative_float": 0.0,
      }
      flags = {
          k: ("FAIL" if checks[k]["pct"] > v else "PASS")
          for k, v in thresholds.items()
      }
      pass_count = sum(1 for v in flags.values() if v == "PASS")
      checks["dcma_results"] = flags
      checks["dcma_score"] = f"{pass_count}/{len(flags)}"
  
      return json.dumps(checks, indent=2)