목적

기존에 시도하다가 멈춘 PUBG API 프로젝트가 있는데 이를 다시 제대로 해보기 위해 일단 API에서 제공되는 데이터들을 상세히 분석해서 아이디어를 도출하기 위함.

이번 포스팅 목적 : 누구나 따라할 수 있도록 재밌는 그래프 하나를 도출할 것(과정을 잘 이해하면 여러 아이디어가 떠오를 수도 있다.)

환경설정

• MAC OS
• python 3.10 가상환경 구성

(아나콘다 설치되어있다고 가정)터미널 에서
conda create --name my_jupyter_env python=3.10
conda activate my_jupyter_env
conda install -y jupyter pandas matplotlib seaborn numpy

pip install dotenv

• 주피터 노트북 실행

터미널 에서
jupyter notebook

https://github.com/pubg/api-assets/blob/master/dictionaries/telemetry/mapName.json
https://github.com/pubg/api-assets/blob/master/dictionaries/telemetry/item/itemId.json
https://github.com/pubg/api-assets/blob/master/dictionaries/telemetry/damageCauserName.json
에서 json 다운로드 받아서 ipynb 생성한 파일과 같은 루트(프로젝트 루트)에 두기

프로젝트 루트에 .env 만들고 PUBG API key 받아서 .env에 다음과 같이 작성
(당신의 API KEY 입력 부분에 PUBG API key 쓰세요.)

API_KEY=당신의 API KEY 입력

.env API KEY 가져오기

import os
from dotenv import load_dotenv

# .env 파일 로드
load_dotenv()

# 환경 변수 가져오기
API_KEY = os.getenv("API_KEY")

print(API_KEY)  

시즌 ID 얻기

import requests

# API URL
url = "https://api.pubg.com/shards/steam/seasons"

# 요청 헤더 설정
headers = {
    "Authorization": "Bearer " + API_KEY,
    "Accept": "application/vnd.api+json"
}

# GET 요청 보내기
response = requests.get(url, headers=headers)

# 응답 상태 코드 확인
if response.status_code == 200:
    # JSON 응답 데이터 출력
    print(response.json())
else:
    print(f"Error {response.status_code}: {response.text}")


season_data = response.json()

for i in season_data.get('data'):
    isCurrentSeason = i.get('attributes').get('isCurrentSeason')
    if isCurrentSeason:
        SEASON_ID = i.get('id')
        print(SEASON_ID)

SEASON_ID에 현재 시즌에 대한 id 값이 들어온다.

플레이어 이름 설정

PLAYER_NAME = 'your_steam_player_name'

플레이어 최근 매치 가져오기

url_get_player = 'https://api.pubg.com/shards/steam/players?filter[playerNames]=' + PLAYER_NAME

# GET 요청 보내기
response_get_player = requests.get(url_get_player, headers=headers)

# 응답 상태 코드 확인
if response_get_player.status_code == 200:
    # JSON 응답 데이터 출력
    print(response_get_player.json())
else:
    print(f"Error {response_get_player.status_code}: {response_get_player.text}")
    
player_data = response_get_player.json()

my_matchs = player_data.get('data')[0].get('relationships').get('matches').get('data')
latest_match_item = my_matchs[0]
latest_match_item_id = my_matchs[0].get('id')

print("MY MATCH ID = " + latest_match_item_id)

매치 조회

url_get_match_data = 'https://api.pubg.com/shards/steam/matches/' + latest_match_item_id

# GET 요청 보내기
response_get_match_data = requests.get(url_get_match_data, headers=headers)

# 응답 상태 코드 확인
if response_get_match_data.status_code == 200:
    # JSON 응답 데이터 출력
    print(response_get_match_data.json())
else:
    print(f"Error {response_get_match_data.status_code}: {response_get_match_data.text}")

이 json결과(match 세부정보 제공)보다 더 세부적인 데이터(매치 진행동안 시간 및 타입별 유저들의 행동 데이터)를 얻을 수 있다.

이 데이터에서 asset부분에서 URL을 찾는다.

match_data = response_get_match_data.json()
asset_id = match_data.get('data').get('relationships').get('assets').get('data')[0].get('id')
asset_url = next((item["attributes"]["URL"] for item in match_data["included"] if item["id"] == asset_id), None)
print(asset_url)

이 asset_url로 하여금 PUBG가 관리하는 CDN의 json을 가져올 수 있다. 이 json은 매치진행동안 플레이어들의 모든 행동데이터를 보여주는 만큼 굉장히 용량이 크다.

