utils/parser.py

import os
import time
from collections import defaultdict

import pandas as pd
from line_profiler import profile

from base.constant import FRAME_PER_SECOND
from schools import *
from utils.lua import parse_player, parse_damage

FRAME_TYPE, INDEX_TYPE = int, int
PLAYER_ID_TYPE, TARGET_ID_TYPE, PET_ID_TYPE = str, str, str
CASTER_ID_TYPE = PLAYER_ID_TYPE | PET_ID_TYPE
ENTITY_ID_TYPE, ENTITY_NAME_TYPE = CASTER_ID_TYPE | TARGET_ID_TYPE, str
SKILL_ID_TYPE, SKILL_LEVEL_TYPE, SKILL_STACK_TYPE = int, int, int
ACTUAL_DAMAGE, ACTUAL_CRITICAL_STRIKE = int, bool
SKILL_TYPE = Tuple[SKILL_ID_TYPE, SKILL_LEVEL_TYPE, SKILL_STACK_TYPE]
BUFF_ID_TYPE, BUFF_LEVEL_TYPE, BUFF_STACK_TYPE = int, int, int
BUFF_TYPE = Tuple[BUFF_ID_TYPE, BUFF_LEVEL_TYPE]

TAG_COLUMNS = ["frame", "time", "player_id", "caster_id", "target_id", "skill", "skill_name"]
CALCULATE_COLUMNS = ["skill_id", "skill_level", "skill_stack", "current_status", "target_status", "snapshot_index"]
COMPARE_COLUMNS = ["actual_critical_strike", "actual_damage"]
COLUMNS = TAG_COLUMNS + CALCULATE_COLUMNS + COMPARE_COLUMNS

LABEL_MAPPING = {
    2: "远程武器",
    3: "上衣",
    4: "帽子",
    5: "项链",
    6: "戒指1",
    7: "戒指2",
    8: "腰带",
    9: "腰坠",
    10: "下装",
    11: "鞋子",
    12: "护腕",
    0: "近战武器"
}
EMBED_MAPPING: Dict[tuple, int] = {(5, 24449 - i): 8 - i for i in range(8)}


