Plan Analytics — Volume, Intensity & Muscle-Group Coverage

Mix.install([
  {:wpl_validator, path: Path.join(__DIR__, "..")},
  {:jason, "~> 1.4"},
  {:kino, "~> 0.13"},
  {:kino_vega_lite, "~> 0.1"}
])

Goal

Load the canonical muscle-tagged-workout.json and multi-phase-progression.json, validate them, then compute and visualize:

1. Weekly volume per muscle group — Schoenfeld-style sets/muscle/week tracking, the analytics gap that motivated the v1.3.0 schema additions.
2. Weekly volume + intensity progression across phases, with deload weeks called out via Week.is_deload.

This is the kind of read-after-validate tooling a coach app or tracker would build on top of WPL.

Sets per muscle group (schema v1.3.0+)

Before muscle-group taxonomy was in the schema, computing weekly sets per muscle required an out-of-band exercise→muscle map maintained by every consuming tool. With v1.3.0 the primary_muscles and secondary_muscles arrays live on the activity itself.

muscle_doc =
  Path.join([__DIR__, "..", "priv", "conformance", "valid", "muscle-tagged-workout.json"])
  |> File.read!()
  |> Jason.decode!()

muscle_result = WPL.Validator.validate(muscle_doc)
muscle_result.valid?

defmodule MuscleVolume do
  @doc """
  For each `sets_reps` exercise activity, attribute its sets to every
  primary muscle (full count) and every secondary muscle (half count, a
  common volume-tracking convention used by Helms / Schoenfeld-style
  hypertrophy programming).

  Returns a sorted list of `{muscle, sets}` rows.
  """
  def per_muscle_sets(doc) do
    plan = doc["plan"]

    plan["phases"]
    |> List.wrap()
    |> Enum.flat_map(fn phase ->
      phase["weeks"]
      |> List.wrap()
      |> Enum.flat_map(fn week ->
        week["days"]
        |> List.wrap()
        |> Enum.flat_map(fn day ->
          day["blocks"]
          |> List.wrap()
          |> Enum.flat_map(fn block ->
            block["activities"]
            |> List.wrap()
            |> Enum.flat_map(&attribute/1)
          end)
        end)
      end)
    end)
    |> Enum.reduce(%{}, fn {muscle, sets}, acc ->
      Map.update(acc, muscle, sets, &(&1 + sets))
    end)
    |> Enum.sort_by(fn {_muscle, sets} -> -sets end)
    |> Enum.map(fn {muscle, sets} -> %{muscle: muscle, sets: sets} end)
  end

  defp attribute(%{"type" => "exercise"} = a) do
    sets = get_in(a, ["prescription", "sets"]) || 0
    primary = a["primary_muscles"] || []
    secondary = a["secondary_muscles"] || []

    Enum.map(primary, &{&1, sets * 1.0}) ++
      Enum.map(secondary, &{&1, sets * 0.5})
  end

  defp attribute(_), do: []
end

rows = MuscleVolume.per_muscle_sets(muscle_doc)
Kino.DataTable.new(rows)

VegaLite.new(width: 600, height: 320, title: "Sets per muscle group (1.0× primary, 0.5× secondary)")
|> VegaLite.data_from_values(rows)
|> VegaLite.mark(:bar)
|> VegaLite.encode_field(:y, "muscle", type: :nominal, sort: "-x", title: nil)
|> VegaLite.encode_field(:x, "sets", type: :quantitative, title: "Sets")

This single-day fixture is small (one workout), but the same code runs on a full mesocycle and gives the coach a per-muscle weekly-volume chart with no exercise catalog lookup required.

Phase progression — load + validate

plan_path =
  Path.join([
    __DIR__,
    "..",
    "priv",
    "conformance",
    "valid",
    "multi-phase-progression.json"
  ])

doc = plan_path |> File.read!() |> Jason.decode!()
result = WPL.Validator.validate(doc)
result.valid?

Walk the plan

Plans are nested deeply (plan → phases → weeks → days → blocks → activities). Here’s a tiny walker that flattens the tree into one row per exercise set, ready for analytics.