지금까지의 API 이용을 정리하면

Season ID 확보 -> season id와 플레이어 이름으로 플레이어 최근 매치 ID 확보 -> 매치ID로 매치 상세정보 확보 -> 매치 상세정보의 asset URL로 telemetry_json 확보

Telemetry_data 분석

매치속 시계열 행동 데이터(Telemetry_data)에는 매우 방대한 정보가 들어있기에 필요한 소스만 찾아서 나열해보기로 한다.

무엇이 필요할지는 펍지 API docs를 확인해서 관심사를 결정했다.

펍지 telemetry api docs

우선 내가 관심있는 것은 딜량에 관한 지표이기 때문에

플레이어 공격 데이터인 LogPlayerAttack과 내 공격에 데미지를 받았는지 알기 위해 LogPlayerTakeDamage를 활용하기로 한다.

매치 메타 데이터 확보

우선 telemetry_data가 아닌 그 이전 match_data에서 다음과 같은 정보를 뽑아내도록 하자.

## 매치 정보
match_meta_data = match_data.get('data').get('attributes')
match_meta_data

match_gameMode = match_meta_data.get('gameMode')

# mapName 매핑 데이터 로드
with open("mapName.json", "r", encoding="utf-8") as f:
    map_name_mapping = json.load(f)

# mapName 매핑 적용
map_name_key = match_meta_data.get("mapName", "")
mapped_map_name = map_name_mapping.get(map_name_key, "알 수 없는 맵")

# 결과 출력
print(f"맵 이름: {mapped_map_name}")

간단하게 해당 매치의 맵 이름과 게임 모드(squad, solo, duo)를 변수화 했다.

dataframe으로 LogPlayerAttack 확인해보기

import pandas as pd
import requests
from IPython.display import display

# 파일 URL에서 데이터 가져오기
response_asset_url = requests.get(asset_url)
telemetry_data = response_asset_url.json()

# LogPlayerAttack 이벤트만 필터링
log_player_attack_events = [event for event in telemetry_data if event.get("_T") == "LogPlayerAttack"]

# DataFrame 생성
df = pd.DataFrame(log_player_attack_events)

# attacker와 weapon 정보를 개별 컬럼으로 변환
attacker_df = df["attacker"].apply(pd.Series)
weapon_df = df["weapon"].apply(pd.Series)

# 필요하면 attachedItems 리스트도 풀어서 개별 컬럼으로 만들 수 있음
if "attachedItems" in weapon_df.columns:
    attached_items_df = weapon_df["attachedItems"].apply(lambda x: ", ".join(x) if isinstance(x, list) else "")
    weapon_df["attachedItems"] = attached_items_df

# 기존 데이터프레임에 병합
df = pd.concat([df, attacker_df.add_prefix("attacker_"), weapon_df.add_prefix("weapon_")], axis=1)

# 기존 컬럼 삭제 (딕셔너리 전체가 들어있는 컬럼)
df.drop(columns=["attacker", "weapon"], inplace=True)

# DataFrame 출력
display(df.head())  # 상위 5개 행만 출력

fireWeaponStackCount는 특정한 총으로 발사한 누적 횟수이다. 특정한 총이란, 만약 총A로 100발을 발사한다면 100번째 나의 데이터의 fireWeaponStackCount는 100이 될 것이다. 만약 그 후 총B로 쏘게 된다면 0부터 다시 카운트된다. 그 후 다시 총A로 쏜다면 100부터 다시 카운팅 된다.

데이터 요약 관찰

df_desc = df.describe(include="all")
display(df_desc)

_D는 시간을 나타낸다.(unique를 보니 동시 시간에 발생한 이벤트들이 있는 것으로 보인다.)
_T는 이벤트의 타입이다. LogPlayerAttack으로만 뽑았기에 unique 값이 1인 것을 확인할 수 있다.
attacker_name이 53으로 보이는데 게임에 입장한 수가 90명이 넘을테니 공격도 못해보고 죽는 플레이어가 많다는 것을 알 수 있다.

describe를 통해 데이터에 대한 여러 통찰을 얻을 수 있다.

fireWeaponStackCount

이 데이터가 도저히 이해되지 않았어서(위에서 이미 설명했지만 의미를 파악한 과정을 서술한다.) 데이터를 찍어보았다.

import matplotlib.pyplot as plt

# fireWeaponStackCount의 유니크한 값과 개수 확인
unique_values_counts = df["fireWeaponStackCount"].value_counts()