class BaseParser:
    current_player: PLAYER_ID_TYPE
    current_caster: CASTER_ID_TYPE
    current_target: TARGET_ID_TYPE
    current_skill: SKILL_ID_TYPE

    current_frame: FRAME_TYPE
    current_index: INDEX_TYPE

    id2name: Dict[ENTITY_ID_TYPE, ENTITY_NAME_TYPE]
    name2id: Dict[ENTITY_NAME_TYPE, ENTITY_ID_TYPE]
    pet2employer: Dict[PET_ID_TYPE, PLAYER_ID_TYPE]

    records: Dict[PLAYER_ID_TYPE, List[dict] | pd.DataFrame]

    skill_display_names: Dict[tuple, str]
    buff_display_names: Dict[tuple, str]

    buffs: Dict[CASTER_ID_TYPE, Dict[BUFF_TYPE, BUFF_STACK_TYPE]]
    buff_intervals: Dict[CASTER_ID_TYPE, Dict[BUFF_TYPE, FRAME_TYPE]]
    target_buffs: Dict[TARGET_ID_TYPE, Dict[PLAYER_ID_TYPE, Dict[BUFF_TYPE, BUFF_STACK_TYPE]]]
    target_buff_intervals: Dict[TARGET_ID_TYPE, Dict[PLAYER_ID_TYPE, Dict[BUFF_TYPE, FRAME_TYPE]]]

    dot_stacks: Dict[TARGET_ID_TYPE, Dict[PLAYER_ID_TYPE, Dict[SKILL_ID_TYPE, int]]]
    dot_ticks: Dict[TARGET_ID_TYPE, Dict[PLAYER_ID_TYPE, Dict[SKILL_ID_TYPE, int]]]
    dot_buffs: Dict[TARGET_ID_TYPE, Dict[PLAYER_ID_TYPE, Dict[SKILL_ID_TYPE, INDEX_TYPE]]]

    next_pet_buffs: Dict[PLAYER_ID_TYPE, List[Dict[BUFF_TYPE, BUFF_STACK_TYPE]]]

    last_dot: Dict[TARGET_ID_TYPE, Dict[PLAYER_ID_TYPE, Dict[SKILL_ID_TYPE, INDEX_TYPE]]]
    next_dot: Dict[TARGET_ID_TYPE, Dict[PLAYER_ID_TYPE, Dict[SKILL_ID_TYPE, int]]]

    start_frame: FRAME_TYPE
    end_frame: FRAME_TYPE

    select_talents: Dict[PLAYER_ID_TYPE, List[int]]
    select_equipments: Dict[PLAYER_ID_TYPE, Dict[int, Dict[str, int | list]]]

    players: Dict[PLAYER_ID_TYPE, School]
    targets: Dict[PLAYER_ID_TYPE, List[TARGET_ID_TYPE]]

    @property
    def current_school(self):
        return self.players[self.current_player]

    @property
    def current_targets(self):
        return self.targets[self.current_player]

    @property
    def current_records(self):
        return self.records[self.current_player]

    @property
    def current_record(self):
        return self.current_records[-1]

    @property
    def current_buffs(self):
        return self.buffs[self.current_caster]

    @property
    def current_buff_intervals(self):
        return self.buff_intervals[self.current_player]

    @property
    def current_target_buffs(self):
        return self.target_buffs[self.current_target][self.current_player]

    @property
    def current_target_buff_intervals(self):
        return self.target_buff_intervals[self.current_target][self.current_player]

    @property
    def current_next_pet_buffs(self):
        return self.next_pet_buffs[self.current_player]

    @property
    def current_dot_buffs(self):
        return self.dot_buffs[self.current_target][self.current_player]

    @property
    def current_dot_stacks(self):
        return self.dot_stacks[self.current_target][self.current_player]

    @property
    def current_dot_ticks(self):
        return self.dot_ticks[self.current_target][self.current_player]

    @property
    def current_last_dot(self):
        return self.last_dot[self.current_target][self.current_player]

    @property
    def current_next_dot(self):
        return self.next_dot[self.current_target][self.current_player]

    def reset(self):
        self.current_frame = 0
        self.current_index = 0

        self.id2name = {}
        self.name2id = {}
        self.pet2employer = {}

        self.records = defaultdict(list)

        self.buff_display_names = {}
        self.buffs = defaultdict(dict)
        self.buff_intervals = defaultdict(dict)
        self.target_buffs = defaultdict(lambda: defaultdict(dict))
        self.target_buff_intervals = defaultdict(lambda: defaultdict(dict))

        self.dot_stacks = defaultdict(lambda: defaultdict(lambda: defaultdict(lambda: 1)))
        self.dot_ticks = defaultdict(lambda: defaultdict(lambda: defaultdict(lambda: 0)))

        self.next_pet_buffs = defaultdict(list)
        self.dot_buffs = defaultdict(lambda: defaultdict(dict))
        self.last_dot = defaultdict(lambda: defaultdict(dict))
        self.next_dot = defaultdict(lambda: defaultdict(dict))

        self.start_frame = 0

        self.select_talents = {}
        self.select_equipments = {}

        self.players = {}
        self.targets = defaultdict(list)

    def refresh_buff(self, buff_id, buff_level, buff_stack=1):
        buff = self.current_school.buffs[buff_id]
        buff_tuple = (buff_id, buff_level)
        stack = max(min(self.current_buffs.get(buff_tuple, 0) + buff_stack, buff.max_stack), 0)
        if stack:
            self.current_buffs[buff_tuple] = stack
            if buff.interval > 0:
                self.current_buff_intervals[buff_tuple] = self.current_frame + buff.interval + 1
        else:
            self.current_buffs.pop(buff_tuple, None)
            self.current_buff_intervals.pop(buff_tuple, None)

    def refresh_target_buff(self, buff_id, buff_level, buff_stack=1):
        buff = self.current_school.buffs[buff_id]
        buff_tuple = (buff_id, buff_level)
        stack = max(min(self.current_target_buffs.get(buff_tuple, 0) + buff_stack, buff.max_stack), 0)
        if stack:
            self.current_target_buffs[buff_tuple] = stack
            if buff.interval > 0:
                self.current_target_buff_intervals[buff_tuple] = self.current_frame + buff.interval + 1
        else:
            self.current_target_buffs.pop(buff_tuple, None)
            self.current_target_buff_intervals.pop(buff_tuple, None)

    def clear_buff(self, buff_id, buff_level):
        buff_tuple = (buff_id, buff_level)
        self.current_buffs.pop(buff_tuple, None)
        self.current_buff_intervals.pop(buff_tuple, None)

    def clear_target_buff(self, buff_id, buff_level):
        buff_tuple = (buff_id, buff_level)
        self.current_target_buffs.pop(buff_tuple, None)
        self.current_target_buff_intervals.pop(buff_tuple, None)

    def buff_timer(self):
        for caster_id, buffs in self.buff_intervals.items():
            pop_buffs = []
            for buff, end_frame in buffs.items():
                if end_frame < self.current_frame:
                    self.buffs[caster_id].pop(buff, None)
                    pop_buffs.append(buff)
            for pop_buff in pop_buffs:
                buffs.pop(pop_buff)
        for target_id in self.target_buff_intervals:
            for caster_id, buffs in self.target_buff_intervals[target_id].items():
                pop_buffs = []
                for buff, end_frame in buffs.items():
                    if end_frame < self.current_frame:
                        self.target_buffs[target_id][caster_id].pop(buff, None)
                        pop_buffs.append(buff)
                for pop_buff in pop_buffs:
                    buffs.pop(pop_buff)