defmodule Walk do
  @doc """
  Flatten the plan tree to one row per `sets_reps` exercise activity.
  `global_week` is cumulative across phases (so a 2+2+2-week plan numbers
  weeks 1..6, regardless of how many weeks each phase contains).
  """
  def to_set_rows(doc) do
    phases = doc["plan"]["phases"] || []

    {rows, _} =
      Enum.flat_map_reduce(Enum.with_index(phases), 0, fn {phase, phase_idx}, weeks_so_far ->
        weeks = phase["weeks"] || []

        rows =
          for {week, week_idx} <- Enum.with_index(weeks),
              day <- week["days"] || [],
              block <- day["blocks"] || [],
              activity <- block["activities"] || [],
              activity["type"] == "exercise",
              prescription = activity["prescription"],
              prescription["type"] == "sets_reps" do
            %{
              phase: phase["name"],
              phase_order: phase_idx + 1,
              week: week["name"],
              global_week: weeks_so_far + week_idx + 1,
              exercise: activity["exercise_ref"],
              sets: prescription["sets"] || 1,
              reps: reps_value(prescription["reps"]),
              volume: (prescription["sets"] || 1) * reps_value(prescription["reps"]),
              intensity_pct_1rm: get_in(prescription, ["weight", "value"])
            }
          end

        {rows, weeks_so_far + length(weeks)}
      end)

    rows
  end

  # Treat any reps shape uniformly: prefer target, else midpoint of min/max.
  defp reps_value(%{"target" => t}), do: t
  defp reps_value(%{"min" => mn, "max" => mx}), do: div(mn + mx, 2)
  defp reps_value(%{"min" => mn}), do: mn
  defp reps_value(_), do: 1
end

rows = Walk.to_set_rows(doc)
Kino.DataTable.new(rows)

Aggregate by week

weekly =
  rows
  |> Enum.group_by(& &1.global_week)
  |> Enum.map(fn {wk, group} ->
    sample = hd(group)
    intensities = group |> Enum.map(& &1.intensity_pct_1rm) |> Enum.reject(&is_nil/1)

    %{
      week_no: wk,
      phase: sample.phase,
      label: "W#{wk} #{sample.phase}",
      total_volume: Enum.sum(Enum.map(group, & &1.volume)),
      avg_intensity:
        if(intensities == [],
          do: nil,
          else: Float.round(Enum.sum(intensities) / length(intensities), 1)
        )
    }
  end)
  |> Enum.sort_by(& &1.week_no)

Kino.DataTable.new(weekly)

Volume chart

Volume should taper into the deload week.

VegaLite.new(width: 600, height: 240, title: "Weekly volume (sets × reps)")
|> VegaLite.data_from_values(weekly, only: ["label", "total_volume", "phase"])
|> VegaLite.mark(:bar)
|> VegaLite.encode_field(:x, "label", type: :ordinal, sort: nil, title: "Week")
|> VegaLite.encode_field(:y, "total_volume", type: :quantitative, title: "Volume (reps)")
|> VegaLite.encode_field(:color, "phase", type: :nominal, title: "Phase")

Intensity chart

Intensity should climb from hypertrophy → strength → peak, then drop on deload.

VegaLite.new(width: 600, height: 240, title: "Weekly average intensity (% 1RM)")
|> VegaLite.data_from_values(weekly, only: ["label", "avg_intensity", "phase"])
|> VegaLite.mark(:line, point: true)
|> VegaLite.encode_field(:x, "label", type: :ordinal, sort: nil, title: "Week")
|> VegaLite.encode_field(:y, "avg_intensity",
  type: :quantitative,
  title: "% 1RM",
  scale: [domain: [50, 100]]
)
|> VegaLite.encode_field(:color, "phase", type: :nominal, title: "Phase")

Try it on your own plan

Replace the plan_path at the top with another fixture (e.g. ppl-split.json) or your own JSON. The walker only counts sets_reps exercise activities — extend it for cardio (continuous/intervals) or nutrition macros if your plan needs that.