diff options
| author | Adam Janovsky | 2023-06-08 13:39:32 +0200 |
|---|---|---|
| committer | Adam Janovsky | 2023-06-08 13:39:32 +0200 |
| commit | 617fa6c71324624f5b4ba9a8693740bb173871bb (patch) | |
| tree | 97d1cf639164bbdcbe093d4629a6ef131f603ca7 | |
| parent | 902993a6c3afa8a29274307518c31307f7696b88 (diff) | |
| download | sec-certs-617fa6c71324624f5b4ba9a8693740bb173871bb.tar.gz sec-certs-617fa6c71324624f5b4ba9a8693740bb173871bb.tar.zst sec-certs-617fa6c71324624f5b4ba9a8693740bb173871bb.zip | |
refactor reference notebook
| -rw-r--r-- | notebooks/cc/references.ipynb | 559 | ||||
| -rw-r--r-- | notebooks/fixed_sankey_plot.py | 400 |
2 files changed, 480 insertions, 479 deletions
diff --git a/notebooks/cc/references.ipynb b/notebooks/cc/references.ipynb index c6d2ddb0..15b2a1f4 100644 --- a/notebooks/cc/references.ipynb +++ b/notebooks/cc/references.ipynb @@ -1,6 +1,7 @@ { "cells": [ { + "attachments": {}, "cell_type": "markdown", "metadata": { "collapsed": true, @@ -36,7 +37,12 @@ "import pandas as pd\n", "import seaborn as sns\n", "import numpy as np\n", - "from pysankey import sankey\n", + "import sys\n", + "from pathlib import Path\n", + "\n", + "\n", + "sys.path.insert(0, \"./../\")\n", + "from fixed_sankey_plot import sankey\n", "\n", "%matplotlib inline\n", "\n", @@ -68,7 +74,10 @@ "#sns.set_palette(\"deep\")\n", "#sns.set_context(\"notebook\") # Set to \"paper\" for use in paper :)\n", "\n", - "#plt.rcParams['figure.figsize'] = (10, 6)" + "#plt.rcParams['figure.figsize'] = (10, 6)\n", + "\n", + "RESULTS_DIR = Path(\"./results/references\")\n", + "RESULTS_DIR.mkdir(exist_ok=True, parents=True)" ] }, { @@ -96,6 +105,7 @@ ] }, { + "attachments": {}, "cell_type": "markdown", "metadata": { "pycharm": { @@ -107,6 +117,7 @@ ] }, { + "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ @@ -132,43 +143,20 @@ "df[\"has_outgoing_indirect_references\"] = df.indirectly_referencing.notnull()\n", "df[\"has_incoming_indirect_references\"] = df.indirectly_referenced_by.notnull()\n", "\n", - "#df.loc[:, [\"directly_referenced_by\", \"indirectly_referenced_by\", \"directly_referencing\", \"indirectly_referencing\"]].notnull().describe()\n", - "\n", - "print(f\"\\\\newcommand{{\\\\numCcAllDirectReferencing}}{{{df.has_outgoing_direct_references.sum()}}}\")\n", - "print(f\"\\\\newcommand{{\\\\numCcAllNotDirectReferencing}}{{{len(df) - df.has_outgoing_direct_references.sum()}}}\")\n", - "\n", "df_id_rich[\"has_outgoing_direct_references\"] = df_id_rich.directly_referencing.notnull()\n", "df_id_rich[\"has_incoming_direct_references\"] = df_id_rich.directly_referenced_by.notnull()\n", "df_id_rich[\"has_outgoing_indirect_references\"] = df_id_rich.indirectly_referencing.notnull()\n", "df_id_rich[\"has_incoming_indirect_references\"] = df_id_rich.indirectly_referenced_by.notnull()\n", "\n", - "print(f\"\\\\newcommand{{\\\\numCcWithIdDirectReferencing}}{{{df_id_rich.has_outgoing_direct_references.sum()}}}\")\n", - "print(f\"\\\\newcommand{{\\\\numCcWithIdNotDirectReferencing}}{{{len(df_id_rich) - df_id_rich.has_outgoing_direct_references.sum()}}}\")\n", - "\n", - "#df_id_rich.loc[:, [\"directly_referenced_by\", \"indirectly_referenced_by\", \"directly_referencing\", \"indirectly_referencing\"]].notnull().describe()" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "print(f\"\\\\newcommand{{\\\\numCCActiveDirectReferencing}}{{{df_id_rich.loc[df_id_rich.status == 'active'].has_outgoing_direct_references.sum()}}}\")\n", - "\n", - "archived_cert_id_list = set(df_id_rich[df_id_rich.status == \"archived\"].cert_id)\n", - "def contains_archived_cert_reference(referencing):\n", - " if pd.isnull(referencing):\n", - " return False\n", - " return bool(archived_cert_id_list.intersection(referencing))\n", - "print(f\"\\\\newcommand{{\\\\numCCActiveDirectReferencingArchived}}{{{df_id_rich[df_id_rich.status == 'active'].directly_referencing.apply(contains_archived_cert_reference).sum()}}}\")" + "archived_cert_id_list = set(df_id_rich[df_id_rich.status == \"archived\"].cert_id)" ] }, { + "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ - "### Plot direct references per category" + "### Plot direct references per category (count plot)" ] }, { @@ -195,7 +183,18 @@ " title=f\"Countplot of {' '.join(col.split('_'))}\",\n", " )\n", " countplot.tick_params(axis=\"x\", rotation=90)\n", - " countplot.legend(title=' '.join(col.split('_')), bbox_to_anchor=(1.05, 1), loc=2, borderaxespad=0.0)\n" + " countplot.legend(title=' '.join(col.split('_')), bbox_to_anchor=(1.05, 1), loc=2, borderaxespad=0.0)\n", + "\n", + "plt.savefig(str(RESULTS_DIR / \"references_countplot.pdf\"), bbox_inches=\"tight\")\n", + "plt.close(figure)\n" + ] + }, + { + "attachments": {}, + "cell_type": "markdown", + "metadata": {}, + "source": [ + "### Plot direct references per category (Sankey diagram)" ] }, { @@ -216,10 +215,6 @@ "exploded = exploded.loc[exploded.ref_category.notnull()]\n", "\n", "exploded_with_refs = exploded.loc[exploded.ref_category != \"No references\"]\n", - "print(f\"\\\\newcommand{{\\\\numCCDirectRefsSameCategory}}{{{(exploded_with_refs.category == exploded_with_refs.ref_category).sum()}}}\")\n", - "print(f\"\\\\newcommand{{\\\\numCCDirectRefsOtherCategory}}{{{(exploded_with_refs.category != exploded_with_refs.ref_category).sum()}}}\")\n", - "print(f\"\\\\newcommand{{\\\\numCCDirectRefs}}{{{len(exploded_with_refs)}}}\")\n", - "print(f\"\\\\newcommand{{\\\\numCCDirectRefsFromSmartcards}}{{{(exploded_with_refs.category == 'ICs, Smart Cards and Smart Card-Related Devices and Systems').sum()}}}\")\n", "\n", "all_categories = set(exploded.category.unique()) | set(exploded.ref_category.unique())\n", "colors = list(sns.color_palette(\"hls\", len(all_categories), as_cmap=False).as_hex())\n", @@ -231,16 +226,17 @@ "\n", "sankey(exploded.category, exploded.ref_category, colorDict=color_dict, leftLabels=list(exploded.category.unique()), rightLabels=list(exploded.ref_category.unique()), fontsize=12, ax=axes)\n", "\n", - "figure.savefig(\"category_references.pdf\", bbox_inches=\"tight\")\n", - "figure.savefig(\"category_references.pgf\", bbox_inches=\"tight\")\n", + "figure.savefig(str(RESULTS_DIR / \"category_references.pdf\"), bbox_inches=\"tight\")\n", + "figure.savefig(str(RESULTS_DIR / \"category_references.pgf\"), bbox_inches=\"tight\")\n", "plt.close(figure)" ] }, { + "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ - "### Plot direct references per scheme" + "### Plot direct references per scheme (count plot)" ] }, { @@ -267,14 +263,18 @@ " title=f\"Countplot of {' '.join(col.split('_'))}\",\n", " )\n", " countplot.tick_params(axis=\"x\", rotation=90)\n", - " countplot.legend(title=' '.join(col.split('_')), bbox_to_anchor=(1.05, 1), loc=2, borderaxespad=0.0)" + " countplot.legend(title=' '.join(col.split('_')), bbox_to_anchor=(1.05, 1), loc=2, borderaxespad=0.0)\n", + "\n", + "plt.savefig(str(RESULTS_DIR / \"references_per_scheme_countplot.pdf\"), bbox_inches=\"tight\")\n", + "plt.close(figure)" ] }, { + "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ - "### Number of certificates referencing archived certificates" + "### Number of certificates referencing archived certificates (count plot)" ] }, { @@ -296,6 +296,7 @@ "cert_ids = set(df.loc[((df.cert_id.notnull()) & (df.status == \"archived\")), \"cert_id\"].tolist())\n", "df[\"references_archived_cert\"] = df.directly_referenced_by.map(references_archived_cert)\n", "\n", + "# TODO: We should limit on the number of certificates that referenced an archived certificate at some point where they were active as well.\n", "print(f\"Number of certificates that reference some archived certificate: {df.loc[df.references_archived_cert].shape[0]}\")\n", "\n", "col_to_depict = [\"category\", \"scheme\"]\n", @@ -316,10 +317,11 @@ ] }, { + "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ - "### Count scheme references" + "### Count scheme references (Sankey diagram)" ] }, { @@ -350,22 +352,13 @@ "\n", "sankey(exploded.scheme, exploded.ref_scheme, colorDict=color_dict, leftLabels=list(exploded.scheme.unique()), rightLabels=list(exploded.ref_scheme.unique()), fontsize=7, ax=axes)\n", "\n", - "figure.savefig(\"scheme_references.pdf\", bbox_inches=\"tight\")\n", - "figure.savefig(\"scheme_references.pgf\", bbox_inches=\"tight\")\n", + "figure.savefig(str(RESULTS_DIR / \"scheme_references.pdf\"), bbox_inches=\"tight\")\n", + "figure.savefig(str(RESULTS_DIR / \"scheme_references.pgf\"), bbox_inches=\"tight\")\n", "plt.close(figure)" ] }, { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "print(f\"\\\\newcommand{{\\\\numCCUSReferencing}}{{{len(df_id_rich.loc[(df_id_rich.scheme == 'US') & (df_id_rich.directly_referencing.notnull())])}}}\")\n", - "print(f\"\\\\newcommand{{\\\\numCCUS}}{{{len(df_id_rich.loc[(df_id_rich.scheme == 'US')])}}}\")" - ] - }, - { + "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ @@ -384,6 +377,7 @@ }, "outputs": [], "source": [ + "# TODO: Again, this plot is neither shown nor saved as a figure\n", "df_temporal = df.loc[df.year_from < 2022].groupby([\"year_from\"])[\"directly_referencing\"].count().reset_index().set_index(\"year_from\")\n", "n_issued_certs = df.groupby(\"year_from\").name.count().reset_index().rename(columns={\"name\": \"n_certs\"}).set_index(\"year_from\")\n", "df_temporal.directly_referencing = 100 * df_temporal.directly_referencing / n_issued_certs.n_certs\n", @@ -393,6 +387,7 @@ ] }, { + "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ @@ -400,19 +395,7 @@ ] }, { - "cell_type": "code", - "execution_count": null, - "metadata": { - "pycharm": { - "name": "#%%\n" - } - }, - "outputs": [], - "source": [ - "# Plotting w.r.t. scheme and category (both are interesting)" - ] - }, - { + "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ @@ -436,6 +419,7 @@ ] }, { + "attachments": {}, "cell_type": "markdown", "metadata": { "pycharm": { @@ -468,6 +452,7 @@ ] }, { + "attachments": {}, "cell_type": "markdown", "metadata": { "pycharm": { @@ -500,6 +485,7 @@ ] }, { + "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ @@ -537,6 +523,7 @@ ] }, { + "attachments": {}, "cell_type": "markdown", "metadata": {}, "source": [ @@ -583,10 +570,12 @@ "metadata": {}, "outputs": [], "source": [ + "# TODO: Not sure what this thing does\n", "nx.draw(view, pos=nx.planar_layout(view), with_labels=True)" ] }, { + "attachments": {}, "cell_type": "markdown", "metadata": { "pycharm": { @@ -725,429 +714,41 @@ ] }, { + "attachments": {}, + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## LaTeX commands" + ] + }, + { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ - "import logging\n", - "import warnings\n", - "from collections import defaultdict\n", - "from typing import Any, Dict, List, Optional, Set, Tuple, Union\n", - "\n", - "import matplotlib.pyplot as plt\n", - "import numpy as np\n", - "import pandas as pd\n", - "import seaborn as sns\n", - "from numpy import float64, ndarray\n", - "from pandas.core.frame import DataFrame\n", - "from pandas.core.series import Series\n", - "\n", - "class PySankeyException(Exception):\n", - " \"\"\"Generic PySankey Exception.\"\"\"\n", - "\n", - "\n", - "class NullsInFrame(PySankeyException):\n", - " pass\n", - "\n", - "\n", - "class LabelMismatch(PySankeyException):\n", - " pass\n", - "\n", - "LOGGER = logging.getLogger(__name__)\n", - "\n", - "\n", - "def check_data_matches_labels(\n", - " labels: Union[List[str], Set[str]], data: Series, side: str\n", - ") -> None:\n", - " \"\"\"Check whether data matches labels.\n", - " Raise a LabelMismatch Exception if not.\"\"\"\n", - " if len(labels) > 0:\n", - " if isinstance(data, list):\n", - " data = set(data)\n", - " if isinstance(data, pd.Series):\n", - " data = set(data.unique().tolist())\n", - " if isinstance(labels, list):\n", - " labels = set(labels)\n", - " if labels != data:\n", - " msg = \"\\n\"\n", - " if len(labels) <= 20:\n", - " msg = \"Labels: \" + \",\".join(labels) + \"\\n\"\n", - " if len(data) < 20:\n", - " msg += \"Data: \" + \",\".join(data)\n", - " raise LabelMismatch(f\"{side} labels and data do not match.{msg}\")\n", - "\n", - "\n", - "def sankey(\n", - " left: Union[List, ndarray, Series],\n", - " right: Union[ndarray, Series],\n", - " leftWeight: Optional[ndarray] = None,\n", - " rightWeight: Optional[ndarray] = None,\n", - " colorDict: Optional[Dict[str, str]] = None,\n", - " leftLabels: Optional[List[str]] = None,\n", - " rightLabels: Optional[List[str]] = None,\n", - " aspect: int = 4,\n", - " rightColor: bool = False,\n", - " fontsize: int = 14,\n", - " figureName: Optional[str] = None,\n", - " closePlot: bool = False,\n", - " figSize: Optional[Tuple[int, int]] = None,\n", - " ax: Optional[Any] = None,\n", - ") -> Any:\n", - " \"\"\"\n", - " Make Sankey Diagram showing flow from left-->right\n", - " Inputs:\n", - " left = NumPy array of object labels on the left of the diagram\n", - " right = NumPy array of corresponding labels on the right of the diagram\n", - " len(right) == len(left)\n", - " leftWeight = NumPy array of weights for each strip starting from the\n", - " left of the diagram, if not specified 1 is assigned\n", - " rightWeight = NumPy array of weights for each strip starting from the\n", - " right of the diagram, if not specified the corresponding leftWeight\n", - " is assigned\n", - " colorDict = Dictionary of colors to use for each label\n", - " {'label':'color'}\n", - " leftLabels = order of the left labels in the diagram\n", - " rightLabels = order of the right labels in the diagram\n", - " aspect = vertical extent of the diagram in units of horizontal extent\n", - " rightColor = If true, each strip in the diagram will be be colored\n", - " according to its left label\n", - " figSize = tuple setting the width and height of the sankey diagram.\n", - " Defaults to current figure size\n", - " ax = optional, matplotlib axes to plot on, otherwise uses current axes.\n", - " Output:\n", - " ax : matplotlib Axes\n", - " \"\"\"\n", - " ax, leftLabels, leftWeight, rightLabels, rightWeight = init_values(\n", - " ax,\n", - " closePlot,\n", - " figSize,\n", - " figureName,\n", - " left,\n", - " leftLabels,\n", - " leftWeight,\n", - " rightLabels,\n", - " rightWeight,\n", - " )\n", - " plt.rc(\"text\", usetex=False)\n", - " plt.rc(\"font\", family=\"serif\")\n", - " data_frame = _create_dataframe(left, leftWeight, right, rightWeight)\n", - " # Identify all labels that appear 'left' or 'right'\n", - " all_labels = pd.Series(\n", - " np.r_[data_frame.left.unique(), data_frame.right.unique()]\n", - " ).unique()\n", - " LOGGER.debug(\"Labels to handle : %s\", all_labels)\n", - " leftLabels, rightLabels = identify_labels(data_frame, leftLabels, rightLabels)\n", - " colorDict = create_colors(all_labels, colorDict) # type: ignore\n", - " ns_l, ns_r = determine_widths(data_frame, leftLabels, rightLabels)\n", - " # Determine positions of left label patches and total widths\n", - " leftWidths, topEdge = _get_positions_and_total_widths(\n", - " data_frame, leftLabels, \"left\"\n", - " )\n", - " # Determine positions of right label patches and total widths\n", - " rightWidths, topEdge = _get_positions_and_total_widths(\n", - " data_frame, rightLabels, \"right\"\n", - " )\n", - " # Total vertical extent of diagram\n", - " xMax = topEdge / aspect\n", - " draw_vertical_bars(\n", - " ax,\n", - " colorDict, # type: ignore\n", - " fontsize,\n", - " leftLabels,\n", - " leftWidths,\n", - " rightLabels,\n", - " rightWidths,\n", - " xMax, # type: ignore\n", - " )\n", - " plot_strips(\n", - " ax,\n", - " colorDict, # type: ignore\n", - " data_frame,\n", - " leftLabels,\n", - " leftWidths,\n", - " ns_l,\n", - " ns_r,\n", - " rightColor,\n", - " rightLabels,\n", - " rightWidths,\n", - " xMax,\n", - " )\n", - " if figSize is not None:\n", - " plt.gcf().set_size_inches(figSize)\n", - " save_image(figureName)\n", - " if closePlot:\n", - " plt.close()\n", - " return ax\n", - "\n", - "\n", - "def save_image(figureName: Optional[str]) -> None:\n", - " if figureName is not None:\n", - " file_name = f\"{figureName}.png\"\n", - " plt.savefig(file_name, bbox_inches=\"tight\", dpi=150)\n", - " LOGGER.info(\"Sankey diagram generated in '%s'\", file_name)\n", - "\n", - "\n", - "def identify_labels(\n", - " dataFrame: DataFrame, leftLabels: List[str], rightLabels: List[str]\n", - ") -> Tuple[ndarray, ndarray]:\n", - " # Identify left labels\n", - " if len(leftLabels) == 0:\n", - " leftLabels = pd.Series(dataFrame.left.unique()).unique()\n", - " else:\n", - " check_data_matches_labels(leftLabels, dataFrame[\"left\"], \"left\")\n", - " # Identify right labels\n", - " if len(rightLabels) == 0:\n", - " rightLabels = pd.Series(dataFrame.right.unique()).unique()\n", - " else:\n", - " check_data_matches_labels(rightLabels, dataFrame[\"right\"], \"right\")\n", - " return leftLabels, rightLabels\n", - "\n", - "\n", - "def init_values(\n", - " ax: Optional[Any],\n", - " closePlot: bool,\n", - " figSize: Optional[Tuple[int, int]],\n", - " figureName: Optional[str],\n", - " left: Union[List, ndarray, Series],\n", - " leftLabels: Optional[List[str]],\n", - " leftWeight: Optional[ndarray],\n", - " rightLabels: Optional[List[str]],\n", - " rightWeight: Optional[ndarray],\n", - ") -> Tuple[Any, List[str], ndarray, List[str], ndarray]:\n", - " deprecation_warnings(closePlot, figSize, figureName)\n", - " if ax is None:\n", - " ax = plt.gca()\n", - " if leftWeight is None:\n", - " leftWeight = []\n", - " if rightWeight is None:\n", - " rightWeight = []\n", - " if leftLabels is None:\n", - " leftLabels = []\n", - " if rightLabels is None:\n", - " rightLabels = []\n", - " # Check weights\n", - " if len(leftWeight) == 0:\n", - " leftWeight = np.ones(len(left))\n", - " if len(rightWeight) == 0:\n", - " rightWeight = leftWeight\n", - " return ax, leftLabels, leftWeight, rightLabels, rightWeight\n", - "\n", - "\n", - "def deprecation_warnings(\n", - " closePlot: bool, figSize: Optional[Tuple[int, int]], figureName: Optional[str]\n", - ") -> None:\n", - " warn = []\n", - " if figureName is not None:\n", - " msg = \"use of figureName in sankey() is deprecated\"\n", - " warnings.warn(msg, DeprecationWarning)\n", - " warn.append(msg[7:-14])\n", - " if closePlot is not False:\n", - " msg = \"use of closePlot in sankey() is deprecated\"\n", - " warnings.warn(msg, DeprecationWarning)\n", - " warn.append(msg[7:-14])\n", - " if figSize is not None:\n", - " msg = \"use of figSize in sankey() is deprecated\"\n", - " warnings.warn(msg, DeprecationWarning)\n", - " warn.append(msg[7:-14])\n", - " if warn:\n", - " LOGGER.warning(\n", - " \" The following arguments are deprecated and should be removed: %s\",\n", - " \", \".join(warn),\n", - " )\n", - "\n", - "\n", - "def determine_widths(\n", - " dataFrame: DataFrame, leftLabels: ndarray, rightLabels: ndarray\n", - ") -> Tuple[Dict, Dict]:\n", - " # Determine widths of individual strips\n", - " ns_l: Dict = defaultdict()\n", - " ns_r: Dict = defaultdict()\n", - " for leftLabel in leftLabels:\n", - " left_dict = {}\n", - " right_dict = {}\n", - " for rightLabel in rightLabels:\n", - " left_dict[rightLabel] = dataFrame[\n", - " (dataFrame.left == leftLabel) & (dataFrame.right == rightLabel)\n", - " ].leftWeight.sum()\n", - " right_dict[rightLabel] = dataFrame[\n", - " (dataFrame.left == leftLabel) & (dataFrame.right == rightLabel)\n", - " ].rightWeight.sum()\n", - " ns_l[leftLabel] = left_dict\n", - " ns_r[leftLabel] = right_dict\n", - " return ns_l, ns_r\n", - "\n", - "\n", - "def draw_vertical_bars(\n", - " ax: Any,\n", - " colorDict: Union[Dict[str, Tuple[float, float, float]], Dict[str, str]],\n", - " fontsize: int,\n", - " leftLabels: ndarray,\n", - " leftWidths: Dict,\n", - " rightLabels: ndarray,\n", - " rightWidths: Dict,\n", - " xMax: float64,\n", - ") -> None:\n", - " # Draw vertical bars on left and right of each label's section & print label\n", - " for leftLabel in leftLabels:\n", - " ax.fill_between(\n", - " [-0.02 * xMax, 0],\n", - " 2 * [leftWidths[leftLabel][\"bottom\"]],\n", - " 2 * [leftWidths[leftLabel][\"bottom\"] + leftWidths[leftLabel][\"left\"]],\n", - " color=colorDict[leftLabel],\n", - " alpha=0.99,\n", - " )\n", - " ax.text(\n", - " -0.05 * xMax,\n", - " leftWidths[leftLabel][\"bottom\"] + 0.5 * leftWidths[leftLabel][\"left\"],\n", - " leftLabel,\n", - " {\"ha\": \"right\", \"va\": \"center\"},\n", - " fontsize=fontsize,\n", - " )\n", - " for rightLabel in rightLabels:\n", - " ax.fill_between(\n", - " [xMax, 1.02 * xMax],\n", - " 2 * [rightWidths[rightLabel][\"bottom\"]],\n", - " 2 * [rightWidths[rightLabel][\"bottom\"] + rightWidths[rightLabel][\"right\"]],\n", - " color=colorDict[rightLabel],\n", - " alpha=0.99,\n", - " )\n", - " ax.text(\n", - " 1.05 * xMax,\n", - " rightWidths[rightLabel][\"bottom\"] + 0.5 * rightWidths[rightLabel][\"right\"],\n", - " rightLabel,\n", - " {\"ha\": \"left\", \"va\": \"center\"},\n", - " fontsize=fontsize,\n", - " )\n", - "\n", - "\n", - "def create_colors(\n", - " allLabels: ndarray, colorDict: Optional[Dict[str, str]]\n", - ") -> Union[Dict[str, Tuple[float, float, float]], Dict[str, str]]:\n", - " # If no colorDict given, make one\n", - " if colorDict is None:\n", - " colorDict = {}\n", - " palette = \"hls\"\n", - " colorPalette = sns.color_palette(palette, len(allLabels))\n", - " for i, label in enumerate(allLabels):\n", - " colorDict[label] = colorPalette[i]\n", - " else:\n", - " missing = [label for label in allLabels if label not in colorDict.keys()]\n", - " if missing:\n", - " raise ValueError(\n", - " \"The colorDict parameter is missing values for the following labels : \"\n", - " + \", \".join(missing)\n", - " )\n", - " LOGGER.debug(\"The colordict value are : %s\", colorDict)\n", - " return colorDict\n", - "\n", - "\n", - "def _create_dataframe(\n", - " left: Union[List, ndarray, Series],\n", - " leftWeight: Union[ndarray, Series],\n", - " right: Union[ndarray, Series],\n", - " rightWeight: Union[ndarray, Series],\n", - ") -> DataFrame:\n", - " # Create Dataframe\n", - " if isinstance(left, pd.Series):\n", - " left = left.reset_index(drop=True)\n", - " if isinstance(right, pd.Series):\n", - " right = right.reset_index(drop=True)\n", - " if isinstance(leftWeight, pd.Series):\n", - " leftWeight = leftWeight.reset_index(drop=True)\n", - " if isinstance(rightWeight, pd.Series):\n", - " rightWeight = rightWeight.reset_index(drop=True)\n", - " data_frame = pd.DataFrame(\n", - " {\n", - " \"left\": left,\n", - " \"right\": right,\n", - " \"leftWeight\": leftWeight,\n", - " \"rightWeight\": rightWeight,\n", - " },\n", - " index=range(len(left)),\n", - " )\n", - " if len(data_frame[(data_frame.left.isnull()) | (data_frame.right.isnull())]):\n", - " raise NullsInFrame(\"Sankey graph does not support null values.\")\n", - " return data_frame\n", - "\n", + "def contains_archived_cert_reference(referencing):\n", + " if pd.isnull(referencing):\n", + " return False\n", + " return bool(archived_cert_id_list.intersection(referencing))\n", "\n", - "def plot_strips(\n", - " ax: Any,\n", - " colorDict: Union[Dict[str, Tuple[float, float, float]], Dict[str, str]],\n", - " dataFrame: DataFrame,\n", - " leftLabels: ndarray,\n", - " leftWidths: Dict,\n", - " ns_l: Dict,\n", - " ns_r: Dict,\n", - " rightColor: bool,\n", - " rightLabels: ndarray,\n", - " rightWidths: Dict,\n", - " xMax: float64,\n", - ") -> None:\n", - " # Plot strips\n", - " for leftLabel in leftLabels:\n", - " for rightLabel in rightLabels:\n", - " label_color = leftLabel\n", - " if rightColor:\n", - " label_color = rightLabel\n", - " if (\n", - " len(\n", - " dataFrame[\n", - " (dataFrame.left == leftLabel) & (dataFrame.right == rightLabel)\n", - " ]\n", - " )\n", - " > 0\n", - " ):\n", - " # Create array of y values for each strip, half at left value,\n", - " # half at right, convolve\n", - " ys_d = np.array(\n", - " 50 * [leftWidths[leftLabel][\"bottom\"]]\n", - " + 50 * [rightWidths[rightLabel][\"bottom\"]]\n", - " )\n", - " ys_d = np.convolve(ys_d, 0.05 * np.ones(20), mode=\"valid\")\n", - " ys_d = np.convolve(ys_d, 0.05 * np.ones(20), mode=\"valid\")\n", - " ys_u = np.array(\n", - " 50 * [leftWidths[leftLabel][\"bottom\"] + ns_l[leftLabel][rightLabel]]\n", - " + 50\n", - " * [rightWidths[rightLabel][\"bottom\"] + ns_r[leftLabel][rightLabel]]\n", - " )\n", - " ys_u = np.convolve(ys_u, 0.05 * np.ones(20), mode=\"valid\")\n", - " ys_u = np.convolve(ys_u, 0.05 * np.ones(20), mode=\"valid\")\n", "\n", - " # Update bottom edges at each label so next strip starts at the\n", - " # right place\n", - " leftWidths[leftLabel][\"bottom\"] += ns_l[leftLabel][rightLabel]\n", - " rightWidths[rightLabel][\"bottom\"] += ns_r[leftLabel][rightLabel]\n", - " ax.fill_between(\n", - " np.linspace(0, xMax, len(ys_d)),\n", - " ys_d,\n", - " ys_u,\n", - " alpha=0.65,\n", - " color=colorDict[label_color],\n", - " )\n", - " ax.axis(\"off\")\n", + "print(f\"\\\\newcommand{{\\\\numCcAllDirectReferencing}}{{{df.has_outgoing_direct_references.sum()}}}\")\n", + "print(f\"\\\\newcommand{{\\\\numCcAllNotDirectReferencing}}{{{len(df) - df.has_outgoing_direct_references.sum()}}}\")\n", + "print(f\"\\\\newcommand{{\\\\numCcWithIdDirectReferencing}}{{{df_id_rich.has_outgoing_direct_references.sum()}}}\")\n", + "print(f\"\\\\newcommand{{\\\\numCcWithIdNotDirectReferencing}}{{{len(df_id_rich) - df_id_rich.has_outgoing_direct_references.sum()}}}\")\n", + "print(f\"\\\\newcommand{{\\\\numCCActiveDirectReferencing}}{{{df_id_rich.loc[df_id_rich.status == 'active'].has_outgoing_direct_references.sum()}}}\")\n", + "print(f\"\\\\newcommand{{\\\\numCCActiveDirectReferencingArchived}}{{{df_id_rich[df_id_rich.status == 'active'].directly_referencing.apply(contains_archived_cert_reference).sum()}}}\")\n", "\n", + "print(\"\")\n", + "print(f\"\\\\newcommand{{\\\\numCCDirectRefsSameCategory}}{{{(exploded_with_refs.category == exploded_with_refs.ref_category).sum()}}}\")\n", + "print(f\"\\\\newcommand{{\\\\numCCDirectRefsOtherCategory}}{{{(exploded_with_refs.category != exploded_with_refs.ref_category).sum()}}}\")\n", + "print(f\"\\\\newcommand{{\\\\numCCDirectRefs}}{{{len(exploded_with_refs)}}}\")\n", + "print(f\"\\\\newcommand{{\\\\numCCDirectRefsFromSmartcards}}{{{(exploded_with_refs.category == 'ICs, Smart Cards and Smart Card-Related Devices and Systems').sum()}}}\")\n", "\n", - "def _get_positions_and_total_widths(\n", - " df: DataFrame, labels: ndarray, side: str\n", - ") -> Tuple[Dict, float64]:\n", - " \"\"\"Determine positions of label patches and total widths\"\"\"\n", - " widths: Dict = defaultdict()\n", - " for i, label in enumerate(labels):\n", - " label_widths = {}\n", - " label_widths[side] = df[df[side] == label][side + \"Weight\"].sum()\n", - " if i == 0:\n", - " label_widths[\"bottom\"] = 0\n", - " label_widths[\"top\"] = label_widths[side]\n", - " else:\n", - " bottom_width = widths[labels[i - 1]][\"top\"]\n", - " weighted_sum = 0.05 * df[side + \"Weight\"].sum()\n", - " label_widths[\"bottom\"] = bottom_width + weighted_sum\n", - " label_widths[\"top\"] = label_widths[\"bottom\"] + label_widths[side]\n", - " topEdge = label_widths[\"top\"]\n", - " widths[label] = label_widths\n", - " LOGGER.debug(\"%s position of '%s' : %s\", side, label, label_widths)\n", - " return widths, topEdge\n" + "print(\"\")\n", + "print(f\"\\\\newcommand{{\\\\numCCUSReferencing}}{{{len(df_id_rich.loc[(df_id_rich.scheme == 'US') & (df_id_rich.directly_referencing.notnull())])}}}\")\n", + "print(f\"\\\\newcommand{{\\\\numCCUS}}{{{len(df_id_rich.loc[(df_id_rich.scheme == 'US')])}}}\")" ] } ], @@ -1167,7 +768,7 @@ "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", - "version": "3.8.13" + "version": "3.8.16" }, "vscode": { "interpreter": { diff --git a/notebooks/fixed_sankey_plot.py b/notebooks/fixed_sankey_plot.py new file mode 100644 index 00000000..cd7dc894 --- /dev/null +++ b/notebooks/fixed_sankey_plot.py @@ -0,0 +1,400 @@ +# type: ignore + +""" +This is a fork of https://github.com/anazalea/pySankey/blob/master/pysankey/sankey.py. +We've had some problems with the plot, mostly related to resizing (likely, I don't remember now). +This code should fix the problems and should be used to produce figures in the relevant sec-certs papers. +""" + +import logging +import warnings +from collections import defaultdict +from typing import Any, Dict, List, Optional, Set, Tuple, Union + +import matplotlib.pyplot as plt +import numpy as np +import pandas as pd +import seaborn as sns +from numpy import float64, ndarray +from pandas.core.frame import DataFrame +from pandas.core.series import Series + + +class PySankeyException(Exception): + """Generic PySankey Exception.""" + + +class NullsInFrame(PySankeyException): + pass + + +class LabelMismatch(PySankeyException): + pass + + +LOGGER = logging.getLogger(__name__) + + +def check_data_matches_labels(labels: Union[List[str], Set[str]], data: Series, side: str) -> None: + """Check whether data matches labels. + Raise a LabelMismatch Exception if not.""" + if len(labels) > 0: + if isinstance(data, list): + data = set(data) + if isinstance(data, pd.Series): + data = set(data.unique().tolist()) + if isinstance(labels, list): + labels = set(labels) + if labels != data: + msg = "\n" + if len(labels) <= 20: + msg = "Labels: " + ",".join(labels) + "\n" + if len(data) < 20: + msg += "Data: " + ",".join(data) + raise LabelMismatch(f"{side} labels and data do not match.{msg}") + + +def sankey( + left: Union[List, ndarray, Series], + right: Union[ndarray, Series], + leftWeight: Optional[ndarray] = None, + rightWeight: Optional[ndarray] = None, + colorDict: Optional[Dict[str, str]] = None, + leftLabels: Optional[List[str]] = None, + rightLabels: Optional[List[str]] = None, + aspect: int = 4, + rightColor: bool = False, + fontsize: int = 14, + figureName: Optional[str] = None, + closePlot: bool = False, + figSize: Optional[Tuple[int, int]] = None, + ax: Optional[Any] = None, +) -> Any: + """ + Make Sankey Diagram showing flow from left-->right + Inputs: + left = NumPy array of object labels on the left of the diagram + right = NumPy array of corresponding labels on the right of the diagram + len(right) == len(left) + leftWeight = NumPy array of weights for each strip starting from the + left of the diagram, if not specified 1 is assigned + rightWeight = NumPy array of weights for each strip starting from the + right of the diagram, if not specified the corresponding leftWeight + is assigned + colorDict = Dictionary of colors to use for each label + {'label':'color'} + leftLabels = order of the left labels in the diagram + rightLabels = order of the right labels in the diagram + aspect = vertical extent of the diagram in units of horizontal extent + rightColor = If true, each strip in the diagram will be be colored + according to its left label + figSize = tuple setting the width and height of the sankey diagram. + Defaults to current figure size + ax = optional, matplotlib axes to plot on, otherwise uses current axes. + Output: + ax : matplotlib Axes + """ + ax, leftLabels, leftWeight, rightLabels, rightWeight = init_values( + ax, + closePlot, + figSize, + figureName, + left, + leftLabels, + leftWeight, + rightLabels, + rightWeight, + ) + plt.rc("text", usetex=False) + plt.rc("font", family="serif") + data_frame = _create_dataframe(left, leftWeight, right, rightWeight) + # Identify all labels that appear 'left' or 'right' + all_labels = pd.Series(np.r_[data_frame.left.unique(), data_frame.right.unique()]).unique() + LOGGER.debug("Labels to handle : %s", all_labels) + leftLabels, rightLabels = identify_labels(data_frame, leftLabels, rightLabels) + colorDict = create_colors(all_labels, colorDict) # type: ignore + ns_l, ns_r = determine_widths(data_frame, leftLabels, rightLabels) + # Determine positions of left label patches and total widths + leftWidths, topEdge = _get_positions_and_total_widths(data_frame, leftLabels, "left") + # Determine positions of right label patches and total widths + rightWidths, topEdge = _get_positions_and_total_widths(data_frame, rightLabels, "right") + # Total vertical extent of diagram + xMax = topEdge / aspect + draw_vertical_bars( + ax, + colorDict, # type: ignore + fontsize, + leftLabels, + leftWidths, + rightLabels, + rightWidths, + xMax, # type: ignore + ) + plot_strips( + ax, + colorDict, # type: ignore + data_frame, + leftLabels, + leftWidths, + ns_l, + ns_r, + rightColor, + rightLabels, + rightWidths, + xMax, + ) + if figSize is not None: + plt.gcf().set_size_inches(figSize) + save_image(figureName) + if closePlot: + plt.close() + return ax + + +def save_image(figureName: Optional[str]) -> None: + if figureName is not None: + file_name = f"{figureName}.png" + plt.savefig(file_name, bbox_inches="tight", dpi=150) + LOGGER.info("Sankey diagram generated in '%s'", file_name) + + +def identify_labels(dataFrame: DataFrame, leftLabels: List[str], rightLabels: List[str]) -> Tuple[ndarray, ndarray]: + # Identify left labels + if len(leftLabels) == 0: + leftLabels = pd.Series(dataFrame.left.unique()).unique() + else: + check_data_matches_labels(leftLabels, dataFrame["left"], "left") + # Identify right labels + if len(rightLabels) == 0: + rightLabels = pd.Series(dataFrame.right.unique()).unique() + else: + check_data_matches_labels(rightLabels, dataFrame["right"], "right") + return leftLabels, rightLabels + + +def init_values( + ax: Optional[Any], + closePlot: bool, + figSize: Optional[Tuple[int, int]], + figureName: Optional[str], + left: Union[List, ndarray, Series], + leftLabels: Optional[List[str]], + leftWeight: Optional[ndarray], + rightLabels: Optional[List[str]], + rightWeight: Optional[ndarray], +) -> Tuple[Any, List[str], ndarray, List[str], ndarray]: + deprecation_warnings(closePlot, figSize, figureName) + if ax is None: + ax = plt.gca() + if leftWeight is None: + leftWeight = [] + if rightWeight is None: + rightWeight = [] + if leftLabels is None: + leftLabels = [] + if rightLabels is None: + rightLabels = [] + # Check weights + if len(leftWeight) == 0: + leftWeight = np.ones(len(left)) + if len(rightWeight) == 0: + rightWeight = leftWeight + return ax, leftLabels, leftWeight, rightLabels, rightWeight + + +def deprecation_warnings(closePlot: bool, figSize: Optional[Tuple[int, int]], figureName: Optional[str]) -> None: + warn = [] + if figureName is not None: + msg = "use of figureName in sankey() is deprecated" + warnings.warn(msg, DeprecationWarning) + warn.append(msg[7:-14]) + if closePlot is not False: + msg = "use of closePlot in sankey() is deprecated" + warnings.warn(msg, DeprecationWarning) + warn.append(msg[7:-14]) + if figSize is not None: + msg = "use of figSize in sankey() is deprecated" + warnings.warn(msg, DeprecationWarning) + warn.append(msg[7:-14]) + if warn: + LOGGER.warning( + " The following arguments are deprecated and should be removed: %s", + ", ".join(warn), + ) + + +def determine_widths(dataFrame: DataFrame, leftLabels: ndarray, rightLabels: ndarray) -> Tuple[Dict, Dict]: + # Determine widths of individual strips + ns_l: Dict = defaultdict() + ns_r: Dict = defaultdict() + for leftLabel in leftLabels: + left_dict = {} + right_dict = {} + for rightLabel in rightLabels: + left_dict[rightLabel] = dataFrame[ + (dataFrame.left == leftLabel) & (dataFrame.right == rightLabel) + ].leftWeight.sum() + right_dict[rightLabel] = dataFrame[ + (dataFrame.left == leftLabel) & (dataFrame.right == rightLabel) + ].rightWeight.sum() + ns_l[leftLabel] = left_dict + ns_r[leftLabel] = right_dict + return ns_l, ns_r + + +def draw_vertical_bars( + ax: Any, + colorDict: Union[Dict[str, Tuple[float, float, float]], Dict[str, str]], + fontsize: int, + leftLabels: ndarray, + leftWidths: Dict, + rightLabels: ndarray, + rightWidths: Dict, + xMax: float64, +) -> None: + # Draw vertical bars on left and right of each label's section & print label + for leftLabel in leftLabels: + ax.fill_between( + [-0.02 * xMax, 0], + 2 * [leftWidths[leftLabel]["bottom"]], + 2 * [leftWidths[leftLabel]["bottom"] + leftWidths[leftLabel]["left"]], + color=colorDict[leftLabel], + alpha=0.99, + ) + ax.text( + -0.05 * xMax, + leftWidths[leftLabel]["bottom"] + 0.5 * leftWidths[leftLabel]["left"], + leftLabel, + {"ha": "right", "va": "center"}, + fontsize=fontsize, + ) + for rightLabel in rightLabels: + ax.fill_between( + [xMax, 1.02 * xMax], + 2 * [rightWidths[rightLabel]["bottom"]], + 2 * [rightWidths[rightLabel]["bottom"] + rightWidths[rightLabel]["right"]], + color=colorDict[rightLabel], + alpha=0.99, + ) + ax.text( + 1.05 * xMax, + rightWidths[rightLabel]["bottom"] + 0.5 * rightWidths[rightLabel]["right"], + rightLabel, + {"ha": "left", "va": "center"}, + fontsize=fontsize, + ) + + +def create_colors( + allLabels: ndarray, colorDict: Optional[Dict[str, str]] +) -> Union[Dict[str, Tuple[float, float, float]], Dict[str, str]]: + # If no colorDict given, make one + if colorDict is None: + colorDict = {} + palette = "hls" + colorPalette = sns.color_palette(palette, len(allLabels)) + for i, label in enumerate(allLabels): + colorDict[label] = colorPalette[i] + else: + missing = [label for label in allLabels if label not in colorDict.keys()] + if missing: + raise ValueError( + "The colorDict parameter is missing values for the following labels : " + ", ".join(missing) + ) + LOGGER.debug("The colordict value are : %s", colorDict) + return colorDict + + +def _create_dataframe( + left: Union[List, ndarray, Series], + leftWeight: Union[ndarray, Series], + right: Union[ndarray, Series], + rightWeight: Union[ndarray, Series], +) -> DataFrame: + # Create Dataframe + if isinstance(left, pd.Series): + left = left.reset_index(drop=True) + if isinstance(right, pd.Series): + right = right.reset_index(drop=True) + if isinstance(leftWeight, pd.Series): + leftWeight = leftWeight.reset_index(drop=True) + if isinstance(rightWeight, pd.Series): + rightWeight = rightWeight.reset_index(drop=True) + data_frame = pd.DataFrame( + { + "left": left, + "right": right, + "leftWeight": leftWeight, + "rightWeight": rightWeight, + }, + index=range(len(left)), + ) + if len(data_frame[(data_frame.left.isnull()) | (data_frame.right.isnull())]): + raise NullsInFrame("Sankey graph does not support null values.") + return data_frame + + +def plot_strips( + ax: Any, + colorDict: Union[Dict[str, Tuple[float, float, float]], Dict[str, str]], + dataFrame: DataFrame, + leftLabels: ndarray, + leftWidths: Dict, + ns_l: Dict, + ns_r: Dict, + rightColor: bool, + rightLabels: ndarray, + rightWidths: Dict, + xMax: float64, +) -> None: + # Plot strips + for leftLabel in leftLabels: + for rightLabel in rightLabels: + label_color = leftLabel + if rightColor: + label_color = rightLabel + if len(dataFrame[(dataFrame.left == leftLabel) & (dataFrame.right == rightLabel)]) > 0: + # Create array of y values for each strip, half at left value, + # half at right, convolve + ys_d = np.array(50 * [leftWidths[leftLabel]["bottom"]] + 50 * [rightWidths[rightLabel]["bottom"]]) + ys_d = np.convolve(ys_d, 0.05 * np.ones(20), mode="valid") + ys_d = np.convolve(ys_d, 0.05 * np.ones(20), mode="valid") + ys_u = np.array( + 50 * [leftWidths[leftLabel]["bottom"] + ns_l[leftLabel][rightLabel]] + + 50 * [rightWidths[rightLabel]["bottom"] + ns_r[leftLabel][rightLabel]] + ) + ys_u = np.convolve(ys_u, 0.05 * np.ones(20), mode="valid") + ys_u = np.convolve(ys_u, 0.05 * np.ones(20), mode="valid") + + # Update bottom edges at each label so next strip starts at the + # right place + leftWidths[leftLabel]["bottom"] += ns_l[leftLabel][rightLabel] + rightWidths[rightLabel]["bottom"] += ns_r[leftLabel][rightLabel] + ax.fill_between( + np.linspace(0, xMax, len(ys_d)), + ys_d, + ys_u, + alpha=0.65, + color=colorDict[label_color], + ) + ax.axis("off") + + +def _get_positions_and_total_widths(df: DataFrame, labels: ndarray, side: str) -> Tuple[Dict, float64]: + """Determine positions of label patches and total widths""" + widths: Dict = defaultdict() + for i, label in enumerate(labels): + label_widths = {} + label_widths[side] = df[df[side] == label][side + "Weight"].sum() + if i == 0: + label_widths["bottom"] = 0 + label_widths["top"] = label_widths[side] + else: + bottom_width = widths[labels[i - 1]]["top"] + weighted_sum = 0.05 * df[side + "Weight"].sum() + label_widths["bottom"] = bottom_width + weighted_sum + label_widths["top"] = label_widths["bottom"] + label_widths[side] + topEdge = label_widths["top"] + widths[label] = label_widths + LOGGER.debug("%s position of '%s' : %s", side, label, label_widths) + return widths, topEdge |
