complexity_pipeline.ipynb

{
 "cells": [
  {
   "metadata": {},
   "cell_type": "markdown",
   "source": [
    "# OPTIMAL-EM: Accessibility and Complexity Analysis Pipeline\n",
    "\n",
    "This notebook processes a set (target population) of HTML files to perform accessibility analysis using Pa11y with the Axe runner, and complexity analysis. It extracts features from the HTML content, performs dimensionality reduction and clustering, and calculates correlation matrices between accessibility barriers and complexity metrics.\n"
   ],
   "id": "d5dec61831e256aa"
  },
  {
   "metadata": {},
   "cell_type": "markdown",
   "source": [
    "## Notes\n",
    "\n",
    "- **External Dependencies**:\n",
    "  - Check the import statements to ensure that all required libraries are installed. \n",
    "  - **Pa11y**: Ensure that Pa11y is installed and accessible in your environment.\n",
    "    - Install [Pa11y](https://github.com/pa11y/pa11y): `npm install -g pa11y`\n",
    "    - Ensure the Axe runner is available: Pa11y uses [htmlcs](https://github.com/squizlabs/HTML_CodeSniffer) for accessibility checks by default.\n",
    "- **Data Paths**:\n",
    "  - Update `HTML_PATH` if your HTML files are located in a different directory.\n",
    "  - Update `DATA_FILE` if you wish to save the processed data to a different location.\n",
    "- **Visualisation**:\n",
    "  - The code generates plots using Matplotlib and Seaborn.\n",
    "- **Re-running Accessibility Analysis**:\n",
    "  - Set `rerun_accessibility=True` in `process_accessibility_results()` if you want to run a re-analysis of accessibility results (e.g. this is your first time running the notebook or you have updated the HTML files).\n"
   ],
   "id": "cf26c93108ccbc20"
  },
  {
   "metadata": {},
   "cell_type": "code",
   "source": [
    "import os\n",
    "import subprocess\n",
    "import json\n",
    "from collections import Counter\n",
    "from collections.abc import Callable\n",
    "from concurrent.futures import as_completed, ProcessPoolExecutor\n",
    "from typing import Any\n",
    "\n",
    "import pandas as pd\n",
    "import matplotlib.pyplot as plt\n",
    "import seaborn as sns\n",
    "from bs4 import BeautifulSoup\n",
    "from sklearn.cluster import DBSCAN\n",
    "from sklearn.feature_extraction.text import CountVectorizer\n",
    "from sklearn.manifold import TSNE\n",
    "from sklearn.preprocessing import StandardScaler"
   ],
   "id": "4cf19877056cded1",
   "outputs": [],
   "execution_count": null
  },
  {
   "metadata": {},
   "cell_type": "markdown",
   "source": [
    "## Constants:\n",
    "\n",
    "- **HTML_PATH**: The path to the folder containing the HTML files.\n",
    "- **DATA_FILE**: The path to save the processed data.\n",
    "- **RICH_CONTENT_TAGS**: A list of HTML tags considered as rich content.\n",
    "- **INLINE_TAGS**: A list of inline elements."
   ],
   "id": "b929e28767269017"
  },
  {
   "metadata": {},
   "cell_type": "code",
   "source": [
    "# The path to the folder containing the HTML files:\n",
    "HTML_PATH = 'pages/res/'\n",
    "\n",
    "# The path to save the processed data:\n",
    "DATA_FILE = 'collected_data_step2_ec.json'\n",
    "\n",
    "RICH_CONTENT_TAGS = ['a', 'audio', 'button', 'canvas', 'embed', 'iframe', 'img', 'input', 'keygen', 'label', 'math',\n",
    "                     'object', 'select', 'svg', 'textarea', 'video']\n",
    "\n",
    "INLINE_TAGS = [\n",
    "    'a', 'abbr', 'acronym', 'b', 'bdo', 'big', 'br', 'button',\n",
    "    'cite', 'code', 'dfn', 'em', 'i', 'img', 'input', 'kbd', 'label',\n",
    "    'map', 'object', 'output', 'q', 'samp', 'script', 'select',\n",
    "    'small', 'span', 'strong', 'sub', 'sup', 'textarea', 'time', 'tt', 'var'\n",
    "]"
   ],
   "id": "e17d47df995cb0ab",
   "outputs": [],
   "execution_count": null
  },
  {
   "metadata": {},
   "cell_type": "markdown",
   "source": [
    "## Helper Methods\n",
    "\n",
    "### `get_html_structure(soup)`\n",
    "\n",
    "Returns all HTML tags in a string format (e.g., `'head meta div div div'`).\n",
    "\n",
    "### `get_html_block_structure(soup)`\n",
    "\n",
    "Returns only block-level HTML tags in a string format, excluding inline tags (e.g., `'div div div'`).\n"
   ],
   "id": "ed50b1663f652a"
  },
  {
   "metadata": {},
   "cell_type": "code",
   "source": [
    "def get_html_structure(soup):\n",
    "    \"\"\"Returns the HTML tags in a string, e.g., 'head meta div div div'.\"\"\"\n",
    "    tags = \" \".join(tag.name for tag in soup.find_all())\n",
    "    return tags\n",
    "\n",
    "def get_html_block_structure(soup):\n",
    "    \"\"\"Returns only block-level HTML tags in a string, e.g., 'div div div'.\"\"\"\n",
    "    tags = \" \".join(tag.name for tag in soup.find_all() if tag.name not in INLINE_TAGS)\n",
    "    return tags"
   ],
   "id": "4987e2b16fd79558",
   "outputs": [],
   "execution_count": null
  },
  {
   "metadata": {},
   "cell_type": "markdown",
   "source": [
    "## Compute Accessibility Results\n",
    "\n",
    "### `get_accessibility_results(filename, output)`\n",
    "\n",
    "Computes the accessibility results based on impact levels for a given HTML file using Pa11y output. Axe produces accessibility results categorised as critical, serious, moderate, or minor."
   ],
   "id": "3237a6d9174a2236"
  },
  {
   "metadata": {},
   "cell_type": "code",
   "source": [
    "def get_accessibility_results(filename: str, output: list[dict[str, dict[str, str]]]) -> list[int]:\n",
    "    \"\"\"\n",
    "    Computes the accessibility results based on impact levels for a given .html file.\n",
    "    Axe produces accessibility results categorised as critical, serious, moderate, or minor.\n",
    "\n",
    "    :param filename: str: The name of the .html file. Used for logging.\n",
    "    :param output: list[dict[str, dict[str, str]]]: The list of barriers, each containing a 'runnerExtras' key with an\n",
    "        'impact' field.\n",
    "    :return: list[int, int, int, int]: Counts of barriers categorised as [critical, serious, moderate, minor].\n",
    "    \"\"\"\n",
    "    print(f\"Calculating accessibility for {filename}:\")\n",
    "\n",
    "    impacts = [issue['runnerExtras']['impact'] for issue in output]\n",
    "    impact_counter = Counter(impacts)\n",
    "\n",
    "    return [\n",
    "        impact_counter.get('critical', 0),\n",
    "        impact_counter.get('serious', 0),\n",
    "        impact_counter.get('moderate', 0),\n",
    "        impact_counter.get('minor', 0)\n",
    "    ]"
   ],
   "id": "34330d6c00cd2819",
   "outputs": [],
   "execution_count": null
  },
  {
   "metadata": {},
   "cell_type": "code",
   "source": [
    "def process_html_file(filename: str) -> dict[str, Any]:\n",
    "    \"\"\"\n",
    "    Processes a single HTML file for accessibility results and HTML content.\n",
    "\n",
    "    :param filename: str: The name of the HTML file to process.\n",
    "    :return: dict[str, Any]: A dictionary containing filename, accessibility results, and HTML content.\n",
    "    \"\"\"\n",
    "    file_path = os.path.join(HTML_PATH, filename)\n",
    "    print(f\"Processing: {filename}\")\n",
    "\n",
    "    # Run Pa11y as a subprocess with Axe runner:\n",
    "    process = subprocess.Popen(\n",
    "        ['pa11y', '--runner', 'axe', '--reporter', 'json', file_path],\n",
    "        stdout=subprocess.PIPE,\n",
    "        stderr=subprocess.PIPE,\n",
    "        universal_newlines=True\n",
    "    )\n",
    "\n",
    "    try:\n",
    "        stdout_line = process.stdout.readline().strip()\n",
    "        if not stdout_line:\n",
    "            accessibility_results = [0, 0, 0, 0]\n",
    "        else:\n",
    "            pa11y_output = json.loads(stdout_line)\n",
    "            accessibility_results = get_accessibility_results(filename, pa11y_output)\n",
    "    except (json.JSONDecodeError, Exception) as e:\n",
    "        print(f\"Error processing Pa11y output for {filename}: {e}\")\n",
    "        accessibility_results = [0, 0, 0, 0]\n",
    "\n",
    "    with open(file_path, 'r', encoding='utf-8') as f:\n",
    "        content = f.read()\n",
    "\n",
    "    return {\n",
    "        'filename': filename,\n",
    "        'accessibility': accessibility_results,\n",
    "        'html_content': content\n",
    "    }"
   ],
   "id": "913f4e14098e7444",
   "outputs": [],
   "execution_count": null
  },
  {
   "metadata": {},
   "cell_type": "code",
   "source": [
    "def process_accessibility_results(rerun_accessibility: bool = False) -> list[dict[str, Any]]:\n",
    "    \"\"\"\n",
    "    Collects data from HTML files, runs accessibility analysis using Pa11y if needed, and saves the results to a JSON\n",
    "    file.\n",
    "\n",
    "    :param rerun_accessibility: bool: Flag to indicate whether to rerun the accessibility analysis.\n",
    "    :return: a list of dictionaries where each dictionary contains:\n",
    "        - 'filename' (str): The name of the HTML file.\n",
    "        - 'accessibility' (list[int]): Accessibility results categorised as [critical, serious, moderate, minor].\n",
    "        - 'html_content' (str): The HTML content of the file.\n",
    "    \"\"\"\n",
    "    if not rerun_accessibility and os.path.exists(DATA_FILE):\n",
    "        print(\"Loading data from JSON file...\")\n",
    "        with open(DATA_FILE, 'r', encoding='utf-8') as f:\n",
    "            data = json.load(f)\n",
    "        return data\n",
    "    else:\n",
    "        print(\"Collecting data and running accessibility analysis...\")\n",
    "        data = []\n",
    "\n",
    "        with ProcessPoolExecutor() as executor:\n",
    "            futures = [executor.submit(process_html_file, filename)\n",
    "                       for filename in os.listdir(HTML_PATH)\n",
    "                       if filename.endswith('.html') or filename.endswith('.htm')]\n",
    "\n",
    "            for future in as_completed(futures):\n",
    "                try:\n",
    "                    result = future.result()\n",
    "                    data.append(result)\n",
    "                except Exception as e:\n",
    "                    print(f\"Error processing file: {e}\")\n",
    "\n",
    "        print(\"Saving data to JSON file...\")\n",
    "        with open(DATA_FILE, 'w', encoding='utf-8') as f:\n",
    "            json.dump(data, f)\n",
    "\n",
    "        return data"
   ],
   "id": "9c10fdca1575d4e",
   "outputs": [],
   "execution_count": null
  },
  {
   "metadata": {},
   "cell_type": "markdown",
   "source": [
    "## Method to Run Analysis\n",
    "\n",
    "### `run_analysis(data, feature_extraction_func)`\n",
    "\n",
    "Runs the analysis using the provided HTML files and feature extraction function. The analysis includes:\n",
    "\n",
    "1. HTML parsing\n",
    "2. Feature extraction\n",
    "3. Dimensionality reduction\n",
    "4. Clustering\n",
    "5. Calculation of a correlation matrix for accessibility and complexity metrics\n",
    "\n"
   ],
   "id": "2adafd5e11e9656e"
  },
  {
   "metadata": {},
   "cell_type": "code",
   "source": [
    "def run_analysis(\n",
    "    data: list[dict[str, Any]],\n",
    "    feature_extraction_func: Callable[[BeautifulSoup], str]\n",
    ") -> pd.DataFrame:\n",
    "    \"\"\"\n",
    "    Runs the analysis using the provided .html files and feature extraction function. The analysis includes (i) HTML\n",
    "    parsing, (ii) feature extraction, (iii) dimensionality reduction, (iv) clustering, and (v) the calculation of a\n",
    "    correlation matrix for accessibility and complexity metrics.\n",
    "\n",
    "    :param data: list[dict[str, Any]]\n",
    "        A list of dictionaries containing 'filename', 'accessibility', and 'html_content' fields:\n",
    "        - 'filename' (str): The name of the HTML file.\n",
    "        - 'accessibility' (list[int]): Accessibility results as [critical, serious, moderate, minor].\n",
    "        - 'html_content' (str): The HTML content of the file.\n",
    "\n",
    "    :param feature_extraction_func: Callable[[BeautifulSoup], str]\n",
    "        A function that takes a BeautifulSoup object (parsed HTML) and returns a string representing the extracted\n",
    "        features (e.g., HTML tags).\n",
    "\n",
    "    :return: pd.DataFrame\n",
    "        A correlation matrix showing the relationships between the normalised metrics, including complexity and\n",
    "        accessibility barriers.\n",
    "    \"\"\"\n",
    "    filenames = []\n",
    "    accessibility_data = []\n",
    "    complexity_data = []\n",
    "    html_structure_texts = []\n",
    "\n",
    "    for item in data:\n",
    "        filenames.append(item['filename'])\n",
    "        accessibility_data.append(item['accessibility'])\n",
    "\n",
    "        soup = BeautifulSoup(item['html_content'], 'html.parser')\n",
    "\n",
    "        tags_text = feature_extraction_func(soup)\n",
    "        html_structure_texts.append(tags_text)\n",
    "\n",
    "        all_tags = soup.find_all()\n",
    "        rich_content_tags = soup.find_all(RICH_CONTENT_TAGS)\n",
    "        complexity = len(rich_content_tags) / len(all_tags) if len(all_tags) > 0 else 0\n",
    "        complexity_data.append(complexity)\n",
    "\n",
    "    vectorizer = CountVectorizer()\n",
    "    dtm = vectorizer.fit_transform(html_structure_texts)\n",
    "\n",
    "    print(\"Running t-SNE...\")\n",
    "    tsne = TSNE(perplexity=20)\n",
    "    dtm_tsne = tsne.fit_transform(dtm.toarray())\n",
    "\n",
    "    print(\"Running DBSCAN clustering...\")\n",
    "    dbscan = DBSCAN(eps=4, min_samples=2)\n",
    "    clusters = dbscan.fit_predict(dtm_tsne)\n",
    "\n",
    "    plt.figure(figsize=(10, 8))\n",
    "\n",
    "    unique_clusters = set(clusters)\n",
    "    colors = plt.cm.get_cmap('tab10', len(unique_clusters))\n",
    "\n",
    "    for cluster in unique_clusters:\n",
    "        # DBSCAN defines noise as -1:\n",
    "        if cluster == -1:\n",
    "            color = 'k'\n",
    "            marker = 'x'\n",
    "        else:\n",
    "            color = colors(cluster)\n",
    "            marker = 'o'\n",
    "\n",
    "        cluster_points = dtm_tsne[clusters == cluster]\n",
    "\n",
    "        plt.scatter(cluster_points[:, 0], cluster_points[:, 1], c=[color],\n",
    "                    label=f'Cluster {cluster}' if cluster != -1 else 'Noise', marker=marker)\n",
    "\n",
    "    plt.title('DBSCAN')\n",
    "    plt.xlabel('t-SNE Dimension 1')\n",
    "    plt.ylabel('t-SNE Dimension 2')\n",
    "    plt.legend(loc='best')\n",
    "    plt.show()\n",
    "\n",
    "    cluster_data = pd.DataFrame({\n",
    "        'filename': filenames,\n",
    "        'cluster': clusters,\n",
    "        'complexity': complexity_data,\n",
    "        'accessibility': accessibility_data\n",
    "    })\n",
    "\n",
    "    # Expand 'accessibility' into separate columns:\n",
    "    accessibility_df = pd.DataFrame(cluster_data['accessibility'].tolist(),\n",
    "                                    columns=['critical', 'serious', 'moderate', 'minor'])\n",
    "\n",
    "    cluster_data = pd.concat([cluster_data.drop('accessibility', axis=1), accessibility_df], axis=1)\n",
    "    cluster_data['total_barriers'] = cluster_data[['critical', 'serious', 'moderate', 'minor']].sum(axis=1)\n",
    "\n",
    "    # Group by cluster and compute the mean and variance:\n",
    "    cluster_stats = cluster_data.groupby('cluster').agg({\n",
    "        'complexity': ['mean', 'var'],\n",
    "        'total_barriers': 'mean',\n",
    "        'critical': 'mean',\n",
    "        'serious': 'mean',\n",
    "        'moderate': 'mean',\n",
    "        'minor': 'mean'\n",
    "    }).reset_index()\n",
    "\n",
    "    cluster_stats.columns = [\n",
    "        'Cluster', 'Avg. Complexity', 'Complexity Variance',\n",
    "        'Total Barriers', 'Critical Barriers', 'Serious Barriers',\n",
    "        'Moderate Barriers', 'Minor Barriers'\n",
    "    ]\n",
    "\n",
    "    # Flatten multi-level columns:\n",
    "    cluster_stats.columns = ['_'.join(col).strip() if isinstance(col, tuple) else col for col in cluster_stats.columns]\n",
    "\n",
    "    columns_to_normalize = [\n",
    "        'Avg. Complexity',\n",
    "        'Complexity Variance',\n",
    "        'Total Barriers',\n",
    "        'Critical Barriers',\n",
    "        'Serious Barriers',\n",
    "        # 'Moderate Barriers',\n",
    "        # 'Minor Barriers'\n",
    "    ]\n",
    "\n",
    "    scaler = StandardScaler()\n",
    "    normalized_data = scaler.fit_transform(cluster_stats[columns_to_normalize])\n",
    "    normalized_df = pd.DataFrame(normalized_data, columns=columns_to_normalize)\n",
    "    correlation_matrix = normalized_df.corr()\n",
    "    correlation_matrix.to_csv('correlation_matrix.csv', index=False)\n",
    "\n",
    "    # Optionally, visualise the correlation matrix:\n",
    "    plt.figure(figsize=(12, 12))\n",
    "    sns.heatmap(correlation_matrix, cmap='coolwarm', fmt='.2f', annot=True)\n",
    "    plt.title('Correlation Matrix')\n",
    "    plt.show()\n",
    "\n",
    "    return correlation_matrix"
   ],
   "id": "e0f369f1099e283",
   "outputs": [],
   "execution_count": null
  },
  {
   "metadata": {},
   "cell_type": "markdown",
   "source": [
    "## Run the Analysis\n",
    "\n",
    "We will now process the accessibility results and run the analysis using the `get_html_structure` function for feature extraction."
   ],
   "id": "74de42bb139f9945"
  },
  {
   "metadata": {},
   "cell_type": "code",
   "source": [
    "data = process_accessibility_results(rerun_accessibility=False)\n",
    "correlation_matrix_1 = run_analysis(data, get_html_structure)"
   ],
   "id": "49029c5c34ae6261",
   "outputs": [],
   "execution_count": null
  },
  {
   "metadata": {},
   "cell_type": "markdown",
   "source": [
    "## Optional: Run Analysis with Block-Level HTML Structure\n",
    "\n",
    "You can uncomment the code below to run the analysis using block-level HTML structures and compare the results.\n"
   ],
   "id": "dba59cabd97cc270"
  },
  {
   "metadata": {},
   "cell_type": "code",
   "source": [
    "# correlation_matrix_2 = run_analysis(data, get_html_block_structure)\n",
    "\n",
    "# difference_matrix = correlation_matrix_2 - correlation_matrix_1\n",
    "# plt.figure(figsize=(14, 14))\n",
    "# sns.heatmap(difference_matrix, annot=True, cmap='coolwarm', fmt='.2f', center=0)\n",
    "# plt.title('Difference of Correlation Matrices')\n",
    "# plt.show()"
   ],
   "id": "5217e389f30c3429",
   "outputs": [],
   "execution_count": null
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}