# 데이터 개수가 너무 많으면 상위 N개만 선택
top_n = 50  # 예: 상위 50개만 표시
top_unique_values_counts = unique_values_counts.head(top_n)

# 그래프 그리기
plt.figure(figsize=(12, 6))
plt.bar(top_unique_values_counts.index, top_unique_values_counts.values, width=0.5)
plt.xlabel("fireWeaponStackCount (Top 50 Unique Values)")
plt.ylabel("Count")
plt.title("Top 50 Distribution of Unique fireWeaponStackCount Values")
plt.yscale("log")  # 데이터 분포가 클 경우 로그 스케일 적용

# X축 값 회전하여 보기 쉽게 조정
plt.xticks(rotation=90)

# 그래프 표시
plt.show()

발사 이벤트에 관련한 것이라고 이름에서 유추할 수 있었고 낮은 숫자일 수록 많다는 것을 보고 누적값이겠거니 예상할 수 있었다.

특정 플레이어 공격 이벤트 그래프화

import matplotlib.pyplot as plt

# 특정 플레이어의 데이터
df_filtered = df[df["attacker_name"] == "meang9_-"]

# 무기별 발사 횟수를 시간 순서대로 정렬
df_filtered = df_filtered.sort_values(by="_D")

# 무기별 발사 횟수 흐름을 그래프로 시각화
plt.figure(figsize=(12, 6))

# 무기별로 데이터 분리
for weapon in df_filtered["weapon_itemId"].unique():
    weapon_df = df_filtered[df_filtered["weapon_itemId"] == weapon]
    plt.plot(weapon_df.index, weapon_df["fireWeaponStackCount"], marker='o', linestyle='-', label=weapon)

# 그래프 설정
plt.xlabel("Sequence (Event Order)")
plt.ylabel("Cumulative Fire Count")
plt.title("Weapon Fire Count Progression for meang9_-")
plt.legend(title="Weapon Item ID", loc="upper left")
plt.grid(True)

# 그래프 표시
plt.show()

전체 이벤트 2800여개중 특정 플레이어의 이벤트만 총기별로 찍어본 것이다.
이걸 보고 모든 공격 행위(총 한발 발사와 같은)가 독립적인 행으로 존재함을 알 수 있었다.

특정 플레이어 공격 이벤트와 해당 공격으로 인한 피해 이벤트 함께 그래프화

피해 데미지 데이터 얻기

log_player_take_damage_events = [event for event in telemetry_data if event.get("_T") == "LogPlayerTakeDamage"]

# DataFrame 생성
df_player_damage = pd.DataFrame(log_player_take_damage_events)

filtered_damage_types = [
    "Damage_Gun",
    "Damage_Explosion_Grenade",
    "Damage_Explosion_PanzerFaustWarhead",
    "Damage_Explosion_Mortar"
]

df_filtered_damage = df_player_damage[df_player_damage["damageTypeCategory"].isin(filtered_damage_types)]

무기로 공격당한 데이터만 필터링 했을때 attack_id가 -1이 아닌 값으로 나타났다. 그래서 피해 데이터에 나타난 attack_id 전체가 이전에 미리 생성한 공격 이벤트 dataFrame에 모두 존재하는지 확인해보았다.

# df_filtered_damage의 attackId 값들이 df의 attackId에 모두 존재하는지 확인
all_exist = df_filtered_damage['attackId'].isin(df['attackId']).all()

# 결과 출력
all_exist

# 결과 np.True_ 나타나면 실험 성공

모두 존재함을 확인했다.

데미지 막대 그래프 그려보기

# df_filtered에서 공격 ID(attackId)와 일치하는 피해 데이터 필터링
df_damage_filtered = df_filtered_damage[df_filtered_damage["attackId"].isin(df_filtered["attackId"])]

# 시간순 정렬
df_damage_filtered = df_damage_filtered.sort_values(by="_D")

# 그래프 그리기
plt.figure(figsize=(12, 6))
plt.bar(df_damage_filtered["_D"], df_damage_filtered["damage"], color="red", alpha=0.7)

# 그래프 설정
plt.xlabel("Time (Event Order)")
plt.ylabel("Damage Dealt")
plt.title("Damage Dealt Over Time by meang9_-")
plt.xticks(rotation=45)
plt.grid(axis="y", linestyle="--", alpha=0.7)

# 그래프 표시
plt.show()

내가 조작시킨 캐릭터는 6번의 데미지 이벤트를 일으켰다. 이것을 위의 그래프 공격시도 그래프와 합쳐보도록 하자. x축은 2800여개의 시간 데이터를 기반으로 축 범위를 설정하고 y축은 공격시도count, 데미지 두 가지를 섞도록 하자.