class Parser(BaseParser):
    @property
    def duration(self):
        return round((self.end_frame - self.start_frame) / FRAME_PER_SECOND, 2)

    @staticmethod
    def parse_equipments(detail):
        select_equipments = {}
        for row in detail:
            if not (label := LABEL_MAPPING.get(row[0])):
                continue
            select_equipment = select_equipments[label] = {}
            select_equipment['equipment'] = row[2]
            select_equipment['strength_level'] = row[3]
            if isinstance(row[4], list):
                select_equipment['embed_levels'] = [EMBED_MAPPING.get(tuple(e), 0) for e in row[4]]
            else:
                select_equipment['embed_levels'] = []
            select_equipment['enchant'] = row[5]
        return select_equipments

    @staticmethod
    def parse_talents(detail):
        return [row[1] for row in detail]

    def parse_player(self, row):
        detail = row.strip("{}").split(",")
        player_id, school_id = detail[0], int(detail[3])
        if player_id in self.id2name or school_id not in SUPPORT_SCHOOL:
            return

        if detail := parse_player(row):
            player_name = detail[1]
            school = SUPPORT_SCHOOL[school_id]
            self.id2name[player_id] = player_name
            self.name2id[player_name] = player_id
            self.select_equipments[player_id] = self.parse_equipments(detail[5])
            self.select_talents[player_id] = self.parse_talents(detail[6])
            if any(talent not in school.talent_gains for talent in self.select_talents[player_id]):
                return
            self.players[player_id] = school

    def parse_npc(self, row):
        detail = row.strip("{}").split(",")
        npc_id, npc_name, employer_id = detail[0], detail[1].strip('"'), detail[3]
        if npc_id in self.id2name or not npc_name:
            return

        self.id2name[npc_id] = npc_name
        self.name2id[npc_name] = npc_id
        if employer_id in self.players:
            self.pet2employer[npc_id] = employer_id

    def parse_pet(self, row):
        pet_id = row.strip("{}")
        if player_id := self.pet2employer.get(pet_id):
            if self.next_pet_buffs[player_id]:
                pet_buff_stacks = self.next_pet_buffs[player_id].pop()
            else:
                pet_buff_stacks = {}
            self.buffs[pet_id] = {**self.buffs[player_id].copy(), **pet_buff_stacks}

    def parse_buff(self, row):
        detail = row.strip("{}").split(",")
        caster_id = detail[0]
        if caster_id in self.pet2employer:
            player_id = self.pet2employer[caster_id]
            if caster_id in self.buffs:
                buffs = self.buffs[caster_id]
            elif self.next_pet_buffs[player_id]:
                buffs = self.next_pet_buffs[player_id][0]
            else:
                buffs = {}
                self.next_pet_buffs[player_id].append(buffs)
        else:
            player_id = caster_id
            buffs = self.buffs[player_id]

        if player_id not in self.players:
            return

        buff_id, buff_stack, buff_level = int(detail[4]), int(detail[5]), int(detail[8])
        if buff_id not in self.players[player_id].buffs:
            return

        if buff_stack:
            buffs[(buff_id, buff_level)] = buff_stack
        else:
            buffs.pop((buff_id, buff_level), None)

    @profile
    def parse_skill(self, row):
        detail = row.strip("{}").split(",")
        caster_id, target_id = detail[0], detail[1]
        if caster_id in self.pet2employer:
            player_id = self.pet2employer[caster_id]
        else:
            player_id = caster_id

        if player_id not in self.players:
            return

        react, skill_id, skill_level = int(detail[2]), int(detail[4]), int(detail[5])
        actual_critical_strike, actual_damage = detail[6] == "true", parse_damage(row)
        if react or skill_id not in self.players[player_id].skills:
            return

        if not self.start_frame:
            self.start_frame = self.current_frame

        self.current_player = player_id
        self.current_caster = caster_id
        if target_id in self.id2name and target_id not in self.current_targets:
            self.current_targets.append(target_id)
        self.current_target = target_id
        self.current_skill = skill_id

        skill = self.players[player_id].skills[skill_id]
        skill.skill_level = skill_level
        self.current_records.append({
            "frame": self.current_frame, **self.current_buffs, **self.current_target_buffs,
            "player_id": self.current_player, "caster_id": self.current_caster, "target_id": self.current_target,
            "skill_id": skill_id, "skill_level": skill_level, "skill_stack": 1, "skill_name": skill.skill_name,
            "actual_critical_strike": actual_critical_strike, "actual_damage": actual_damage,
           })
        self.set_status()
        skill.parse(self)
        self.current_index += 1

    @profile
    def set_status(self):
        current_status = []
        for (buff_id, buff_level), buff_stack in self.current_buffs.items():
            buff = self.current_school.buffs[buff_id]
            if buff.gain_attributes:
                current_status.append((buff_id, buff_level, buff_stack))
            elif buff.gain_skills and self.current_skill in buff.gain_skills:
                current_status.append((buff_id, buff_level, buff_stack))
        self.current_record['current_status'] = tuple(sorted(current_status))

        target_status = []
        for (buff_id, buff_level), buff_stack in self.current_target_buffs.items():
            buff = self.current_school.buffs[buff_id]
            if buff.gain_attributes:
                target_status.append((buff_id, buff_level, buff_stack))
            elif buff.gain_skills and self.current_skill in buff.gain_skills:
                target_status.append((buff_id, buff_level, buff_stack))
        self.current_record['target_status'] = tuple(sorted(target_status))

    def convert_records(self):
        for player_id, records in self.records.items():
            records = pd.DataFrame(records)
            records.snapshot_index = records.snapshot_index.fillna(-1)
            records = records.fillna(0)
            records['time'] = (records.frame - self.start_frame) / FRAME_PER_SECOND
            skills = [""] * len(records)
            for skill_tuple, indices in records.groupby(["skill_id", "skill_level", "skill_stack"]).indices.items():
                skill_id, skill_level, skill_stack = skill_tuple
                skill = self.players[player_id].skills[skill_id]
                skill.skill_level, skill.skill_stack = skill_level, skill_stack
                for index in indices:
                    skills[index] = skill.display_name
            records["skill"] = skills
            buffs = {}
            for buff_tuple in [column for column in records.columns if isinstance(column, tuple)]:
                buff_id, buff_level = buff_tuple
                buff = self.players[player_id].buffs[buff_id]
                buff.buff_level = buff_level
                buffs[buff_tuple] = buff.display_name
            records = records.rename(columns=buffs)
            columns = [c for c in COLUMNS if c in records.columns] + [c for c in records.columns if c not in COLUMNS]
            records = records[columns]
            self.records[player_id] = records

    @profile
    def __call__(self, file_name):
        self.file_name = os.path.basename(file_name)
        self.reset()
        lines = open(file_name, encoding="gbk").readlines()
        rows = []
        for line in lines:
            row = line.split("\t")
            rows.append(row)
            if row[4] == "4":
                self.parse_player(row[-1])
            elif row[4] == "8":
                self.parse_npc(row[-1])

        for player_id, school in self.players.items():
            school.prepare(self, player_id)
            for talent_id in self.select_talents[player_id]:
                school.talent_gains[talent_id].add_skills(school.skills)

        for row in rows:
            if (current_frame := int(row[1])) != self.current_frame:
                self.current_frame = current_frame
                self.buff_timer()

            if row[4] == "6":
                self.parse_pet(row[-1].strip())
            elif row[4] == "13":
                self.parse_buff(row[-1].strip())
            elif row[4] == "21":
                self.parse_skill(row[-1].strip())

        self.end_frame = self.current_frame

        for player_id, school in self.players.items():
            for talent_id in self.select_talents[player_id]:
                school.talent_gains[talent_id].sub_skills(school.skills)

        self.convert_records()