Spaces:
Sleeping
Sleeping
File size: 5,160 Bytes
6492314 e530ff5 6492314 e530ff5 6492314 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 | import os
import json
import time
import urllib.request
from src.preprocessing.preprocess_17lands import all_preprocessing_for_set, download_17lands
from src.preprocessing.embeddings import create_set_embedding_scryfall_language, combine_all_embeddings, normalize_embedding
from src.bert.embedd_text import embed_text_llama
from src.utils import utils
def combine_new_set(embedding_path, all_embedding_path):
new_set = utils.get_embedding_dict(embedding_path)
all_data = utils.get_embedding_dict(all_embedding_path) if os.path.exists(all_embedding_path) else {}
for k, v in new_set.items():
if k not in all_data:
all_data[k] = v
utils.dump_embedding_dict(all_data, all_embedding_path)
def _done_sets(out_folder):
path = os.path.join(out_folder, 'processed_sets.json')
if os.path.exists(path):
with open(path) as f:
return set(json.load(f))
return set()
def _mark_done(tag, out_folder):
path = os.path.join(out_folder, 'processed_sets.json')
done = _done_sets(out_folder)
done.add(tag)
with open(path, 'w') as f:
json.dump(list(done), f)
def _card_names_from_csv(raw_data_folder, tag):
"""Extract all card names seen in a set's draft CSV from the header row."""
import csv
set_dir = os.path.join(raw_data_folder, tag)
csv_files = [f for f in os.listdir(set_dir) if f.endswith('.csv')]
if not csv_files:
return None
with open(os.path.join(set_dir, csv_files[0]), newline='') as f:
header = next(csv.reader(f))
return [c.replace('pack_card_', '') for c in header if c.startswith('pack_card_')]
def download_gih_wr(tag, gih_folder):
"""Download per-card GIH (Games In Hand) win rate from 17lands for a set.
Saves filtered results to gih_folder/{tag}_gih.json. Skips on failure."""
os.makedirs(gih_folder, exist_ok=True)
out_path = os.path.join(gih_folder, f'{tag}_gih.json')
if os.path.exists(out_path):
return
url = f'https://www.17lands.com/card_ratings/data?expansion={tag}&event_type=PremierDraft&start_date=2020-01-01'
print(f'Downloading GIH WR for {tag}...')
try:
req = urllib.request.Request(url, headers={'User-Agent': 'MTG-IL-Research/1.0'})
with urllib.request.urlopen(req, timeout=30) as resp:
data = json.loads(resp.read().decode())
def parse_wr(val):
if val is None:
return None
if isinstance(val, str):
return float(val.rstrip('%')) / 100
return float(val)
filtered = []
for e in data:
# 17lands uses game_in_hand_win_rate as primary GIH metric; fall back to ever_drawn
wr = parse_wr(e.get('game_in_hand_win_rate'))
if wr is None:
wr = parse_wr(e.get('ever_drawn_win_rate'))
if wr is not None:
filtered.append({'name': e['name'], 'ever_drawn_win_rate': wr})
with open(out_path, 'w') as f:
json.dump(filtered, f)
print(f' Saved GIH WR for {len(filtered)} cards in {tag}')
time.sleep(2) # be respectful to 17lands servers
except Exception as e:
print(f' GIH download skipped for {tag}: {e}')
def ensure_set_ready(tag, config):
"""Ensure a set's CSV, LMDB, and card embeddings all exist, downloading/building as needed."""
raw_data_folder = config['raw_data_folder']
super_folder = config['super_folder']
out_folder = os.path.dirname(config['embedding_path'])
# GIH download is independent of the main pipeline — run unconditionally so
# already-processed sets still get their GIH data on first run.
gih_folder = os.path.join(out_folder, 'gih_wr')
download_gih_wr(tag, gih_folder)
train_lmdb = os.path.join(super_folder, tag, 'train.lmdb')
test_lmdb = os.path.join(super_folder, tag, 'test.lmdb')
lmdb_exists = os.path.exists(train_lmdb) and os.path.exists(test_lmdb)
if tag in _done_sets(out_folder) and lmdb_exists:
return
os.makedirs(super_folder, exist_ok=True)
set_embedding = os.path.join(out_folder, f'{tag}_embedding.pt')
raw_dir = os.path.join(raw_data_folder, tag)
if not os.path.isdir(raw_dir) or not any(f.endswith('.csv') for f in os.listdir(raw_dir)):
download_17lands(tag, raw_data_folder)
if not lmdb_exists:
all_preprocessing_for_set(tag, raw_data_folder, super_folder)
if not os.path.exists(set_embedding):
print(f'Creating card embeddings for {tag}')
card_keys = _card_names_from_csv(raw_data_folder, tag)
create_set_embedding_scryfall_language(
[tag], embedding_fn=embed_text_llama, out_folder=out_folder,
number_vector=True, card_keys=card_keys,
)
combine_new_set(set_embedding, config['embedding_path'])
_mark_done(tag, out_folder)
if __name__ == '__main__':
config = utils.load_config('src/configs/config.yaml')
for tag in config['sets_to_use']:
ensure_set_ready(tag, config)
|