그래프 합치기 최종

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd

#  데이터프레임 복사 및 UTC → KST 변환
df_filtered = df[df["attacker_name"] == "meang9_-"].copy()
df_filtered_damage = df_player_damage[df_player_damage["attackId"].isin(df_filtered["attackId"])].copy()

df_filtered["_D"] = pd.to_datetime(df_filtered["_D"]) + pd.Timedelta(hours=9)
df_filtered_damage["_D"] = pd.to_datetime(df_filtered_damage["_D"]) + pd.Timedelta(hours=9)

#  최초 시간 기준 상대적 시간(분) 계산
start_time = df_filtered["_D"].min()
df_filtered["relative_time"] = (df_filtered["_D"] - start_time).dt.total_seconds() / 60
df_filtered_damage["relative_time"] = (df_filtered_damage["_D"] - start_time).dt.total_seconds() / 60

#  매치 진행시간을 실제 데이터 기반으로 계산 (UTC 변환 후 최소~최대 시간 차이)
match_duration_min = (df_filtered["_D"].max() - df_filtered["_D"].min()).total_seconds() / 60

#  그래프 제목용 데이터 추출
match_date = start_time.strftime("%Y-%m-%d")
match_time = start_time.strftime("%p %I시 %M분")  # AM/PM 변환
player_name = df_filtered["attacker_name"].iloc[0]  # 플레이어 이름

#  색상 매핑 (무기 및 데미지 원인 통합)
unique_weapons = pd.concat([df_filtered["unified_weapon_name"], df_filtered_damage["unified_weapon_name"]]).dropna().unique()
colors = plt.colormaps['tab10']
color_mapping = {weapon: colors(i % 10) for i, weapon in enumerate(unique_weapons)}

#  그래프 생성
fig, ax1 = plt.subplots(figsize=(12, 6))

#  첫 번째 y축 (무기 발사 횟수 - 선 그래프)
ax1.set_xlabel("시간 (분)")
ax1.set_ylabel("누적 무기 발사 카운트", color='tab:blue')

for weapon in df_filtered["unified_weapon_name"].dropna().unique():
    weapon_df = df_filtered[df_filtered["unified_weapon_name"] == weapon]
    ax1.plot(weapon_df["relative_time"], weapon_df["fireWeaponStackCount"], 
             marker='o', linestyle='-', color=color_mapping[weapon], label=weapon)

ax1.tick_params(axis='y', labelcolor='tab:blue')
ax1.grid(True)

#  두 번째 y축 (피해량 - 바 그래프)
ax2 = ax1.twinx()
ax2.set_ylabel("가한 데미지 (Damage)", color='tab:red')

for weapon in df_filtered_damage["unified_weapon_name"].dropna().unique():
    weapon_df = df_filtered_damage[df_filtered_damage["unified_weapon_name"] == weapon]
    ax2.bar(weapon_df["relative_time"], weapon_df["damage"], alpha=0.4,
            color=color_mapping[weapon], label=weapon, width=0.05)

ax2.tick_params(axis='y', labelcolor='tab:red')

#  x축 조정 (0, 2, 4, 6, ... 형태 + "분" 단위 추가)
xticks = np.arange(0, df_filtered["relative_time"].max() + 2, 2)
ax1.set_xticks(xticks)
ax1.set_xticklabels([f"{int(x)}분" for x in xticks], fontsize=10, rotation=45)

#  제목 추가 (한국 시간 적용)
plt.title(f"{match_date} : {match_time} {player_name} 매치 딜량 분포\n"
          f"맵: {mapped_map_name} | 모드: {match_gameMode} | 진행 시간: {match_duration_min:.1f}분")

#  범례 설정
ax1.legend(title="발사한 무기", loc="upper left")
ax2.legend(title="데미지를 준 무기", loc="upper right")

# 그래프 표시
plt.show()

다음과 같은 결과가 나온다.

희미하게 보이는 막대그래프가 공격 시도가 유효한(데미지를 준) 공격이었을 때 데미지를 막대그래프로 표현한 것이다.

무기마다 그래프 색이 다른데, 피해 데이터, 공격 데이터 모두 총기정보에 대한 칼럼이 있어 이를 api가 제공하는 json파일을 참고하여 자동 매핑시켜서 색상을 통일했다.

이 분석 스크립트에서 수행한 그래프뽑기 기능을 코드를 리펙토링해서 디스코드 봇 같은 것으로 제공할 예정이다.