Spaces:
Sleeping
Sleeping
| 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) | |