Synchronize Canvas outcomes (learning mastery gradebook) with Ed code challenge grading criteria.

canvas_assessment_sync.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "canvas-assessment-sync.ipynb",
      "provenance": []
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    }
  },
  "cells": [
    {
      "cell_type": "code",
      "metadata": {
        "id": "rn0bFFJ29Pvj"
      },
      "source": [
        "!pip install --upgrade -q aiohttp[speedups] nest_asyncio canvasapi\n",
        "\n",
        "import aiohttp\n",
        "import asyncio\n",
        "import io\n",
        "import itertools\n",
        "import pandas as pd\n",
        "import nest_asyncio\n",
        "import sys\n",
        "\n",
        "nest_asyncio.apply()\n",
        "\n",
        "from canvasapi import Canvas\n",
        "\n",
        "api_key = \"\"\n",
        "canvas = Canvas(\"https://canvas.uw.edu\", api_key)\n",
        "token = input(\"Ed token: \")"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "rlVD6DFAlJaP"
      },
      "source": [
        "# Canvas course (URL identifier)\n",
        "course = canvas.get_course(...)\n",
        "\n",
        "# Ed challenge identifier to Canvas assignment (URL identifier)\n",
        "assignments = {\n",
        "    \"...\": course.get_assignment(...),\n",
        "}\n",
        "\n",
        "# UW email addresses to Ed preferred email addresses\n",
        "email_remap = {\n",
        "}\n",
        "users = {email_remap.get(user.email, user.email): user.id for user in course.get_users()}\n",
        "\n",
        "# Rubric item names (must be the same between Ed and Canvas)\n",
        "rubrics = [\n",
        "    \"Behavior\",\n",
        "    \"Concepts\",\n",
        "    \"Quality\",\n",
        "    \"Testing\",\n",
        "    \"Writeup\",\n",
        "]"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "D03IhDxb7Hkf"
      },
      "source": [
        "# https://canvas.instructure.com/doc/api/submissions.html#method.submissions_api.update\n",
        "def rubric_assessment(assignment, email, rubric_ratings):\n",
        "    try:\n",
        "        result = {}\n",
        "        for rubric, rating in rubric_ratings:\n",
        "            criterion = next(c for c in assignment.rubric if c[\"description\"] == rubric)\n",
        "            criterion_rating = next(r for r in criterion[\"ratings\"] if r[\"description\"] == rating)\n",
        "            result[criterion[\"id\"]] = {\n",
        "                \"points\": criterion_rating[\"points\"],\n",
        "                \"rating_id\": criterion_rating[\"id\"],\n",
        "            }\n",
        "        return result\n",
        "    except:\n",
        "        print(f\"{assignment.name} <{email}> - {rubric_ratings}\", file=sys.stderr)\n",
        "        raise\n",
        "\n",
        "params = {\n",
        "    'students': '1',\n",
        "    'type': 'latest-with-feedback',\n",
        "    'numbers': '0',\n",
        "    'scores': '0',\n",
        "    'score_type': 'passfail',\n",
        "    'feedback': '1',\n",
        "    'tz': 'America/Los_Angeles',\n",
        "}\n",
        "\n",
        "# https://canvas.instructure.com/doc/api/submissions.html#method.submissions_api.bulk_update\n",
        "async def grade_data(session, challenge, assignment):\n",
        "    async with session.post(\n",
        "        f\"https://us.edstem.org/api/challenges/{challenge}/results\",\n",
        "        params=params,\n",
        "        data={\"_token\": token}\n",
        "        ) as response:\n",
        "\n",
        "        payload = await response.text()\n",
        "        df = pd.read_csv(io.StringIO(payload)).set_index(\"email\")\n",
        "        return {\n",
        "            users[email]: {\n",
        "                \"posted_grade\": \"pass\" if passed else \"fail\",\n",
        "                \"rubric_assessment\": rubric_assessment(assignment, email, rubric_ratings)\n",
        "            }\n",
        "            for email, passed, *rubric_ratings\n",
        "            in zip(df.index, df[\"failed\"] == 0, *(\n",
        "                zip(itertools.repeat(rubric), df[rubric])\n",
        "                for rubric in rubrics if rubric in df\n",
        "            ))\n",
        "            if email in users\n",
        "        }\n",
        "\n",
        "async def main(mapping):\n",
        "    async with aiohttp.ClientSession() as session:\n",
        "        return await asyncio.gather(*[\n",
        "            grade_data(session, challenge, assignment)\n",
        "            for challenge, assignment in mapping.items()\n",
        "        ])\n",
        "        \n",
        "progress = [\n",
        "    assignment.submissions_bulk_update(grade_data=data)\n",
        "    for assignment, data in zip(assignments.values(), asyncio.run(main(assignments)))\n",
        "    if data\n",
        "]"
      ],
      "execution_count": null,
      "outputs": []
    }
  ]
}

No longer maintained. Use canvas_sync.py instead!