shoes_clustering.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Shoes images clustering\n",
    "\n",
    "The goal of this notebook is to combine shoes in pictures into big groups by their style with a help from crowd.\n",
    "\n",
    "We are going to use a simple dataset consists of 82 images from Toloka tutorial (Drutsa et al. 2020)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Prepare environment and import all we'll need."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!pip3 install toloka-kit==0.1.26\n",
    "!pip3 install pandas\n",
    "!pip3 install ipyplot"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import time\n",
    "import logging\n",
    "\n",
    "from datetime import datetime, timedelta\n",
    "from sys import stdout\n",
    "from time import sleep\n",
    "from typing import Dict, List, Optional, Set\n",
    "from random import sample, shuffle\n",
    "from getpass import getpass\n",
    "\n",
    "import ipyplot\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "\n",
    "from matplotlib import pyplot as plt\n",
    "from PIL import Image\n",
    "from urllib.request import urlopen\n",
    "\n",
    "import toloka.client as toloka\n",
    "\n",
    "from crowdkit.aggregation import DawidSkene\n",
    "import json\n",
    "from collections import Counter\n",
    "\n",
    "from exam_check import exam_check\n",
    "from calculate_quality import calculate_quality\n",
    "from crowd_clustering_aggregation import clustering_aggregation, Prior, AggregationAssignment\n",
    "\n",
    "logging.basicConfig(\n",
    "    format='[%(levelname)s] %(name)s: %(message)s',\n",
    "    level=logging.INFO,\n",
    "    stream=stdout,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Сreate toloka-client instance. All api calls will go through it. More about OAuth token in Toloka-Kit [Learn the basics example](https://github.com/Toloka/toloka-kit/tree/main/examples/0.getting_started/0.learn_the_basics)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "toloka_client = toloka.TolokaClient(getpass('Enter your token:'), 'PRODUCTION')  # Or switch to 'SANDBOX'\n",
    "logging.info(toloka_client.get_requester())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Creating Training and Exam project"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As our task is rather uncommon for crowdsourcing platforms we need our workers to train before proceeding to main pools. So, this is how it is going to work:\n",
    "\n",
    "* We create a Training and Exam project with training pool for workers to train and main pool to check workers' performance;\n",
    "* After training workers are going to pass an main pool at least 3 times as an exam;\n",
    "* After first exam worker is to be assigned a Shoes Clustering Skill, the skill value is calculated as percentage of correctly chosen clusters divided by number of exams expected (3 in our case);\n",
    "* Only those workers who have been assigned the Shoes Clustering Skill with 100% skill value are allowed to perform on the main project\n",
    "\n",
    "Let's start with creating Training and Exam project from a json file with the help of auxilliary functions:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def create_project_from_file(project_config_path: str, create: bool = True) -> toloka.Project:\n",
    "    with open(project_config_path) as project_config_file:\n",
    "        json_string = project_config_file.read()\n",
    "    project = toloka.Project.from_json(json_string)\n",
    "    if create:\n",
    "        return toloka_client.create_project(project)\n",
    "    else:\n",
    "        return project"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "training_exam_project = create_project_from_file('configs/shoes/training_exam/project.json')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Then we create training and add traininig tasks:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def create_training_from_file(training_config_path: str, project_id: str, create: bool = True) -> toloka.Training:\n",
    "    with open(training_config_path) as training_config_file:\n",
    "        json_string = training_config_file.read()\n",
    "    training = toloka.Training.from_json(json_string)\n",
    "    training.project_id = project_id\n",
    "    if create:\n",
    "        return toloka_client.create_training(training)\n",
    "    else:\n",
    "        return training"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "training = create_training_from_file('configs/shoes/training_exam/training.json', training_exam_project.id)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def create_tasks_from_directory(training_tasks_path: str, training_id: str, \n",
    "                                         create: bool = True) -> List[toloka.Task]:\n",
    "    tasks = []\n",
    "    \n",
    "    for training_task_filename in os.listdir(training_tasks_path):\n",
    "        training_task_file_path = os.path.join(training_tasks_path, training_task_filename)\n",
    "        if os.path.isfile(training_task_file_path) and training_task_file_path.endswith('.json'):\n",
    "            with open(training_task_file_path) as training_task_config_file:\n",
    "                json_string = training_task_config_file.read()\n",
    "            \n",
    "            task = toloka.Task.from_json(json_string)\n",
    "            task.pool_id = training_id\n",
    "            \n",
    "            tasks.append(task)\n",
    "    if create:\n",
    "        return toloka_client.create_tasks(tasks)\n",
    "    else:\n",
    "        return tasks"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "tasks = create_tasks_from_directory('configs/shoes/training_exam/training_tasks/', training.id)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "And finally create pool with tasks to be an exam for workers:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def create_pool_from_file(pool_config_path: str, project_id: str, \n",
    "                          training_id: Optional[str] = None, create: bool = True) -> toloka.Pool:\n",
    "    with open(pool_config_path) as pool_config_file:\n",
    "        json_string = pool_config_file.read()\n",
    "    pool = toloka.Pool.from_json(json_string)\n",
    "    pool.project_id = project_id\n",
    "    if training_id is not None:\n",
    "        pool.quality_control.training_requirement.training_pool_id = training_id\n",
    "    pool.will_expire = datetime.now() + timedelta(days=7)\n",
    "    if create:\n",
    "        return toloka_client.create_pool(pool)\n",
    "    else:\n",
    "        return pool"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "exam_pool = create_pool_from_file('configs/shoes/training_exam/exam_pool.json', training_exam_project.id, training.id)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def create_task_suites_from_directory(task_suites_path: str, pool_id: str) -> List[toloka.TaskSuite]:\n",
    "    task_suites = []\n",
    "    \n",
    "    for task_suite_file_name in os.listdir(task_suites_path):\n",
    "        task_suite_file_path = os.path.join(task_suites_path, task_suite_file_name)\n",
    "        if os.path.isfile(task_suite_file_path) and task_suite_file_path.endswith('.json'):\n",
    "            with open(task_suite_file_path) as task_config_file:\n",
    "                json_string = task_config_file.read()\n",
    "            \n",
    "            task_suite = toloka.Task.from_json(json_string)\n",
    "            task_suite.pool_id = pool_id\n",
    "            \n",
    "            task_suites.append(task_suite)\n",
    "    \n",
    "    return toloka_client.create_task_suites(task_suites, allow_defaults=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "task_suites = create_task_suites_from_directory('configs/shoes/training_exam/exam_tasks/', exam_pool.id)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Last thing we do on this Training and Exam project preparation is skill creation:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "skill = toloka_client.create_skill(name='Shoes clustering')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Our Training and Exam project is ready! But we won't open training and exam pools yet as we have payed exam. It is recommended to open payed exams exactly when they are needed and close when they aren't needed anymore."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Creating a main project\n",
    "\n",
    "Let's proceed on creating a main project."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "main_project = create_project_from_file('configs/shoes/project/project.json')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "main_pool = create_pool_from_file('configs/shoes/project/pool.json', main_project.id, create=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "There is one thing we have to do manually: in a worker's filter we need to change skill id to the one we've created"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "main_pool.filter.and_[0].or_[0].key=skill.id\n",
    "main_pool.quality_control.configs[3].rules[0].action.parameters.skill_id = skill.id\n",
    "main_pool=toloka_client.create_pool(main_pool)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we create task suites:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data = pd.read_csv('configs/shoes/project/dataset.tsv')\n",
    "images = set(x[0] for x in data.values)\n",
    "\n",
    "objects_per_hit_number = 10 # number of objects in each HIT\n",
    "objects_number = len(images) # number of objects\n",
    "expecting_hits = min(objects_per_hit_number, 21) # expecting number of HITs to which a data item belongs\n",
    "\n",
    "initial_items = max(objects_per_hit_number // expecting_hits, 1)\n",
    "unique_hits = objects_number * expecting_hits // objects_per_hit_number\n",
    "print('Initial items:', initial_items, '\\nUnique HITs:', unique_hits)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def hit_samples(images: Set[str], unique_hits: int, objects_number: int, \n",
    "           objects_per_hit_number: int, initial_items: int) -> List[toloka.Task]:\n",
    "    hits = np.array_split(list(images), min(objects_number, unique_hits))\n",
    "    for i, hit in enumerate(hits):\n",
    "        not_in_hit = images - set(hit)\n",
    "        hit_sample = sample(not_in_hit, objects_per_hit_number - initial_items)\n",
    "        hits[i] = np.append(hit, hit_sample)\n",
    "    return hits\n",
    "\n",
    "hits = hit_samples(images, unique_hits, objects_number, objects_per_hit_number, initial_items)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "task_suites = [\n",
    "    toloka.task_suite.TaskSuite(\n",
    "        pool_id=main_pool.id,\n",
    "        tasks=[\n",
    "            toloka.task.Task(input_values={'images': hit.tolist()})\n",
    "        ]\n",
    "    )\n",
    "    for hit in hits\n",
    "]\n",
    "\n",
    "task_suites = toloka_client.create_task_suites(task_suites, allow_defaults=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Receiving responses\n",
    "\n",
    "So, we have finished all preparations, now is the time to start labelling.\n",
    "\n",
    "We are going to open all our pools: training, exam and main pool. All pools will stay open untill the main pool is finished."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "period: timedelta = timedelta(seconds=60)\n",
    "\n",
    "training = toloka_client.open_training(training.id)\n",
    "exam_pool = toloka_client.open_pool(exam_pool.id)\n",
    "main_pool = toloka_client.open_pool(main_pool.id)\n",
    "\n",
    "while main_pool.is_open():\n",
    "    exam_check(toloka_client, exam_pool, skill)\n",
    "    op = toloka_client.get_analytics([toloka.analytics_request.CompletionPercentagePoolAnalytics(subject_id=main_pool.id)])\n",
    "    percentage = toloka_client.wait_operation(op).details['value'][0]['result']['value']\n",
    "    logging.info(f'Pool {main_pool.id} - {percentage}%')\n",
    "\n",
    "    sleep(period.total_seconds())\n",
    "    main_pool = toloka_client.get_pool(main_pool.id)\n",
    "\n",
    "exam_check(toloka_client, exam_pool, skill)\n",
    "training = toloka_client.close_training(training.id)\n",
    "exam_pool = toloka_client.close_pool(exam_pool.id)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Aggregation\n",
    "\n",
    "After getting our data labelled, we need to aggregate our data using our implementation of algorithm described in [Crowdclustering by Ryan Gomes et. al.](http://vision.caltech.edu/~gomes/papers/crowd_clust_final.pdf) and [Incremental Learning of Nonparametric Bayesian Mixture Models by Ryan Gomes et. al.](http://www.vision.caltech.edu/gomes/papers/gomes_cvpr_08.pdf). Our implementation is based on [original Matlab implementation](http://www.vision.caltech.edu/gomes/software.html).\n",
    "\n",
    "It will take quite a while for aggregation process to converge."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "assignments_raw = toloka_client.get_assignments_df(pool_id=main_pool.id)[\n",
    "        ['INPUT:images', 'OUTPUT:result', 'GOLDEN:result', 'ASSIGNMENT:link', 'ASSIGNMENT:assignment_id',\n",
    "         'ASSIGNMENT:worker_id', 'ASSIGNMENT:status']]\n",
    "prior = Prior(1, 5, 10, 1)\n",
    "cluster_dict, id_to_img = clustering_aggregation(assignments_raw, 'INPUT:images', prior)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Finally, let's output results of clustering by crowd:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def draw_cluster(images):\n",
    "    step = 30\n",
    "    for i in range(0, len(images), step):\n",
    "        part = images[i:i+step]\n",
    "        plt.figure(figsize=(100, 100))\n",
    "        for i, image_name in enumerate(part):\n",
    "            ax = plt.subplot(len(images) // 5 + 1, 5, i + 1)\n",
    "            img = Image.open(urlopen(image_name))\n",
    "            im = ax.imshow(img)\n",
    "            ax.axis('off')\n",
    "        plt.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "for i, cluster in cluster_dict.items():\n",
    "    print(('-------------' + str(i) + '-------------').center(100))\n",
    "    draw_cluster(cluster)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "pycharm": {
     "name": "#%% md\n"
    }
   },
   "source": [
    "# Quality evaluation"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "After getting results, we can evaluate quality of clusters. To do this we use appoach called <i>Intruders</i> described in https://proceedings.neurips.cc/paper/2009/hash/f92586a25bb3145facd64ab20fd554ff-Abstract.html."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The idea is to make HITs with one image from one cluster and others from another cluster. We ask worker to find this extra image and if this object was found, then objects from these two clusters are quite different from each other. The proportion of correct answers will be \"quality\" in this case."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def make_intruder_hits(cluster_dict, N_HITS=100, images_on_page=8) -> List[List[str]]:\n",
    "    hits_list = list()\n",
    "    for _ in range(N_HITS):\n",
    "        cluster, intrusor_cluster = sample(cluster_dict.keys(), 2)\n",
    "        images = cluster_dict[cluster]\n",
    "        intrusor_images = cluster_dict[intrusor_cluster]\n",
    "        while len(images) < images_on_page - 1:\n",
    "            cluster, intrusor_cluster = sample(cluster_dict.keys(), 2)\n",
    "            images = cluster_dict[cluster]\n",
    "            intrusor_images = cluster_dict[intrusor_cluster]\n",
    "        representatives = sample(images, images_on_page - 1)\n",
    "        intrusor = sample(intrusor_images, 1)\n",
    "        HIT = representatives + intrusor\n",
    "        shuffle(HIT)\n",
    "        hits_list.append(HIT)\n",
    "    return hits_list"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "N_HITS = 100\n",
    "images_on_page = 8\n",
    "intruder_hits = make_intruder_hits(cluster_dict, N_HITS, images_on_page)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create main project and training"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "intruders_project = create_project_from_file('configs/shoes/quality_evaluation/project/project.json')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "intruders_training = create_training_from_file(\n",
    "    'configs/shoes/quality_evaluation/project/training.json', intruders_project.id\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "tasks = create_tasks_from_directory('configs/shoes/quality_evaluation/training_tasks/', intruders_training.id)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "intruders_pool = create_pool_from_file(\n",
    "    'configs/shoes/quality_evaluation/project/pool.json', intruders_project.id, intruders_training.id\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "task_suites = [\n",
    "    toloka.task_suite.TaskSuite(\n",
    "        pool_id=intruders_pool.id,\n",
    "        tasks=[\n",
    "            toloka.task.Task(input_values={'images': hit})\n",
    "        ]\n",
    "    )\n",
    "    for hit in intruder_hits\n",
    "]\n",
    "\n",
    "task_suites = toloka_client.create_task_suites(task_suites, allow_defaults=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Receiving responses"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "period: timedelta = timedelta(seconds=60)\n",
    "\n",
    "intruders_training = toloka_client.open_training(intruders_training.id)\n",
    "intruders_pool = toloka_client.open_pool(intruders_pool.id)\n",
    "\n",
    "while intruders_pool.is_open():\n",
    "    op = toloka_client.get_analytics(\n",
    "        [toloka.analytics_request.CompletionPercentagePoolAnalytics(subject_id=intruders_pool.id)]\n",
    "    )\n",
    "    percentage = toloka_client.wait_operation(op).details['value'][0]['result']['value']\n",
    "    logging.info(f'Pool {intruders_pool.id} - {percentage}%')\n",
    "\n",
    "    sleep(period.total_seconds())\n",
    "    intruders_pool = toloka_client.get_pool(intruders_pool.id)\n",
    "\n",
    "intruders_training = toloka_client.close_training(intruders_training.id)\n",
    "intruders_pool = toloka_client.close_pool(intruders_pool.id)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "assignments_raw = toloka_client.get_assignments_df(pool_id=intruders_pool.id)[\n",
    "        ['INPUT:images', 'OUTPUT:answer', 'GOLDEN:answer', 'ASSIGNMENT:link', 'ASSIGNMENT:assignment_id',\n",
    "         'ASSIGNMENT:worker_id', 'ASSIGNMENT:status']]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "calculate_quality(assignments_raw, cluster_dict, intruder_hits)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}