Logs_Simulator.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "provenance": []
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "source": [
        "Instructions for use: From the Github, click \"Open in Colab\". You may need to log into Google's Colab, a Jupyter Python notebook service with free access. Hit File > Save a Copy in Drive (you only have to do this once).\n",
        "\n",
        "You will also need to import the functions in `CoreSimulator.ipynb` to run this\n",
        "notebook.\n",
        "\n",
        "1.   Go to [the core simulator notebook](https://github.com/Amarantine-xiv/Amas-FF14-Combat-Sim/blob/main/CoreSimulator.ipynb) and open in Colab\n",
        "2.   File -> Download -> as .py\n",
        "3.   Add the file into your google drive, ideally in the same folder as this copy of `Logs_Simulator.ipynb`\n",
        "4.   When running `Setup`, you will be asked to allow the sheet to access files to read the python. Check the boxes.\n",
        "\n",
        "To get fflogs data, you will need `FFLOGS_CLIENT_ID` and `FFLOGS_CLIENT_SECRET` set as secrets. These values can be obtained here: https://www.fflogs.com/api/clients/\n",
        "\n",
        "To set secrets, click the Key icon to the left.\n",
        "\n"
      ],
      "metadata": {
        "id": "Kltlu22SKVxG"
      }
    },
    {
      "cell_type": "code",
      "execution_count": 19,
      "metadata": {
        "id": "DPMDrVTu21FT",
        "cellView": "form",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "a26a7d52-3cdf-4c94-a53b-f3d37ceafd0f"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Drive already mounted at /content/gdrive/; to attempt to forcibly remount, call drive.mount(\"/content/gdrive/\", force_remount=True).\n"
          ]
        }
      ],
      "source": [
        "#@title Setup\n",
        "#\n",
        "from google.colab import drive\n",
        "\n",
        "# This will ask for permissions, grant them.\n",
        "drive.mount('/content/gdrive/')\n",
        "\n",
        "import sys\n",
        "import os\n",
        "# If this path is incorrect, modify to whatever folder contains coresimulator.py\n",
        "sys.path.append('/content/gdrive/My Drive/Colab Notebooks')\n",
        "\n",
        "import coresimulator as cr\n",
        "\n",
        "from google.colab import userdata\n",
        "# Ensure FFLOGS values are set when using this. Please set\n",
        "# as secrets and do not use them in plaintext.\n",
        "if not (userdata.get(\"FFLOGS_CLIENT_ID\") or userdata.get(\"FFLOGS_CLIENT_SECRET\")):\n",
        "  raise Exception(f\"Please set FFLOGs tokens to access logs. https://www.fflogs.com/api/clients/\")\n",
        "\n",
        "# TODO: move a lot of the implementation into python files on git\n",
        "# So it can also just be added onto drive"
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "#@title Imports and logging setup\n",
        "import numpy as np\n",
        "import matplotlib.pyplot as plt\n",
        "import pandas as pd\n",
        "import dataclasses\n",
        "import functools\n",
        "import requests\n",
        "import pprint\n",
        "\n",
        "pd.options.mode.copy_on_write = True\n",
        "\n",
        "from typing import *\n",
        "\n",
        "import logging\n",
        "logger = logging.getLogger()\n",
        "# Modify this value to set logging level.\n",
        "logging_level = logging.INFO\n",
        "logger.setLevel(logging_level)\n",
        "logger.log(logging_level, f\"Set logging level to {logging_level}\")\n",
        "\n",
        "LOG = logging.getLogger(\"logs_simulator\")\n",
        "\n",
        "# Pip req imports\n",
        "!pip install coreapi==2.3.3 python-graphql-client==0.4.3\n",
        "import coreapi\n",
        "import python_graphql_client\n"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "cellView": "form",
        "id": "zP6e_MGYNOhM",
        "outputId": "f2b5ef2a-3b6b-4b52-8eed-0144e2af3664"
      },
      "execution_count": 20,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stderr",
          "text": [
            "INFO:root:Set logging level to 20\n"
          ]
        },
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Requirement already satisfied: coreapi==2.3.3 in /usr/local/lib/python3.10/dist-packages (2.3.3)\n",
            "Requirement already satisfied: python-graphql-client==0.4.3 in /usr/local/lib/python3.10/dist-packages (0.4.3)\n",
            "Requirement already satisfied: coreschema in /usr/local/lib/python3.10/dist-packages (from coreapi==2.3.3) (0.0.4)\n",
            "Requirement already satisfied: requests in /usr/local/lib/python3.10/dist-packages (from coreapi==2.3.3) (2.31.0)\n",
            "Requirement already satisfied: itypes in /usr/local/lib/python3.10/dist-packages (from coreapi==2.3.3) (1.2.0)\n",
            "Requirement already satisfied: uritemplate in /usr/local/lib/python3.10/dist-packages (from coreapi==2.3.3) (4.1.1)\n",
            "Requirement already satisfied: aiohttp~=3.0 in /usr/local/lib/python3.10/dist-packages (from python-graphql-client==0.4.3) (3.9.3)\n",
            "Requirement already satisfied: websockets>=5.0 in /usr/local/lib/python3.10/dist-packages (from python-graphql-client==0.4.3) (12.0)\n",
            "Requirement already satisfied: aiosignal>=1.1.2 in /usr/local/lib/python3.10/dist-packages (from aiohttp~=3.0->python-graphql-client==0.4.3) (1.3.1)\n",
            "Requirement already satisfied: attrs>=17.3.0 in /usr/local/lib/python3.10/dist-packages (from aiohttp~=3.0->python-graphql-client==0.4.3) (23.2.0)\n",
            "Requirement already satisfied: frozenlist>=1.1.1 in /usr/local/lib/python3.10/dist-packages (from aiohttp~=3.0->python-graphql-client==0.4.3) (1.4.1)\n",
            "Requirement already satisfied: multidict<7.0,>=4.5 in /usr/local/lib/python3.10/dist-packages (from aiohttp~=3.0->python-graphql-client==0.4.3) (6.0.5)\n",
            "Requirement already satisfied: yarl<2.0,>=1.0 in /usr/local/lib/python3.10/dist-packages (from aiohttp~=3.0->python-graphql-client==0.4.3) (1.9.4)\n",
            "Requirement already satisfied: async-timeout<5.0,>=4.0 in /usr/local/lib/python3.10/dist-packages (from aiohttp~=3.0->python-graphql-client==0.4.3) (4.0.3)\n",
            "Requirement already satisfied: charset-normalizer<4,>=2 in /usr/local/lib/python3.10/dist-packages (from requests->coreapi==2.3.3) (3.3.2)\n",
            "Requirement already satisfied: idna<4,>=2.5 in /usr/local/lib/python3.10/dist-packages (from requests->coreapi==2.3.3) (3.6)\n",
            "Requirement already satisfied: urllib3<3,>=1.21.1 in /usr/local/lib/python3.10/dist-packages (from requests->coreapi==2.3.3) (2.0.7)\n",
            "Requirement already satisfied: certifi>=2017.4.17 in /usr/local/lib/python3.10/dist-packages (from requests->coreapi==2.3.3) (2024.2.2)\n",
            "Requirement already satisfied: jinja2 in /usr/local/lib/python3.10/dist-packages (from coreschema->coreapi==2.3.3) (3.1.3)\n",
            "Requirement already satisfied: MarkupSafe>=2.0 in /usr/local/lib/python3.10/dist-packages (from jinja2->coreschema->coreapi==2.3.3) (2.1.5)\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "#@title Custom FFLOGs constants\n",
        "\"\"\"Defines and documents all custom columns in the data.\"\"\"\n",
        "\n",
        "# Not custom, but used a lot so recorded\n",
        "ABILITY_ID = \"abilityGameID\"\n",
        "TYPE = \"type\"\n",
        "CAST = \"cast\"\n",
        "DAMAGE = \"damage\"\n",
        "APPLY_BUFF = \"applybuff\"\n",
        "APPLY_DEBUFF = \"applydebuff\"\n",
        "REFRESH_BUFF = \"refreshbuff\"\n",
        "REMOVE_BUFF = \"removebuff\"\n",
        "REMOVE_DEBUFF = \"removedebuff\"\n",
        "PREPARE = \"calculateddamage\"\n",
        "\n",
        "\n",
        "# Human readable name for abilities\n",
        "ABILITY_NAME = \"ability_name\"\n",
        "\n",
        "\n",
        "# Human readable source name for actors\n",
        "SOURCE_NAME = \"source_name\"\n",
        "SOURCE_TYPE = \"source_type\"\n",
        "SOURCE_SUBTYPE = \"source_subtype\"\n",
        "SOURCE_PET_ACTOR = \"source_pet_actor\"\n",
        "SOURCE_ID = \"sourceID\"\n",
        "\n",
        "# Human readable target name for actors\n",
        "TARGET_NAME = \"target_name\"\n",
        "TARGET_TYPE = \"target_type\"\n",
        "TARGET_SUBTYPE = \"target_subtype\"\n",
        "TARGET_PET_ACTOR = \"target_pet_actor\"\n",
        "TARGET_ID = \"targetID\"\n",
        "\n",
        "# Fight time, based on lb update tick to reset gauge\n",
        "FIGHT_TIME = \"fight_time\"\n",
        "\n",
        "# Bonus percent, estimate the value of combos\n",
        "# 1 - basePotency/bonusPotency = bonusPercent\n",
        "BONUS_PERCENT = \"bonusPercent\"\n",
        "AMOUNT = \"amount\"\n",
        "LOG_DAMAGE = \"unmitigatedAmount\"  # amount column is affected by overkill\n",
        "LOG_HIT_TYPE = \"hitType\"\n",
        "LOG_DHIT = \"directHit\"\n",
        "LOG_MULTIPLIER = \"multiplier\"\n",
        "\n",
        "# Replaced names\n",
        "CRIT = \"crit\"\n",
        "DHIT = \"direct_hit\"\n",
        "\n",
        "# Special name added in for an assumed prepull apply\n",
        "ESTIMATED_APPLY_BUFF = \"estimatedapplybuff\"\n",
        "\n",
        "# Special name for \"correct\" card\n",
        "CARD_MATCHES_JOB = \"card_matches_job\"\n",
        "\n",
        "# Mappings (target_subtype)\n",
        "JOB = \"Job\"\n",
        "\n",
        "# Breaking melee vs ranged in seperate mapping\n",
        "TARGET_SUBTYPE_TO_JOB_MELEE = {\n",
        "    \"DarkKnight\": \"DRK\",\n",
        "    \"Gunbreaker\": \"GNB\",\n",
        "    \"Paladin\": \"PLD\",\n",
        "    \"Warrior\": \"WAR\",\n",
        "    \"Ninja\": \"NIN\",\n",
        "    \"Samurai\": \"SAM\",\n",
        "    \"Reaper\": \"RPR\",\n",
        "    \"Monk\": \"MNK\",\n",
        "    \"Dragoon\": \"DRG\",\n",
        "}\n",
        "\n",
        "TARGET_SUBTYPE_TO_JOB_RANGED = {\n",
        "    \"Dancer\": \"DNC\",\n",
        "    \"Bard\": \"BRD\",\n",
        "    \"Machinist\": \"MCH\",\n",
        "    \"RedMage\": \"RDM\",\n",
        "    \"BlackMage\": \"BLM\",\n",
        "    \"Summoner\": \"SMN\",\n",
        "    \"Astrologian\": \"AST\",\n",
        "    \"Scholar\": \"SCH\",\n",
        "    \"Sage\": \"SGE\",\n",
        "    \"WhiteMage\": \"WHM\",\n",
        "}\n",
        "\n",
        "TARGET_SUBTYPE_TO_JOB = {\n",
        "    **TARGET_SUBTYPE_TO_JOB_MELEE,\n",
        "    **TARGET_SUBTYPE_TO_JOB_RANGED,\n",
        "}\n"
      ],
      "metadata": {
        "cellView": "form",
        "id": "Ck9F1u_HOhen"
      },
      "execution_count": 17,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "#@title FFLOGS Api\n",
        "\n",
        "def get_auth_token() -> str:\n",
        "    \"\"\"Fetches an auth token via https://www.fflogs.com/oauth/token.\n",
        "\n",
        "    See https://www.fflogs.com/api/docs/\n",
        "\n",
        "    And https://www.fflogs.com/v2-api-docs/ff/report.doc.html\n",
        "\n",
        "    Requires FFLOGS_CLIENT_ID and FFLOGS_CLIENT_SECRET set as variables.\n",
        "    These can be obtained here: https://www.fflogs.com/api/clients/\n",
        "\n",
        "    Please keep these private to yourself and do not upload them to\n",
        "    github or share them.\n",
        "\n",
        "    :returns: Auth token for FFLogs APIv2\n",
        "    \"\"\"\n",
        "    LOG.debug(f\"Fetching fflogs auth token\")\n",
        "    x = requests.post(\n",
        "        url=\"https://www.fflogs.com/oauth/token\",\n",
        "        data={\"grant_type\": \"client_credentials\"},\n",
        "        auth=(\n",
        "            userdata.get(\"FFLOGS_CLIENT_ID\"),\n",
        "            userdata.get(\"FFLOGS_CLIENT_SECRET\"),\n",
        "        ),\n",
        "    )\n",
        "\n",
        "    x.raise_for_status()\n",
        "    return x.json()[\"access_token\"]\n",
        "\n",
        "\n",
        "@dataclasses.dataclass(frozen=True, repr=False)\n",
        "class FFLogClientV2:\n",
        "    \"\"\"Base class for setting up an FFLogsClient.\"\"\"\n",
        "\n",
        "    # Caches outcomes from https://www.fflogs.com/v2-api-docs/ff/reportability.doc.html\n",
        "    # Maps IDs to ability data\n",
        "    # Structure:\n",
        "    # {\n",
        "    #     log_code: {\n",
        "    #         ability_id (str): ability_data (dict)\n",
        "    #         ...\n",
        "    #     },\n",
        "    #     ...\n",
        "    # }\n",
        "    ability_id_cache: Dict = dataclasses.field(default_factory=dict)\n",
        "\n",
        "    # Caches outcomes from https://www.fflogs.com/v2-api-docs/ff/reportactor.doc.html\n",
        "    # Maps ids to actor data\n",
        "    actor_cache: Dict = dataclasses.field(default_factory=dict)\n",
        "\n",
        "    # Constructs a client on init\n",
        "    client: python_graphql_client.GraphqlClient = dataclasses.field(\n",
        "        default_factory=lambda: (\n",
        "            python_graphql_client.GraphqlClient(\n",
        "                endpoint=\"https://www.fflogs.com/api/v2/client\",\n",
        "                headers={\"Authorization\": f\"Bearer {get_auth_token()}\"},\n",
        "            )\n",
        "        )\n",
        "    )\n",
        "\n",
        "    def get_fights(self, code: str, fight_type: str = \"Kills\") -> List[Dict]:\n",
        "        \"\"\"Fetches the high level fights from a log.\"\"\"\n",
        "        LOG.debug(f\"Fetching fflogs reportData:report:fights via {code}\")\n",
        "        return self.client.execute(\n",
        "            query=\"\"\"query reportData($code: String!, $kills: KillType!){\n",
        "                  reportData\n",
        "                  {\n",
        "                    report(code: $code){\n",
        "                      fights(killType: $kills){\n",
        "                        encounterID\n",
        "                        startTime\n",
        "                        endTime\n",
        "                        name\n",
        "                        kill\n",
        "                        standardComposition\n",
        "                        combatTime\n",
        "                      }\n",
        "                    }\n",
        "                  }\n",
        "                }\n",
        "                \"\"\",\n",
        "            variables={\n",
        "                \"code\": code,\n",
        "                \"kills\": fight_type,\n",
        "            },\n",
        "        )[\"data\"][\"reportData\"][\"report\"][\"fights\"]\n",
        "\n",
        "    def get_abilities(self, code: str) -> List[Dict]:\n",
        "        \"\"\"Fetches the high level fights from a log.\"\"\"\n",
        "        LOG.debug(f\"Fetching fflogs reportData:report:fights via {code}\")\n",
        "        return self.client.execute(\n",
        "            query=\"\"\"query reportData($code: String!){\n",
        "                  reportData\n",
        "                  {\n",
        "                    report(code: $code){\n",
        "                      masterData{\n",
        "                        abilities{\n",
        "                          gameID\n",
        "                          name\n",
        "                          type\n",
        "                        }\n",
        "                      }\n",
        "                    }\n",
        "                  }\n",
        "                }\n",
        "                \"\"\",\n",
        "            variables={\n",
        "                \"code\": code,\n",
        "            },\n",
        "        )[\"data\"][\"reportData\"][\"report\"][\"masterData\"][\"abilities\"]\n",
        "\n",
        "    def stream_fight_events(\n",
        "        self, code: str, fight_type: str = \"Kills\"\n",
        "    ) -> Iterator:\n",
        "        \"\"\"Streams fight event data\n",
        "\n",
        "        Structure:\n",
        "            yield fight_name, [{events...}]\n",
        "        \"\"\"\n",
        "        self._populate_caches(code=code)\n",
        "\n",
        "        for fight in self.get_fights(code=code, fight_type=fight_type):\n",
        "            yield (\n",
        "                fight[\"name\"],\n",
        "                self._process_fight_data(\n",
        "                    data=self.get_raw_events_from_fight(\n",
        "                        code=code,\n",
        "                        fight=fight,\n",
        "                        fight_type=fight_type,\n",
        "                    ),\n",
        "                    code=code,\n",
        "                ),\n",
        "            )\n",
        "\n",
        "    def _process_fight_data(self, data: Iterator, code: str) -> List:\n",
        "        \"\"\"Applies actor and ability name replacement.\"\"\"\n",
        "        return [\n",
        "            self._process_raw_event(data=raw_data_event, code=code)\n",
        "            for raw_data_event in data\n",
        "        ]\n",
        "\n",
        "    def _process_raw_event(self, data: Dict, code: str) -> Dict:\n",
        "        if \"abilityGameID\" in data:\n",
        "            data[ABILITY_NAME] = self._get_ability(\n",
        "                data[\"abilityGameID\"], code=code\n",
        "            )\n",
        "\n",
        "        if \"sourceID\" in data:\n",
        "            data[SOURCE_NAME] = self._get_actor_name(\n",
        "                data[\"sourceID\"], code=code\n",
        "            )\n",
        "            source_actor = self._get_actor(data[\"sourceID\"], code=code)\n",
        "            if source_actor:\n",
        "                data[SOURCE_TYPE] = source_actor[\"type\"]\n",
        "                data[SOURCE_SUBTYPE] = source_actor[\"subType\"]\n",
        "                if source_actor[\"petOwner\"]:\n",
        "                    data[\n",
        "                        SOURCE_PET_ACTOR\n",
        "                    ] = self._get_actor_name(\n",
        "                        source_actor[\"petOwner\"], code=code\n",
        "                    )\n",
        "\n",
        "        if \"targetID\" in data:\n",
        "            data[TARGET_NAME] = self._get_actor_name(\n",
        "                data[\"targetID\"], code=code\n",
        "            )\n",
        "            target_actor = self._get_actor(data[\"targetID\"], code=code)\n",
        "            if target_actor:\n",
        "                data[TARGET_TYPE] = target_actor[\"type\"]\n",
        "                data[TARGET_SUBTYPE] = target_actor[\"subType\"]\n",
        "                if target_actor[\"petOwner\"]:\n",
        "                    data[\n",
        "                        TARGET_PET_ACTOR\n",
        "                    ] = self._get_actor_name(\n",
        "                        target_actor[\"petOwner\"], code=code\n",
        "                    )\n",
        "\n",
        "        return data\n",
        "\n",
        "    def _get_ability(self, ability_id: str, code: str) -> str:\n",
        "        if code not in self.ability_id_cache:\n",
        "            self._populate_caches(code)\n",
        "\n",
        "        return (\n",
        "            \"UNKNOWN\"\n",
        "            if ability_id not in self.ability_id_cache[code]\n",
        "            else (self.ability_id_cache[code][ability_id][\"name\"])\n",
        "        )\n",
        "\n",
        "    def _get_actor(self, actor_id: str, code: str) -> Optional[Dict]:\n",
        "        \"\"\"Gets the full actor data.\"\"\"\n",
        "        if code not in self.actor_cache:\n",
        "            self._populate_caches(code)\n",
        "\n",
        "        return self.actor_cache[code].get(actor_id)\n",
        "\n",
        "    def _get_actor_name(self, actor_id: str, code: str) -> Optional[str]:\n",
        "        \"\"\"Gets the full actor name. Adjusts pet names to include owner.\"\"\"\n",
        "        actor: Optional[Dict] = self._get_actor(actor_id=actor_id, code=code)\n",
        "        if actor is None:\n",
        "            return None\n",
        "\n",
        "        if not actor[\"petOwner\"]:\n",
        "            return actor[\"name\"]\n",
        "\n",
        "        pet_owner: Optional[Dict] = self._get_actor(\n",
        "            actor_id=actor[\"petOwner\"], code=code\n",
        "        )[\"name\"]\n",
        "\n",
        "        return f\"{actor['name']} ({pet_owner})\"\n",
        "\n",
        "    def _populate_caches(self, code: str) -> None:\n",
        "        LOG.debug(f\"Fetching fflogs reportData:report:masterData via {code}\")\n",
        "        log_data = self.client.execute(\n",
        "            query=\"\"\"query reportData($code: String!){\n",
        "                          reportData\n",
        "                          {\n",
        "                            report(code: $code){\n",
        "                              masterData{\n",
        "                                abilities {\n",
        "                                  gameID\n",
        "                                  name\n",
        "                                  type\n",
        "                                }\n",
        "                                actors {\n",
        "                                  id\n",
        "                                  name\n",
        "                                  petOwner\n",
        "                                  subType\n",
        "                                  type\n",
        "                                }\n",
        "                              }\n",
        "                            }\n",
        "                          }\n",
        "                        }\n",
        "                        \"\"\",\n",
        "            variables={\"code\": code},\n",
        "        )[\"data\"][\"reportData\"][\"report\"][\"masterData\"]\n",
        "\n",
        "        self.ability_id_cache[code] = {\n",
        "            ability[\"gameID\"]: ability for ability in log_data[\"abilities\"]\n",
        "        }\n",
        "\n",
        "        self.actor_cache[code] = {\n",
        "            actor[\"id\"]: actor for actor in log_data[\"actors\"]\n",
        "        }\n",
        "\n",
        "        LOG.debug(f\"Stored actor and ability ids.\")\n",
        "\n",
        "        return\n",
        "\n",
        "    def get_raw_events_from_fight(\n",
        "        self, code: str, fight: Dict, fight_type: str = \"Kills\"\n",
        "    ) -> Iterator:\n",
        "        \"\"\"Gets the raw events from fight.\"\"\"\n",
        "        next_timestamp = start_time = fight[\"startTime\"]\n",
        "        end_time = fight[\"endTime\"]\n",
        "        LOG.debug(\n",
        "            f\"Fetching fflogs reportData:report:events via {code} [{start_time}, {end_time}]\"\n",
        "        )\n",
        "\n",
        "        # Pretty animation\n",
        "\n",
        "        percent_complete = (\n",
        "            \"{:5.2f}%\".format(\n",
        "                (float(next_timestamp) - float(start_time))\n",
        "                / (float(end_time) - float(start_time))\n",
        "                * 100.0\n",
        "            )\n",
        "            if (float(end_time) > float(start_time))\n",
        "            else \"100%\"\n",
        "        )\n",
        "        animation: str = f\"Loading {fight['name']} ({percent_complete})\"\n",
        "        while next_timestamp is not None and int(next_timestamp) < int(\n",
        "            end_time\n",
        "        ):\n",
        "            # Pretty animation\n",
        "            print(animation, end=\"\\r\")\n",
        "            page_data, next_timestamp = self._get_events_from_fight(\n",
        "                code=code,\n",
        "                start_time=next_timestamp,\n",
        "                end_time=end_time,\n",
        "                fight_type=fight_type,\n",
        "            )\n",
        "\n",
        "            for event_data in page_data:\n",
        "                yield event_data\n",
        "\n",
        "            percent_complete = \"{:5.2f}%\".format(\n",
        "                (float(next_timestamp or end_time) - float(start_time))\n",
        "                / (float(end_time) - float(start_time))\n",
        "                * 100.0\n",
        "            )\n",
        "            animation: str = f\"Loading {fight['name']} ({percent_complete})\"\n",
        "        print(f\"{animation}: DONE\")\n",
        "\n",
        "    def _get_events_from_fight(\n",
        "        self,\n",
        "        code: str,\n",
        "        start_time: str,\n",
        "        end_time: str,\n",
        "        fight_type: str = \"Kills\",\n",
        "    ) -> Tuple:\n",
        "        \"\"\"Inner function for the timestamp management.\"\"\"\n",
        "        raw_data = self.client.execute(\n",
        "            query=\"\"\"query reportData($code: String!, $startTime: Float!, $endTime: Float!, $kills: KillType!, $limit: Int!){\n",
        "                          reportData\n",
        "                          {\n",
        "                            report(code: $code){\n",
        "                              events(killType: $kills, startTime: $startTime, endTime: $endTime, limit: $limit){\n",
        "                                data\n",
        "                                nextPageTimestamp\n",
        "                              }\n",
        "                            }\n",
        "                          }\n",
        "                        }\n",
        "                        \"\"\",\n",
        "            variables={\n",
        "                \"code\": code,\n",
        "                \"startTime\": start_time,\n",
        "                \"endTime\": end_time,\n",
        "                \"kills\": fight_type,\n",
        "                \"limit\": 10000,\n",
        "            },\n",
        "        )[\"data\"][\"reportData\"][\"report\"][\"events\"]\n",
        "\n",
        "        return raw_data[\"data\"], raw_data[\"nextPageTimestamp\"]"
      ],
      "metadata": {
        "cellView": "form",
        "id": "59n-B-brRT65"
      },
      "execution_count": 4,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "#@title Basic FFLOGS event processing logic.\n",
        "\n",
        "def get_fflog_events_dataframe(data: List) -> pd.DataFrame:\n",
        "    \"\"\"Converts fflog events into a dataframe with timestamp set.\"\"\"\n",
        "    event_data = pd.DataFrame(data)\n",
        "\n",
        "    # Run lb-update hard sync to ensure damage taken instances are consistent\n",
        "    potential_start_times = [\n",
        "        event_data[event_data[TYPE] == DAMAGE][\"timestamp\"].min()\n",
        "    ]\n",
        "    if event_data[TYPE].isin([\"limitbreakupdate\"]).any():\n",
        "        potential_start_times.append(\n",
        "            event_data[event_data[TYPE] == \"limitbreakupdate\"][\n",
        "                \"timestamp\"\n",
        "            ].min()\n",
        "        )\n",
        "\n",
        "    start_timestamp = min(potential_start_times)\n",
        "\n",
        "    event_data.insert(\n",
        "        0,\n",
        "        FIGHT_TIME,\n",
        "        ((event_data[\"timestamp\"] - start_timestamp) / 1000.0).round(4),\n",
        "    )\n",
        "    del event_data[\"timestamp\"]\n",
        "    return event_data"
      ],
      "metadata": {
        "cellView": "form",
        "id": "jJoaJQvbhEKW"
      },
      "execution_count": 5,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "#@title FFLOGS casting event processing logic.\n",
        "_DEFAULT_IGNORE_ABILITY_ID = {\n",
        "    33218,  # secondary activation of tech, signals buff activation but not damage.\n",
        "}\n",
        "\n",
        "_DEFAULT_IGNORE_ABILITY_NAME = {}\n",
        "\n",
        "DATA_TO_CRIT_BOOL = {\n",
        "    1.0: False,\n",
        "    2.0: True,\n",
        "}\n",
        "\n",
        "\n",
        "def process_player_casts(\n",
        "    fight_events: pd.DataFrame,\n",
        "    ignore_ids: Optional[Set[int]] = None,\n",
        "    ignore_names: Optional[Set[int]] = None,\n",
        ") -> pd.DataFrame:\n",
        "    \"\"\"Processes all casts into a DF with all relevant data.\n",
        "\n",
        "    Specifically, does the following:\n",
        "        - Tracks all Cast events, removing all weird edge-cases (default or supplied)\n",
        "        - If there is a related \"calculateddamage\" event, add that info\n",
        "\n",
        "    To focus on casts by a single player, simply filter by source_name before using this.\n",
        "    \"\"\"\n",
        "    ability_ids_to_ignore = (\n",
        "        ignore_ids if ignore_ids is not None else _DEFAULT_IGNORE_ABILITY_ID\n",
        "    )\n",
        "    ability_names_to_ignore = (\n",
        "        ignore_names\n",
        "        if ignore_names is not None\n",
        "        else _DEFAULT_IGNORE_ABILITY_NAME\n",
        "    )\n",
        "\n",
        "    # Define a filter, if necessary\n",
        "    if ability_names_to_ignore or ability_ids_to_ignore:\n",
        "        if ability_ids_to_ignore:\n",
        "            fight_events = fight_events[\n",
        "                ~fight_events[ABILITY_ID].isin(\n",
        "                    ability_ids_to_ignore\n",
        "                )\n",
        "            ]\n",
        "\n",
        "        if ability_names_to_ignore:\n",
        "            fight_events = fight_events[\n",
        "                ~fight_events[ABILITY_NAME].isin(\n",
        "                    ability_names_to_ignore\n",
        "                )\n",
        "            ]\n",
        "\n",
        "    # Get the specific subframe for particular events.\n",
        "    casts = fight_events[\n",
        "        fight_events[TYPE] == CAST\n",
        "    ]\n",
        "\n",
        "    calculated_damage = fight_events[\n",
        "        fight_events[TYPE] == PREPARE\n",
        "    ]\n",
        "\n",
        "    # Buggy, and no need to keep right now.\n",
        "    # calculated_damage.loc[:, DHIT] = calculated_damage[\n",
        "    #     LOG_DHIT\n",
        "    # ].fillna(False)\n",
        "    # calculated_damage.loc[:, CRIT] = calculated_damage[\n",
        "    #     LOG_HIT_TYPE\n",
        "    # ].map(DATA_TO_CRIT_BOOL)\n",
        "\n",
        "    merge_columns = [\n",
        "        FIGHT_TIME,\n",
        "        SOURCE_NAME,\n",
        "        SOURCE_ID,\n",
        "        ABILITY_ID,\n",
        "        ABILITY_NAME,\n",
        "    ]\n",
        "    # Add calculated damage info\n",
        "    casts_with_damage = pd.merge(\n",
        "        left=casts,\n",
        "        right=calculated_damage[\n",
        "            [\n",
        "                *merge_columns,\n",
        "                # CRIT,\n",
        "                # DHIT,\n",
        "                AMOUNT,\n",
        "                LOG_DAMAGE,\n",
        "                LOG_MULTIPLIER,\n",
        "                BONUS_PERCENT,\n",
        "            ]\n",
        "        ],\n",
        "        how=\"left\",\n",
        "        on=merge_columns,\n",
        "    )\n",
        "\n",
        "    return (\n",
        "        casts_with_damage.sort_values(by=[FIGHT_TIME], axis=0)\n",
        "        .dropna(how=\"all\", axis=1)\n",
        "        .reset_index(drop=True)\n",
        "    )\n",
        "\n"
      ],
      "metadata": {
        "cellView": "form",
        "id": "kHs7HxeIhTo4"
      },
      "execution_count": 6,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "#@title FFLOGS raidbuff event processing logic.\n",
        "MELEE_CARDS = {\n",
        "    \"The Arrow\",\n",
        "    \"The Balance\",\n",
        "    \"The Spear\",\n",
        "}\n",
        "\n",
        "RANGED_CARDS = {\n",
        "    \"The Ewer\",\n",
        "    \"The Spire\",\n",
        "    \"The Bole\",\n",
        "}\n",
        "\n",
        "CARDS = {*MELEE_CARDS, *RANGED_CARDS}\n",
        "\n",
        "RAIDBUFFS = {\n",
        "    \"Battle Litany\",\n",
        "    \"Arcane Circle\",\n",
        "    \"Brotherhood\",\n",
        "    \"Divination\",\n",
        "    \"Embolden\",\n",
        "    \"Left Eye\",\n",
        "    # \"Right Eye\",  Ignored since right eye is the personal part of the buff\n",
        "    \"Radiant Finale\",\n",
        "    \"Battle Voice\",\n",
        "    \"Technical Finish\",\n",
        "    \"Devilment\",\n",
        "    \"Searing Light\",\n",
        "    \"Standard Finish\",\n",
        "    \"The Wanderer's Minuet\",\n",
        "    \"Army's Paeon\",\n",
        "    \"Mage's Ballad\",\n",
        "    *CARDS,\n",
        "}\n",
        "\n",
        "DEBUFFS = {\"Vulnerability Up\", \"Chain Stratagem\"}\n",
        "\n",
        "pd.options.mode.copy_on_write = True\n",
        "\n",
        "\n",
        "def process_boss_debuffs(fight_events: pd.DataFrame) -> pd.DataFrame:\n",
        "    \"\"\"Processes all targeted raidbuff attributed to a player into a DF with all relevant data.\n",
        "\n",
        "    Specifically, does the following:\n",
        "        - Tracks all applydebuff of relevant names (DEBUFFS)\n",
        "\n",
        "    Does not include personal debuff applications (SoD, Trick).\n",
        "    \"\"\"\n",
        "    debuff_applications = fight_events[\n",
        "        (\n",
        "            fight_events[TYPE].isin(\n",
        "                [APPLY_DEBUFF, \"refreshdebuff\"]\n",
        "            )\n",
        "        )\n",
        "        & (fight_events[ABILITY_NAME].isin(DEBUFFS))\n",
        "    ]\n",
        "\n",
        "    return (\n",
        "        debuff_applications.sort_values(by=[FIGHT_TIME], axis=0)\n",
        "        .dropna(how=\"all\", axis=1)\n",
        "        .reset_index(drop=True)\n",
        "    )\n",
        "\n",
        "\n",
        "def process_party_buffs(fight_events: pd.DataFrame) -> pd.DataFrame:\n",
        "    \"\"\"Processes all buffs applications into a DF with all relevant data.\n",
        "\n",
        "    Specifically, does the following:\n",
        "        - Tracks all applybuff of relevant names (raidbuffs)\n",
        "    \"\"\"\n",
        "    # Get the specific subframe for particular events.\n",
        "    buff_applications = fight_events[\n",
        "        (\n",
        "            fight_events[TYPE].isin(\n",
        "                [APPLY_BUFF, REFRESH_BUFF]\n",
        "            )\n",
        "        )\n",
        "        & (fight_events[ABILITY_NAME].isin(RAIDBUFFS))\n",
        "    ]\n",
        "\n",
        "    # Get nice column for cards stuff\n",
        "    buff_applications.loc[:, CARD_MATCHES_JOB] = 0\n",
        "    melee_card_usages = buff_applications[ABILITY_NAME].isin(\n",
        "        MELEE_CARDS\n",
        "    )\n",
        "    ranged_card_usages = buff_applications[ABILITY_NAME].isin(\n",
        "        RANGED_CARDS\n",
        "    )\n",
        "    buff_applications.loc[\n",
        "        melee_card_usages,\n",
        "        CARD_MATCHES_JOB,\n",
        "    ] = np.where(\n",
        "        buff_applications.loc[\n",
        "            melee_card_usages, TARGET_SUBTYPE\n",
        "        ].isin(TARGET_SUBTYPE_TO_JOB_MELEE),\n",
        "        2,\n",
        "        1,\n",
        "    )\n",
        "\n",
        "    buff_applications.loc[\n",
        "        ranged_card_usages,\n",
        "        CARD_MATCHES_JOB,\n",
        "    ] = np.where(\n",
        "        ~buff_applications.loc[\n",
        "            ranged_card_usages, TARGET_SUBTYPE\n",
        "        ].isin(TARGET_SUBTYPE_TO_JOB_MELEE),\n",
        "        2,\n",
        "        1,\n",
        "    )\n",
        "\n",
        "    return (\n",
        "        buff_applications.sort_values(by=[FIGHT_TIME], axis=0)\n",
        "        .dropna(how=\"all\", axis=1)\n",
        "        .reset_index(drop=True)\n",
        "    )\n"
      ],
      "metadata": {
        "cellView": "form",
        "id": "EVlBWf7Sj8uc"
      },
      "execution_count": 7,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "#@title FFLOGS prepull event processing logic.\n",
        "ALL_PREPULL = {\n",
        "    \"Medicated\": 30,\n",
        "    \"Swiftcast\": 10,\n",
        "}\n",
        "\n",
        "WAR_PREPULL = {\n",
        "    \"Inner Release\": 15,\n",
        "}\n",
        "\n",
        "PLD_PREPULL = {\n",
        "    \"Fight or Flight\": 20,\n",
        "}\n",
        "\n",
        "DRK_PREPULL = {\n",
        "    \"Delirium\": 15,\n",
        "    \"Blood Weapon\": 15,\n",
        "    \"Blackest Night\": 7,\n",
        "}\n",
        "\n",
        "GNB_PREPULL = {\n",
        "    \"No Mercy\": 20,\n",
        "}\n",
        "\n",
        "RPR_PREPULL = {\n",
        "    \"Arcane Circle\": 20,\n",
        "}\n",
        "\n",
        "MNK_PREPULL = {\n",
        "    \"Riddle of Fire\": 20,\n",
        "    \"Riddle of Wind\": 15,\n",
        "    \"Brotherhood\": 15,\n",
        "}\n",
        "\n",
        "DRG_PREPULL = {\n",
        "    \"Life Surge\": 5,\n",
        "    \"Lance Charge\": 20,\n",
        "    \"Battle Litany\": 15,\n",
        "    \"Right Eye\": 20,\n",
        "    \"Left Eye\": 20,\n",
        "}\n",
        "\n",
        "NIN_PREPULL = {\n",
        "    \"Kassatsu\": 15,\n",
        "}\n",
        "\n",
        "SAM_PREPULL = {\n",
        "    \"Meikyo Shisui\": 15,\n",
        "}\n",
        "\n",
        "BRD_PREPULL = {\n",
        "    \"Battle Voice\": 15,\n",
        "}\n",
        "\n",
        "DNC_PREPULL = {\n",
        "    \"Standard Finish\": 30,\n",
        "    \"Technical Finish\": 20,\n",
        "    \"Devilment\": 20,\n",
        "    # \"Flourish\":\n",
        "}\n",
        "\n",
        "MCH_PREPULL = {}\n",
        "\n",
        "RDM_PREPULL = {\n",
        "    \"Acceleration\": 20,\n",
        "    \"Dualcast\": 5,\n",
        "    \"Embolden\": 20,\n",
        "}\n",
        "\n",
        "BLM_PREPULL = {\n",
        "    \"Triplecast\": 10,\n",
        "    \"Sharpcast\": 30,\n",
        "    \"Ley Lines\": 30,\n",
        "}\n",
        "\n",
        "SMN_PREPULL = {\n",
        "    \"Searing Light\": 30,\n",
        "}\n",
        "\n",
        "WHM_PREPULL = {\n",
        "    \"Presence of Mind\": 15,\n",
        "}\n",
        "\n",
        "AST_PREPULL = {\n",
        "    \"the Arrow\": 15,\n",
        "    \"the Balance\": 15,\n",
        "    \"the Ewer\": 15,\n",
        "    \"the Spire\": 15,\n",
        "    \"the Spear\": 15,\n",
        "    \"the Bole\": 15,\n",
        "    \"Divination\": 15,\n",
        "    \"Earthly Dominance\": 10,\n",
        "    \"Giant Dominance\": 20,\n",
        "}\n",
        "\n",
        "SCH_PREPULL = {}\n",
        "\n",
        "SGE_PREPULL = {\n",
        "    \"Eukrasia\": 30,  # has no duration, 30 is fine.\n",
        "}\n",
        "\n",
        "\n",
        "POTENTIAL_PREPULL_BUFFS = {\n",
        "    **ALL_PREPULL,\n",
        "    **AST_PREPULL,\n",
        "    **BLM_PREPULL,\n",
        "    **BRD_PREPULL,\n",
        "    **DNC_PREPULL,\n",
        "    **DRG_PREPULL,\n",
        "    **DRK_PREPULL,\n",
        "    **GNB_PREPULL,\n",
        "    **MCH_PREPULL,\n",
        "    **MNK_PREPULL,\n",
        "    **NIN_PREPULL,\n",
        "    **RPR_PREPULL,\n",
        "    **PLD_PREPULL,\n",
        "    **SAM_PREPULL,\n",
        "    **SGE_PREPULL,\n",
        "    **SCH_PREPULL,\n",
        "    **SMN_PREPULL,\n",
        "    **WAR_PREPULL,\n",
        "    **WHM_PREPULL,\n",
        "}\n",
        "\n",
        "\n",
        "def prepull_aura_frame(fight_events: pd.DataFrame) -> pd.DataFrame:\n",
        "    \"\"\"Builds a frame depicting prepull auras.\"\"\"\n",
        "    output = []\n",
        "    combatantinfo = fight_events[\n",
        "        fight_events[TYPE] == \"combatantinfo\"\n",
        "    ]\n",
        "    for (\n",
        "        fight_time,\n",
        "        source_id,\n",
        "        source_name,\n",
        "        source_type,\n",
        "        source_subtype,\n",
        "        auras,\n",
        "    ) in zip(\n",
        "        combatantinfo[FIGHT_TIME],\n",
        "        combatantinfo[SOURCE_ID],\n",
        "        combatantinfo[SOURCE_NAME],\n",
        "        combatantinfo[SOURCE_TYPE],\n",
        "        combatantinfo[SOURCE_SUBTYPE],\n",
        "        combatantinfo[\"auras\"],\n",
        "    ):\n",
        "        for aura in auras:\n",
        "            output.append(\n",
        "                {\n",
        "                    FIGHT_TIME: fight_time,\n",
        "                    SOURCE_ID: source_id,\n",
        "                    SOURCE_NAME: source_name,\n",
        "                    SOURCE_TYPE: source_type,\n",
        "                    SOURCE_SUBTYPE: source_subtype,\n",
        "                    **aura,\n",
        "                }\n",
        "            )\n",
        "\n",
        "    return pd.DataFrame(output)\n",
        "\n",
        "\n",
        "def infer_prepull_buff_casts(fight_events: pd.DataFrame) -> pd.DataFrame:\n",
        "    \"\"\"Infers all prepull buffs applications into a DF with all relevant data.\n",
        "\n",
        "    Specifically, does the following:\n",
        "        - Looks at the combatant info row to get starting buffs\n",
        "        - For each one, looks for when the buff falls off in combat\n",
        "        - From this, inferences the original application time\n",
        "    \"\"\"\n",
        "    # Get the specific subframe for particular events.\n",
        "    outputs: List[pd.DataFrame] = []\n",
        "\n",
        "    # Get a quick subframe for the first refresh or removal of a buff\n",
        "    buff_removal_or_refresh = (\n",
        "        fight_events[\n",
        "            (\n",
        "                fight_events[TYPE].isin(\n",
        "                    [REMOVE_BUFF, \"refreshbuff\"]\n",
        "                )\n",
        "            )\n",
        "        ]\n",
        "        .sort_values(by=[FIGHT_TIME], axis=0)\n",
        "        .dropna(how=\"all\", axis=1)\n",
        "        .reset_index(drop=True)\n",
        "    )\n",
        "\n",
        "    # groupby each player, build outputs per player\n",
        "    for player_name, info_data in fight_events[\n",
        "        fight_events[TYPE] == \"combatantinfo\"\n",
        "    ].groupby(by=SOURCE_NAME):\n",
        "        LOG.debug(f\"Finding auras for {player_name}\")\n",
        "        # while info data should be a single row, just loop through it for safety.\n",
        "        for auras in info_data[\"auras\"]:\n",
        "            for aura in auras:\n",
        "                # Check aura-names\n",
        "                duration = POTENTIAL_PREPULL_BUFFS.get(aura[\"name\"])\n",
        "                if duration is None:\n",
        "                    LOG.debug(f\"Skipping {aura}, not tracking.\")\n",
        "                    continue\n",
        "                LOG.debug(f\"Processing auras {aura}.\")\n",
        "\n",
        "                # get the earliest removal or refresh of the buff\n",
        "                specific_buff_removal_or_refresh = buff_removal_or_refresh[\n",
        "                    (\n",
        "                        buff_removal_or_refresh[TARGET_NAME]\n",
        "                        == player_name\n",
        "                    )\n",
        "                    & (\n",
        "                        buff_removal_or_refresh[ABILITY_NAME]\n",
        "                        == aura[\"name\"]\n",
        "                    )\n",
        "                ]\n",
        "\n",
        "                if specific_buff_removal_or_refresh.empty:\n",
        "                    LOG.warning(f\"Unable to find removal event for {aura}\")\n",
        "                    continue\n",
        "\n",
        "                earliest_buff_reference = specific_buff_removal_or_refresh.head(\n",
        "                    1\n",
        "                ).copy(deep=True)\n",
        "                # Inference the original usage, via subtracting the fight-time\n",
        "                earliest_buff_reference.loc[\n",
        "                    :, FIGHT_TIME\n",
        "                ] -= duration\n",
        "                earliest_buff_reference.loc[\n",
        "                    :, TYPE\n",
        "                ] = ESTIMATED_APPLY_BUFF\n",
        "                outputs.append(earliest_buff_reference)\n",
        "\n",
        "    if not outputs:\n",
        "        return pd.DataFrame()\n",
        "\n",
        "    return (\n",
        "        pd.concat(outputs, axis=0)\n",
        "        .round({FIGHT_TIME: 3})\n",
        "        .sort_values(by=[FIGHT_TIME], axis=0)\n",
        "        .dropna(how=\"all\", axis=1)\n",
        "        .reset_index(drop=True)\n",
        "    )\n"
      ],
      "metadata": {
        "cellView": "form",
        "id": "WCAq9ToNkOoP"
      },
      "execution_count": 8,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "#@title Etro API Response Object: Generic Stats.\n",
        "CASTER_STR = {\n",
        "    \"SCH\": 354,\n",
        "    \"SMN\": 354,\n",
        "    \"RDM\": 217,\n",
        "    \"SGE\": 237,\n",
        "    \"WHM\": 217,\n",
        "    \"AST\": 198,\n",
        "    \"BLM\": 198,\n",
        "}\n",
        "\n",
        "\n",
        "class Stats:\n",
        "    \"\"\"Custom object for stats from etro, used for programming convenience.\"\"\"\n",
        "\n",
        "    def __init__(self, etro_data: Dict) -> None:\n",
        "        \"\"\"Initializes from extro data, does some basic operations.\"\"\"\n",
        "        self._etro_data: Dict = etro_data\n",
        "        stats = {\n",
        "            params[\"name\"]: params[\"value\"]\n",
        "            if \"units\" not in params or params[\"units\"] != \"%\"\n",
        "            else (round(params[\"value\"] / 100.0, 4))\n",
        "            for params in etro_data[\"totalParams\"]\n",
        "        }\n",
        "\n",
        "        self.stats = {\n",
        "            \"id\": self._etro_data[\"id\"],\n",
        "            \"url\": f\"https://etro.gg/gearset/{self._etro_data['id']}\",\n",
        "            \"jobAbbrev\": self._etro_data[\"jobAbbrev\"],\n",
        "            # Cleans up job values\n",
        "            **stats,\n",
        "            \"Direct Critical Rate\": (\n",
        "                stats[\"Critical Hit Rate\"] * stats[\"Direct Hit Rate\"]\n",
        "            ),\n",
        "            \"Direct Critical Multiplier\": (\n",
        "                1.25 * stats[\"Critical Hit Multiplier\"]\n",
        "            ),\n",
        "        }\n",
        "\n",
        "    @property\n",
        "    def average_item_level(self) -> Union[str, int, float]:\n",
        "        return self.stats[\"Average Item Level\"]\n",
        "\n",
        "    @property\n",
        "    def crt(self) -> Union[str, int, float]:\n",
        "        return self.stats[\"CRT\"]\n",
        "\n",
        "    @property\n",
        "    def critical_hit_multiplier(self) -> Union[str, int, float]:\n",
        "        return self.stats[\"Critical Hit Multiplier\"]\n",
        "\n",
        "    @property\n",
        "    def critical_hit_rate(self) -> Union[str, int, float]:\n",
        "        return self.stats[\"Critical Hit Rate\"]\n",
        "\n",
        "    @property\n",
        "    def defense(self) -> Union[str, int, float]:\n",
        "        return self.stats[\"DEF\"]\n",
        "\n",
        "    @property\n",
        "    def det(self) -> Union[str, int, float]:\n",
        "        return self.stats[\"DET\"]\n",
        "\n",
        "    @property\n",
        "    def dh(self) -> Union[str, int, float]:\n",
        "        return self.stats[\"DH\"]\n",
        "\n",
        "    @property\n",
        "    def expected_damage(self) -> float:\n",
        "        return self.stats[\"Damage (Expected)\"]\n",
        "\n",
        "    @property\n",
        "    def defense_mitigation(self) -> Union[str, int, float]:\n",
        "        return self.stats[\"Defense Mitigation\"]\n",
        "\n",
        "    @property\n",
        "    def magic_defense_mitigation(self) -> Union[str, int, float]:\n",
        "        return self.stats.get(\n",
        "            \"Magic Defense Mitigation\", self.defense_mitigation\n",
        "        )\n",
        "\n",
        "    @property\n",
        "    def determination_multiplier(self) -> Union[str, int, float]:\n",
        "        return self.stats[\"Determination Multiplier\"]\n",
        "\n",
        "    @property\n",
        "    def direct_critical_multiplier(self) -> Union[str, int, float]:\n",
        "        return self.stats[\"Direct Critical Multiplier\"]\n",
        "\n",
        "    @property\n",
        "    def direct_critical_rate(self) -> Union[str, int, float]:\n",
        "        return self.stats[\"Direct Critical Rate\"]\n",
        "\n",
        "    @property\n",
        "    def direct_hit_rate(self) -> Union[str, int, float]:\n",
        "        return self.stats[\"Direct Hit Rate\"]\n",
        "\n",
        "    @property\n",
        "    def plain_hit_rate(self) -> float:\n",
        "        \"\"\"Rate of plain hits.\"\"\"\n",
        "        return (\n",
        "            1\n",
        "            - self.critical_hit_rate\n",
        "            - self.direct_hit_rate\n",
        "            - self.direct_critical_rate\n",
        "        )\n",
        "\n",
        "    @property\n",
        "    def gcd(self) -> Union[str, int, float]:\n",
        "        return self.stats[\"GCD\"]\n",
        "\n",
        "    @property\n",
        "    def hp(self) -> Union[str, int, float]:\n",
        "        return self.stats[\"HP\"]\n",
        "\n",
        "    @property\n",
        "    def sks(self) -> Union[str, int, float]:\n",
        "        return self.stats.get(\"SKS\")\n",
        "\n",
        "    @property\n",
        "    def sps(self) -> Union[str, int, float]:\n",
        "        return self.stats.get(\"SPS\")\n",
        "\n",
        "    @property\n",
        "    def strength(self) -> Union[str, int, float]:\n",
        "        return self.stats.get(\"STR\", CASTER_STR.get(self.jobabbrev))\n",
        "\n",
        "    @property\n",
        "    def strength_multiplier(self) -> Union[str, int, float]:\n",
        "        return self.stats.get(\"STR Multiplier\")\n",
        "\n",
        "    @property\n",
        "    def mind(self) -> Union[str, int, float]:\n",
        "        return self.stats.get(\"MND\")\n",
        "\n",
        "    @property\n",
        "    def mind_multiplier(self) -> Union[str, int, float]:\n",
        "        return self.stats.get(\"MND Multiplier\")\n",
        "\n",
        "    @property\n",
        "    def intelligence(self) -> Union[str, int, float]:\n",
        "        return self.stats.get(\"INT\")\n",
        "\n",
        "    @property\n",
        "    def intelligence_multiplier(self) -> Union[str, int, float]:\n",
        "        return self.stats.get(\"INT Multiplier\")\n",
        "\n",
        "    @property\n",
        "    def dexterity(self) -> Union[str, int, float]:\n",
        "        return self.stats.get(\"DEX\")\n",
        "\n",
        "    @property\n",
        "    def dexterity_multiplier(self) -> Union[str, int, float]:\n",
        "        return self.stats.get(\"DEX Multiplier\")\n",
        "\n",
        "    @property\n",
        "    def skill_speed_multiplier(self) -> Union[str, int, float]:\n",
        "        return self.stats.get(\"Skill Speed Multiplier\", 1.0)\n",
        "\n",
        "    @property\n",
        "    def spell_speed_multiplier(self) -> Union[str, int, float]:\n",
        "        return self.stats.get(\"Spell Speed Multiplier\", 1.0)\n",
        "\n",
        "    @property\n",
        "    def ten(self) -> Union[str, int, float]:\n",
        "        return self.stats.get(\"TEN\")\n",
        "\n",
        "    @property\n",
        "    def tenacity_mitigation(self) -> Union[str, int, float]:\n",
        "        return self.stats.get(\"Tenacity Mitigation\")\n",
        "\n",
        "    @property\n",
        "    def tenacity_multiplier(self) -> Union[str, int, float]:\n",
        "        return self.stats.get(\"Tenacity Multiplier\", 1.0)\n",
        "\n",
        "    @property\n",
        "    def vit(self) -> Union[str, int, float]:\n",
        "        return self.stats[\"VIT\"]\n",
        "\n",
        "    @property\n",
        "    def weapon_damage(self) -> Union[str, int, float]:\n",
        "        return self.stats.get(\"Weapon Damage\")\n",
        "\n",
        "    @property\n",
        "    def weapon_damage_multiplier(self) -> Union[str, int, float]:\n",
        "        return self.stats[\"Weapon DMG Multiplier\"]\n",
        "\n",
        "    @property\n",
        "    def set_id(self) -> Union[str, int, float]:\n",
        "        return self.stats[\"id\"]\n",
        "\n",
        "    @property\n",
        "    def jobabbrev(self) -> Union[str, int, float]:\n",
        "        return self.stats[\"jobAbbrev\"]\n",
        "\n",
        "    @property\n",
        "    def url(self) -> Union[str, int, float]:\n",
        "        return self.stats[\"url\"]"
      ],
      "metadata": {
        "cellView": "form",
        "id": "5JZ3FG-wk4pS"
      },
      "execution_count": 9,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "#@title Etro API Response Object: Job Stats.\n",
        "\n",
        "class BaseJobStats:\n",
        "    \"\"\"Defines a base framework for job specific stats.\n",
        "\n",
        "    Since the sim does a lot of the work, this just tracks really\n",
        "    basic job specific info like auto speed and mainstat name.\n",
        "    \"\"\"\n",
        "\n",
        "    def __init__(self, stats: Stats) -> None:\n",
        "        \"\"\"Uses base etro stats.\"\"\"\n",
        "        self._stats: Stats = stats\n",
        "\n",
        "    @property\n",
        "    def main_stat(self) -> str:\n",
        "        \"\"\"Mainstat name. Must be set. Must match a base_stats.Stats attribute or etro name.\"\"\"\n",
        "        raise NotImplemented()\n",
        "\n",
        "    @property\n",
        "    def main_stat_value(self) -> int:\n",
        "        return self._stats.stats.get(\n",
        "            self.main_stat, getattr(self._stats, self.main_stat)\n",
        "        )\n",
        "\n",
        "    @property\n",
        "    def auto_speed(self) -> float:\n",
        "        \"\"\"Auto speed. Must be set.\"\"\"\n",
        "        raise NotImplemented()\n",
        "\n",
        "    @property\n",
        "    def adjusted_main_stat(self) -> int:\n",
        "        \"\"\"Adjusted mainstat for parties\"\"\"\n",
        "        return int(self.main_stat_value * 1.05)\n",
        "\n",
        "\n",
        "class SAMStats(BaseJobStats):\n",
        "    main_stat = \"strength\"\n",
        "    auto_speed = 2.64\n",
        "\n",
        "\n",
        "class MNKStats(BaseJobStats):\n",
        "    main_stat = \"strength\"\n",
        "    auto_speed = 2.56\n",
        "\n",
        "\n",
        "class RPRStats(BaseJobStats):\n",
        "    main_stat = \"strength\"\n",
        "    auto_speed = 3.2\n",
        "\n",
        "\n",
        "class DRGStats(BaseJobStats):\n",
        "    main_stat = \"strength\"\n",
        "    auto_speed = 2.80\n",
        "\n",
        "\n",
        "class NINStats(BaseJobStats):\n",
        "    main_stat = \"dexterity\"\n",
        "    auto_speed = 2.56\n",
        "\n",
        "\n",
        "class BRDStats(BaseJobStats):\n",
        "    main_stat = \"dexterity\"\n",
        "    auto_speed = 3.04\n",
        "\n",
        "\n",
        "class MCHStats(BaseJobStats):\n",
        "    main_stat = \"dexterity\"\n",
        "    auto_speed = 2.64\n",
        "\n",
        "\n",
        "class DNCStats(BaseJobStats):\n",
        "    main_stat = \"dexterity\"\n",
        "    auto_speed = 3.12\n",
        "\n",
        "\n",
        "class BLMStats(BaseJobStats):\n",
        "    main_stat = \"intelligence\"\n",
        "    auto_speed = 3.28\n",
        "\n",
        "\n",
        "class RDMStats(BaseJobStats):\n",
        "    main_stat = \"intelligence\"\n",
        "    auto_speed = 3.44\n",
        "\n",
        "\n",
        "class SMNStats(BaseJobStats):\n",
        "    main_stat = \"intelligence\"\n",
        "    auto_speed = 3.12\n",
        "\n",
        "\n",
        "class ASTStats(BaseJobStats):\n",
        "    main_stat = \"mind\"\n",
        "    auto_speed = 3.20\n",
        "\n",
        "\n",
        "class WHMStats(BaseJobStats):\n",
        "    main_stat = \"mind\"\n",
        "    auto_speed = 3.44\n",
        "\n",
        "\n",
        "class SCHStats(BaseJobStats):\n",
        "    main_stat = \"mind\"\n",
        "    auto_speed = 3.12\n",
        "\n",
        "\n",
        "class SGEStats(BaseJobStats):\n",
        "    main_stat = \"mind\"\n",
        "    auto_speed = 2.80\n",
        "\n",
        "\n",
        "class DRKStats(BaseJobStats):\n",
        "    main_stat = \"strength\"\n",
        "    auto_speed = 2.96\n",
        "\n",
        "\n",
        "class WARStats(BaseJobStats):\n",
        "    main_stat = \"strength\"\n",
        "    auto_speed = 3.36\n",
        "\n",
        "\n",
        "class PLDStats(BaseJobStats):\n",
        "    main_stat = \"strength\"\n",
        "    auto_speed = 2.24\n",
        "\n",
        "\n",
        "class GNBStats(BaseJobStats):\n",
        "    main_stat = \"strength\"\n",
        "    auto_speed = 2.80\n",
        "\n",
        "\n",
        "# Mapping all jobs via job abbrev\n",
        "\n",
        "JOB_STATS = {\n",
        "    \"DRK\": DRKStats,\n",
        "    \"PLD\": PLDStats,\n",
        "    \"GNB\": GNBStats,\n",
        "    \"WAR\": WARStats,\n",
        "    \"WHM\": WHMStats,\n",
        "    \"SCH\": SCHStats,\n",
        "    \"AST\": ASTStats,\n",
        "    \"SGE\": SGEStats,\n",
        "    \"SAM\": SAMStats,\n",
        "    \"MNK\": MNKStats,\n",
        "    \"RPR\": RPRStats,\n",
        "    \"DRG\": DRGStats,\n",
        "    \"NIN\": NINStats,\n",
        "    \"DNC\": DNCStats,\n",
        "    \"BRD\": BRDStats,\n",
        "    \"MCH\": MCHStats,\n",
        "    \"BLM\": BLMStats,\n",
        "    \"RDM\": RDMStats,\n",
        "    \"SMN\": SMNStats,\n",
        "}\n",
        "\n",
        "\n",
        "def _get_all_printable_attributes(obj):\n",
        "    output = {}\n",
        "    for attr in sorted(dir(obj)):\n",
        "        if attr.startswith(\"_\"):\n",
        "            # Skip\n",
        "            continue\n",
        "\n",
        "        value = getattr(obj, attr)\n",
        "\n",
        "        if attr == \"job\":\n",
        "            output[attr] = _get_all_printable_attributes(value)\n",
        "        elif not callable(value) and not isinstance(value, pd.DataFrame):\n",
        "            output[attr] = value\n",
        "\n",
        "    return {key: value for key, value in output.items() if value is not None}\n",
        "\n",
        "\n",
        "class FullStats(Stats):\n",
        "    \"\"\"Adds in job specific sub information for each job\"\"\"\n",
        "\n",
        "    @property\n",
        "    def job(self) -> Optional[BaseJobStats]:\n",
        "        \"\"\"Accessor for job stats.\"\"\"\n",
        "        if self.jobabbrev in JOB_STATS:\n",
        "            return JOB_STATS[self.jobabbrev](self)\n",
        "        return None\n",
        "\n",
        "    def __repr__(self) -> str:\n",
        "        \"\"\"Pretty Prints the whole object and all sub stats.\"\"\"\n",
        "        return pprint.pformat(_get_all_printable_attributes(self))\n",
        "\n"
      ],
      "metadata": {
        "cellView": "form",
        "id": "PFdG7bUhlgK0"
      },
      "execution_count": 10,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "#@title Etro API Get Call.\n",
        "\n",
        "@functools.lru_cache\n",
        "def _get_client() -> coreapi.Client:\n",
        "    return coreapi.Client()\n",
        "\n",
        "\n",
        "@functools.lru_cache\n",
        "def _get_document() -> Any:\n",
        "    return _get_client().get(\"https://etro.gg/api/docs/\")\n",
        "\n",
        "\n",
        "@functools.lru_cache\n",
        "def get_gearset_data(set_id: str) -> FullStats:\n",
        "    \"\"\"Gets the gearset data (stats) via an id. Allows url too (handles it for free)\"\"\"\n",
        "    # Handles urls by checking for webpage name and splitting\n",
        "    cleaned_set_id = (\n",
        "        set_id.split(\"/\")[-1] if \"etro.gg/gearset\" in set_id else set_id\n",
        "    )\n",
        "\n",
        "    return FullStats(\n",
        "        _get_client().action(\n",
        "            _get_document(),\n",
        "            [\"gearsets\", \"read\"],\n",
        "            params={\n",
        "                \"id\": cleaned_set_id,\n",
        "            },\n",
        "        )\n",
        "    )"
      ],
      "metadata": {
        "cellView": "form",
        "id": "uW79wDDgnezT"
      },
      "execution_count": 11,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "#@title LogRotationBuilder: Consolidates FFLOGs and ETRO data.\n",
        "APPLY_PREFERRED = {\n",
        "    \"The Arrow\",\n",
        "    \"The Balance\",\n",
        "    \"The Ewer\",\n",
        "    \"The Spire\",\n",
        "    \"The Spear\",\n",
        "    \"The Bole\",\n",
        "}\n",
        "\n",
        "\n",
        "@dataclasses.dataclass\n",
        "class LogRotationBuilder:\n",
        "    \"\"\"Defines a log builder, which integrates etro, fflogs data, and ama's ffxiv sim.\n",
        "\n",
        "    Intended to interact with main.api.fflogs_api.FFLogClientV2.stream_fight_events\n",
        "\n",
        "    :param etro_sets: Mapping of 3-Character jobs ids to etro sets.\n",
        "        To specify a specialize set for a particular fight, see example\n",
        "        ```\n",
        "        {\n",
        "          # Defaults by Player\n",
        "          \"Player 1\": \"https....\",\n",
        "          # Defaults by job\n",
        "          \"DRK\": \"https....\",\n",
        "        }\n",
        "        ```\n",
        "        Sets will be picked prioritizing player name over job\n",
        "    :param fight_events: List of events from fflogs.\n",
        "        This should represent a specific pull.\n",
        "    \"\"\"\n",
        "\n",
        "    etro_sets: Dict[str, str]\n",
        "    fight_events: List[Dict]\n",
        "    _rotations: Dict[str, cr.RotationBuilder] = dataclasses.field(\n",
        "        default_factory=dict\n",
        "    )\n",
        "\n",
        "    def simulate_rotation(\n",
        "        self,\n",
        "        rotation: Optional[str] = None,\n",
        "        display: bool = False,\n",
        "        num_samples: int = 1000,\n",
        "        num_trials: int = 1000,\n",
        "    ) -> Dict:\n",
        "        \"\"\"Simulates a rotation.\n",
        "\n",
        "        Will build if rotations haven't been made.\n",
        "\n",
        "        :param rotation: Specifies rotation to run. if not provided, runs all.\n",
        "          This is be a given player's name\n",
        "        :param display: Specifies if display is wanted. Default is No\n",
        "        :return: Table of outcomes.\n",
        "        \"\"\"\n",
        "        output = {}\n",
        "        rotation_table = self._rotations or self.build_rotation_library()\n",
        "        for rotation_name, rotation_to_use in (\n",
        "            rotation_table\n",
        "            if rotation is None\n",
        "            else {rotation: rotation_table[rotation]}\n",
        "        ).items():\n",
        "            dps, damage, per_skill_damage, t = cr.execute_rotation(\n",
        "                rotation_to_use, num_samples=num_samples\n",
        "            )\n",
        "\n",
        "            print(f\"Results for {rotation_name}: \")\n",
        "            cr.DisplayUtils.print_results(dps, title=\"Average DPS\")\n",
        "            cr.DisplayUtils.print_results(damage, title=\"Average Damage\")\n",
        "\n",
        "            print(\"---Expected max damage over N runs---\")\n",
        "            for num_runs in [1, 5, 10, 20, 50, 100]:\n",
        "                print(\n",
        "                    \"{} Runs: {:.2f}\".format(\n",
        "                        num_runs,\n",
        "                        cr.StatUtils.get_expected_max_in_k_runs(\n",
        "                            dps, num_runs, num_trials=num_trials\n",
        "                        ),\n",
        "                    )\n",
        "                )\n",
        "\n",
        "            if display:\n",
        "                plt.figure()\n",
        "                cr.DisplayUtils.display_results(\n",
        "                    dps,\n",
        "                    title=\"DPS over {} s\".format(\n",
        "                        per_skill_damage[-1][0] / 1000\n",
        "                    ),\n",
        "                )\n",
        "\n",
        "            if display:\n",
        "                WINDOW_LENGTH = 15\n",
        "                plt.figure()\n",
        "                cr.DisplayUtils.display_damage_snapshots_in_time_window(\n",
        "                    rotation_to_use, per_skill_damage, WINDOW_LENGTH\n",
        "                )\n",
        "\n",
        "            print(\"---Times (in ms) when skills get used:---\")\n",
        "            for tmp in rotation_to_use.get_button_press_timing():\n",
        "                print(\"{:>8}: {:>22}\".format(tmp[0], tmp[1]))\n",
        "            print(\"\\n\")\n",
        "\n",
        "            print(\"---Times (in ms) when damage lands:---\")\n",
        "            for sk in per_skill_damage:\n",
        "                print(\n",
        "                    \"time: {:>8}, name: {:>22}, expected_damage: {:9.2f}, variance: {:6.2f}\".format(\n",
        "                        sk[0], sk[1], sk[2], sk[3]\n",
        "                    )\n",
        "                )\n",
        "            output[rotation_name] = {\n",
        "                \"dps\": dps,\n",
        "                \"damage\": damage,\n",
        "                \"t\": t,\n",
        "                \"per_skill_damage\": per_skill_damage,\n",
        "            }\n",
        "\n",
        "        return output\n",
        "\n",
        "    def build_rotation_library(\n",
        "        self, skill_library=cr.SKILL_LIBRARY\n",
        "    ) -> Dict[str, cr.RotationBuilder]:\n",
        "        \"\"\"Builds a rotation library from the specific fight and pull\n",
        "\n",
        "        Process:\n",
        "            Convert and timestamp all events into DataFrame object\n",
        "            Run a groupby over player events via player name.\n",
        "            Infer job name and construct stats object.\n",
        "            Process all the events and convert into rb.add_skill calls.\n",
        "                GL handling all the edgecases lol\n",
        "        \"\"\"\n",
        "        if self._rotations:\n",
        "            return self._rotations\n",
        "\n",
        "        output = {}\n",
        "        event_data = get_fflog_events_dataframe(data=self.fight_events)\n",
        "\n",
        "        events_originated_from_players = event_data[\n",
        "            event_data[SOURCE_TYPE] == \"Player\"\n",
        "        ]\n",
        "\n",
        "        # Run preprocessing for major stuff\n",
        "        # Process all raid buffs via party buffs\n",
        "        raid_buffs = process_party_buffs(\n",
        "            fight_events=events_originated_from_players\n",
        "        )\n",
        "\n",
        "        # Process all raid buffs via boss debuffs\n",
        "        raid_debuffs = process_boss_debuffs(\n",
        "            fight_events=events_originated_from_players\n",
        "        )\n",
        "\n",
        "        # Process all assumed prepull usages\n",
        "        prepull_usages = infer_prepull_buff_casts(\n",
        "            fight_events=event_data\n",
        "        )\n",
        "\n",
        "        for (\n",
        "            player_name,\n",
        "            job_name,\n",
        "        ), player_events in events_originated_from_players.groupby(\n",
        "            by=[\n",
        "                SOURCE_NAME,\n",
        "                SOURCE_SUBTYPE,\n",
        "            ],\n",
        "        ):\n",
        "            if player_name == \"Multiple Players\" or job_name == \"LimitBreak\":\n",
        "                # Ignore these events\n",
        "                continue\n",
        "            job_abbrev: str = TARGET_SUBTYPE_TO_JOB[job_name]\n",
        "            # Get etro set, first checking name specific and then defaults\n",
        "            etro_set_id: Optional[str] = (\n",
        "                self.etro_sets.get(player_name) or self.etro_sets.get(job_abbrev)\n",
        "            )\n",
        "\n",
        "            if etro_set_id is None:\n",
        "                LOG.error(\n",
        "                    f\"Unable to build rotation for {player_name}. No set \"\n",
        "                    f\"found for {job_name} ({job_abbrev}).\"\n",
        "                )\n",
        "                continue\n",
        "\n",
        "            # Construct stats object\n",
        "            etro_stats: FullStats = get_gearset_data(\n",
        "                set_id=etro_set_id\n",
        "            )\n",
        "            rotation_stats: cr.Stats = cr.Stats(\n",
        "                wd=etro_stats.weapon_damage,\n",
        "                weapon_delay=etro_stats.job.auto_speed,\n",
        "                main_stat=etro_stats.job.adjusted_main_stat,\n",
        "                det_stat=etro_stats.det,\n",
        "                dh_stat=etro_stats.dh,\n",
        "                crit_stat=etro_stats.crt,\n",
        "                speed_stat=max(\n",
        "                    etro_stats.sks or 400,\n",
        "                    etro_stats.sps or 400,\n",
        "                ),\n",
        "                job_class=job_abbrev,\n",
        "                tenacity=etro_stats.ten or 400,\n",
        "                healer_or_caster_strength=etro_stats.strength,\n",
        "            )\n",
        "            rb = cr.RotationBuilder(\n",
        "                stats=rotation_stats,\n",
        "                skill_library=skill_library,\n",
        "                enable_autos=False,  # Autos are recorded in events\n",
        "                ignore_trailing_dots=True,\n",
        "                ignore_cast_times=True,\n",
        "                # Manipulation boss actor tick for dot gains is dark tech lol.\n",
        "                # Maybe doable but... why. it's every 3s so assume 1.5\n",
        "                snap_dots_to_server_tick_starting_at=1.5,\n",
        "            )\n",
        "\n",
        "            # Run some processing to help with common problems, etc.\n",
        "            player_casts = process_player_casts(\n",
        "                fight_events=player_events,\n",
        "            )\n",
        "            # Tracking for annoying abilities\n",
        "            dwad_casts = set()\n",
        "            earthly_star: Dict[str, List[float]] = {}\n",
        "            # Process all the actual casts\n",
        "            for (\n",
        "                fight_time,\n",
        "                ability_name,\n",
        "                # bonus_percent,\n",
        "            ) in zip(\n",
        "                player_casts[FIGHT_TIME],\n",
        "                player_casts[ABILITY_NAME],\n",
        "                # player_casts[BONUS_PERCENT],\n",
        "            ):\n",
        "                # DWAD is a Meme\n",
        "                if ability_name == \"Dream Within a Dream\" and (fight_time in dwad_casts):\n",
        "                    LOG.debug(f\"Skipping DWAD cast @ {fight_time} (duplicate)\")\n",
        "                    continue\n",
        "                if ability_name in {\"Earthly Star\", \"Stellar Detonation\"}:\n",
        "                    LOG.debug(f\"Handling {ability_name} cast @ {fight_time} (Earthly Star Logic)\")\n",
        "                    earthly_star[ability_name] = earthly_star.get(ability_name, []) + [fight_time]\n",
        "                    continue\n",
        "                # TODO: confirm modifiers via bonus percent\n",
        "                add_rb_ability(\n",
        "                    rb=rb,\n",
        "                    t=fight_time,\n",
        "                    skill_name=ability_name,\n",
        "                    job_class=job_abbrev,\n",
        "                    buff_effect=False,\n",
        "                )\n",
        "                if ability_name == \"Dream Within a Dream\":\n",
        "                    dwad_casts.add(fight_time)\n",
        "\n",
        "            if not raid_debuffs.empty:\n",
        "                for (\n",
        "                    fight_time,\n",
        "                    ability_name,\n",
        "                    source_name,\n",
        "                    source_subtype,\n",
        "                ) in zip(\n",
        "                    raid_debuffs[FIGHT_TIME],\n",
        "                    raid_debuffs[ABILITY_NAME],\n",
        "                    raid_debuffs[SOURCE_NAME],\n",
        "                    raid_debuffs[SOURCE_SUBTYPE],\n",
        "                ):\n",
        "                    # Ignore if the source = the player, since these are accounted for in casting.\n",
        "                    if source_name == player_name:\n",
        "                        continue\n",
        "                    source_job_abbrev = TARGET_SUBTYPE_TO_JOB[\n",
        "                        source_subtype\n",
        "                    ]\n",
        "\n",
        "                    add_rb_ability(\n",
        "                        rb=rb,\n",
        "                        t=fight_time,\n",
        "                        skill_name=ability_name,\n",
        "                        job_class=source_job_abbrev,\n",
        "                    )\n",
        "\n",
        "            # Filter to ensure the target is the player\n",
        "            raid_buffs_targeting_player = raid_buffs[\n",
        "                raid_buffs[TARGET_NAME] == player_name\n",
        "            ]\n",
        "\n",
        "            if not raid_buffs_targeting_player.empty:\n",
        "                for (\n",
        "                    fight_time,\n",
        "                    ability_name,\n",
        "                    source_name,\n",
        "                    source_subtype,\n",
        "                    card_matches_job,\n",
        "                ) in zip(\n",
        "                    raid_buffs_targeting_player[FIGHT_TIME],\n",
        "                    raid_buffs_targeting_player[ABILITY_NAME],\n",
        "                    raid_buffs_targeting_player[SOURCE_NAME],\n",
        "                    raid_buffs_targeting_player[SOURCE_SUBTYPE],\n",
        "                    raid_buffs_targeting_player[\n",
        "                        CARD_MATCHES_JOB\n",
        "                    ],\n",
        "                ):\n",
        "                    # Ignore if the source = the player, since these are accounted for in casting.\n",
        "                    if (\n",
        "                        source_name == player_name\n",
        "                        and ability_name not in APPLY_PREFERRED\n",
        "                    ):\n",
        "                        # However, ast cards are referenced weird and can also be played on\n",
        "                        # the self, so actually don't ignore those\n",
        "                        continue\n",
        "\n",
        "                    source_job_abbrev = TARGET_SUBTYPE_TO_JOB[\n",
        "                        source_subtype\n",
        "                    ]\n",
        "\n",
        "                    add_rb_ability(\n",
        "                        rb=rb,\n",
        "                        t=fight_time,\n",
        "                        skill_name=ability_name,\n",
        "                        job_class=source_job_abbrev,\n",
        "                        card_matches_job={\n",
        "                            1: False,\n",
        "                            2: True,\n",
        "                        }.get(card_matches_job),\n",
        "                    )\n",
        "\n",
        "            # Filter to ensure target of the prepull buff is the player\n",
        "            player_prepull_buffs = prepull_usages[\n",
        "                prepull_usages[TARGET_NAME] == player_name\n",
        "            ]\n",
        "\n",
        "            if not player_prepull_buffs.empty:\n",
        "                for (\n",
        "                    fight_time,\n",
        "                    ability_name,\n",
        "                    source_subtype,\n",
        "                ) in zip(\n",
        "                    player_prepull_buffs[FIGHT_TIME],\n",
        "                    player_prepull_buffs[ABILITY_NAME],\n",
        "                    player_prepull_buffs[SOURCE_SUBTYPE],\n",
        "                ):\n",
        "                    # Add earthly star estimation, fight time should match expected cast.\n",
        "                    if ability_name in {\"Earthly Dominance\", \"Giant Dominance\"}:\n",
        "                        LOG.debug(f\"Handling {ability_name} cast @ {fight_time} (Earthly Star Logic)\")\n",
        "                        earthly_star[\"Earthly Star\"] = earthly_star.get(\"Earthly Star\", []) + [fight_time]\n",
        "                        continue\n",
        "                    # We can allow the source to be the player since these\n",
        "                    # events are not in cast\n",
        "                    source_job_abbrev = TARGET_SUBTYPE_TO_JOB[\n",
        "                        source_subtype\n",
        "                    ]\n",
        "\n",
        "                    add_rb_ability(\n",
        "                        rb=rb,\n",
        "                        t=fight_time,\n",
        "                        skill_name=ability_name,\n",
        "                        job_class=source_job_abbrev,\n",
        "                    )\n",
        "\n",
        "            if earthly_star:\n",
        "                LOG.debug(f\"Processing AST Earthly star\")\n",
        "                LOG.debug(f\"{earthly_star}\")\n",
        "                # Process earthly star BS\n",
        "                stellar_detonation_casts = sorted(earthly_star.get(\"Stellar Detonation\"))\n",
        "                for earthly_star_cast in sorted(earthly_star[\"Earthly Star\"]):\n",
        "                    # I could optimize this, but since SD casts are capped at 20 max this is like, not an\n",
        "                    # issue to run a O(20) per cast\n",
        "                    # Get the most recent SD cast after the ES cast\n",
        "                    mr_sd_cast: Optional[float] = min([\n",
        "                        cast_time for cast_time in stellar_detonation_casts if cast_time > earthly_star_cast\n",
        "                    ], default=None)\n",
        "\n",
        "                    if mr_sd_cast is None or (mr_sd_cast - earthly_star_cast) > 20.0:\n",
        "                        LOG.debug(f\"Star placement @ {earthly_star_cast} was not detonated early \"\n",
        "                                  f\"given next Stellar Detonation cast {mr_sd_cast}\")\n",
        "                        add_rb_ability(\n",
        "                            rb=rb,\n",
        "                            t=earthly_star_cast,\n",
        "                            skill_name=\"Earthly Star\",\n",
        "                            job_class=job_abbrev,\n",
        "                        )\n",
        "                        continue\n",
        "\n",
        "                    # Can set the initial earthly star cast now\n",
        "                    add_rb_ability(\n",
        "                        rb=rb,\n",
        "                        t=earthly_star_cast,\n",
        "                        skill_name=\"Earthly Star Manual\",\n",
        "                        job_class=job_abbrev,\n",
        "                    )\n",
        "\n",
        "                    # Can set the SD cast too\n",
        "                    add_rb_ability(\n",
        "                        rb=rb,\n",
        "                        t=mr_sd_cast,\n",
        "                        skill_name=(\n",
        "                            \"Stellar Detonation\"\n",
        "                            if (mr_sd_cast - earthly_star_cast) > 10.0 else\n",
        "                            \"Stellar Detonation Small\"\n",
        "                        ),\n",
        "                        job_class=job_abbrev,\n",
        "                    )\n",
        "\n",
        "            output[player_name] = rb\n",
        "\n",
        "        self._rotations = output\n",
        "        return output\n",
        "\n",
        "\n",
        "# These will need to be updated as we test more jobs\n",
        "NAME_FIXES = {\n",
        "    \"Medicated\": \"Grade 8 Tincture\",\n",
        "    \"attack\": \"Auto\",\n",
        "    \"Shot\": \"Auto\",\n",
        "    \"Double Standard Finish\": \"Standard Finish\",\n",
        "    \"Quadruple Technical Finish\": \"Technical Finish\",\n",
        "    \"Vulnerability Up\": \"Mug\",\n",
        "    \"the Arrow\": \"Play\",\n",
        "    \"the Balance\": \"Play\",\n",
        "    \"the Ewer\": \"Play\",\n",
        "    \"the Spire\": \"Play\",\n",
        "    \"the Spear\": \"Play\",\n",
        "    \"the Bole\": \"Play\",\n",
        "    \"Left Eye\": \"Dragon Sight\",\n",
        "    \"Right Eye\": \"Dragon Sight\",\n",
        "    \"Earthly Star Manual\": \"Earthly Star\",\n",
        "    \"Stellar Detonation Small\": \"Stellar Detonation\",\n",
        "}\n",
        "\n",
        "SPECIAL_PARAMS = {\n",
        "    \"Medicated\": {\"job_class\": \"GENERAL\"},\n",
        "    \"Earthly Star Manual\": {\n",
        "        \"skill_modifier\": cr.SkillModifier(with_condition=\"Manual\")\n",
        "    },\n",
        "    \"Stellar Detonation Small\": {\n",
        "        \"skill_modifier\": cr.SkillModifier(with_condition=\"Small\")\n",
        "    },\n",
        "}\n",
        "\n",
        "BUFF_PARAMS = {\n",
        "    \"Standard Finish\": {\n",
        "        \"skill_modifier\": cr.SkillModifier(with_condition=\"Buff Only\")\n",
        "    },\n",
        "    \"Technical Finish\": {\n",
        "        \"skill_modifier\": cr.SkillModifier(with_condition=\"Buff Only\")\n",
        "    },\n",
        "    \"Vulnerability Up\": {\n",
        "        \"skill_modifier\": cr.SkillModifier(with_condition=\"Debuff Only\")\n",
        "    },\n",
        "    \"Left Eye\": {\"skill_modifier\": cr.SkillModifier(with_condition=\"Left Eye\")},\n",
        "}\n",
        "\n",
        "\n",
        "def add_rb_ability(\n",
        "    rb: cr.RotationBuilder,\n",
        "    t: float,\n",
        "    skill_name: str,\n",
        "    job_class: str,\n",
        "    buff_effect: bool = True,\n",
        "    card_matches_job: Optional[bool] = None,\n",
        "):\n",
        "    \"\"\"Helper to add skills, handling all the random edge-cases and name conversions.\"\"\"\n",
        "    corrected_name = NAME_FIXES.get(skill_name, skill_name)\n",
        "    if corrected_name is None:\n",
        "        # Skip\n",
        "        LOG.debug(f\"Ignoring {skill_name} @ {t} (Job: {job_class})\")\n",
        "        return\n",
        "    extra_args = SPECIAL_PARAMS.get(skill_name, {})\n",
        "    extra_buff_args = BUFF_PARAMS.get(skill_name, {}) if buff_effect else {}\n",
        "    if skill_name in CARDS:\n",
        "        if card_matches_job is None:\n",
        "            LOG.warning(\n",
        "                f\"Card usage {skill_name} @ {t} (Job: {job_class}) isn't flagged with effectiveness.\"\n",
        "            )\n",
        "        corrected_name = \"Card\" if card_matches_job else \"Card Small\"\n",
        "\n",
        "    try:\n",
        "        params = {\n",
        "            **dict(\n",
        "                t=t,\n",
        "                skill_name=corrected_name,\n",
        "                job_class=job_class,\n",
        "            ),\n",
        "            **extra_args,\n",
        "            **extra_buff_args,\n",
        "        }\n",
        "        rb.add(**params)\n",
        "        str_params = {\n",
        "            name: value if name != \"skill_modifier\" else str(value)\n",
        "            for name, value in params.items()\n",
        "        }\n",
        "        LOG.debug(f\"Set {str_params}\")\n",
        "    except KeyError as err:\n",
        "        if skill_name in NAME_FIXES:\n",
        "            LOG.debug(\n",
        "                f\"Unable to set {corrected_name} \"\n",
        "                f\"(from {skill_name}) @ {t} (Job: {job_class}): {err}\"\n",
        "            )\n",
        "        else:\n",
        "            LOG.debug(\n",
        "                f\"Unable to set {corrected_name} @ {t} (Job: {job_class}): {err}\"\n",
        "            )"
      ],
      "metadata": {
        "cellView": "form",
        "id": "viTlum5Nprkj"
      },
      "execution_count": 12,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "#@title FFLOGS API Call Functions\n",
        "\n",
        "def get_log_rotation_builders(\n",
        "    codes: List[str],\n",
        "    etro_sets: Dict[str, str],\n",
        "    fight_type: str = \"Kills\",\n",
        ") -> Dict[str, LogRotationBuilder]:\n",
        "    \"\"\"Gets a log via codes, then converts them all into LRB objects.\n",
        "\n",
        "    :param codes: List of all codes to gather sets for.\n",
        "    :param etro_sets: Mapping of (NAME|JOB) -> etro sets (URL|ID)\n",
        "    :param fight_type: Specific fight type to look for.\n",
        "        Use \"Encounters\" to include wipes\n",
        "    \"\"\"\n",
        "    LOG.info(f\"[Get] codes={codes}\")\n",
        "    client = FFLogClientV2()\n",
        "    rotation_builder_objects: Dict[str, LogRotationBuilder] = {}\n",
        "    for code in codes:\n",
        "        encounter_count: Dict = {}\n",
        "        for fight_name, events in (\n",
        "            client.stream_fight_events(\n",
        "                code=code,\n",
        "                fight_type=fight_type,\n",
        "            )\n",
        "        ):\n",
        "            encounter_count[fight_name] = encounter_count.get(fight_name, 0) + 1\n",
        "            rotation_name = f\"{code}-{fight_name}-{encounter_count[fight_name]}\"\n",
        "            rotation_builder_objects[rotation_name] = LogRotationBuilder(\n",
        "                etro_sets=etro_sets,\n",
        "                fight_events=events,\n",
        "            )\n",
        "\n",
        "    return rotation_builder_objects\n"
      ],
      "metadata": {
        "cellView": "form",
        "id": "EvJ_Gbi8qMfy"
      },
      "execution_count": 13,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "#@title Workspace! (Usage Example)\n",
        "\n",
        "# I picked some random 6.4 BIS sets, these are not assumed to be correct for a given log.\n",
        "DEFAULT_ETRO_SETS = {\n",
        "    \"GNB\": \"https://etro.gg/gearset/1dee5389-9906-4690-88b7-55419a342932\",\n",
        "    \"WAR\": \"https://etro.gg/gearset/1103c082-1c80-4bf3-bb56-83734971d5ea\",\n",
        "    \"PLD\": \"https://etro.gg/gearset/3a2d77ff-57e1-434d-b1da-72a7d4f44944\",\n",
        "    \"DRK\": \"https://etro.gg/gearset/dcd2eb34-7c43-4840-a17b-2eb790f19cf4\",\n",
        "    \"AST\": \"https://etro.gg/gearset/83845599-bb32-4539-b829-971501856d7e\",\n",
        "    \"SCH\": \"https://etro.gg/gearset/91cd8ea6-3ee9-4e3c-a38f-7341bd94fee1\",\n",
        "    \"WHM\": \"https://etro.gg/gearset/020093de-e69a-458f-8278-c3406eaa8eb7\",\n",
        "    \"SGE\": \"https://etro.gg/gearset/efc239cb-6371-4d1e-b645-8dd7600575b5\",\n",
        "    \"SAM\": \"https://etro.gg/gearset/d4b6bfc6-a82f-4732-8e55-7c13e094fc1d\",\n",
        "    \"MNK\": \"https://etro.gg/gearset/cc22e1aa-6d78-4ffe-bc53-e8466373b6bc\",\n",
        "    \"RPR\": \"https://etro.gg/gearset/dbedcbfc-21d0-420c-9c91-ed2e946adbe4\",\n",
        "    \"DRG\": \"https://etro.gg/gearset/8a907f52-75a4-4085-9deb-6a63ffa2abd8\",\n",
        "    \"NIN\": \"https://etro.gg/gearset/6556da3a-4514-439e-b4f4-07e0ccc85e93\",\n",
        "    \"BRD\": \"https://etro.gg/gearset/f2426d1e-2da8-4151-bf52-74ca67b5f4a2\",\n",
        "    \"DNC\": \"https://etro.gg/gearset/50746158-5be1-4972-82f4-84a577f4bcce\",\n",
        "    \"MCH\": \"https://etro.gg/gearset/0001cd0d-ee54-4b85-8bb6-8ed79e9e7745\",\n",
        "    \"RDM\": \"https://etro.gg/gearset/6d7a091d-52f5-49ec-9b2e-d7b1d4c45733\",\n",
        "    \"BLM\": \"https://etro.gg/gearset/46e5dde8-3d11-418f-948b-f6452fb0c7e3\",\n",
        "    \"SMN\": \"https://etro.gg/gearset/66f5ec54-c062-467f-811f-5e77a90c7aba\",\n",
        "}\n",
        "\n",
        "lrb_fights_QLMVrhT43RYK2PXN = get_log_rotation_builders(\n",
        "    codes=[\"QLMVrhT43RYK2PXN\"],\n",
        "    etro_sets={\n",
        "        **DEFAULT_ETRO_SETS,\n",
        "        # Add any name specific overrides if found.\n",
        "    },\n",
        ")\n",
        "\n",
        "print(list(lrb_fights_QLMVrhT43RYK2PXN.keys()))\n",
        "\n",
        "kill_6 = lrb_fights_QLMVrhT43RYK2PXN[\"QLMVrhT43RYK2PXN-Athena-6\"]\n",
        "\n",
        "# LETS GO?!\n",
        "output = kill_6.simulate_rotation()"
      ],
      "metadata": {
        "id": "tiDj-DTAuO6W",
        "cellView": "form"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "TODOS\n",
        "\n",
        "\n",
        "*   More random fixes on abilities (right modifiers, name fixes, combo and positionals?)\n",
        "*   Comparison to log damage and log rng\n",
        "*   Ideal gearset inference (hard)\n",
        "*   Suppress pd warnings\n",
        "*   Caching on some of the API calls\n",
        "\n"
      ],
      "metadata": {
        "id": "PWkHlnck2edz"
      }
    }
  ]
}