grid_gen.py

import requests
import json
import math
import os
import logging
from pathlib import Path
from typing import List, Dict, Any, Optional
from scipy.stats import norm
from dotenv import load_dotenv

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# Load environment variables from .env file
load_dotenv()

# Configuration constants
STEAM_ID = os.getenv('STEAM_ID')
STRATZ_TOKEN = os.getenv('STRATZ_TOKEN')
STEAM_USERDATA_PATH = os.getenv('STEAM_USERDATA_PATH', str(
    Path.home() / 'Library/Application Support/Steam/userdata'))
DOTA_CFG_PATH = f"{STEAM_USERDATA_PATH}/{STEAM_ID}/570/remote/cfg"

# Layout configuration constants
LAYOUT_CONFIG = {
    "category_width": 455,
    "category_height": 75,
    "category_spacing": 20,
    "all_heroes_width": 600,
    "all_heroes_height": 600,
    "all_heroes_x_offset": 500,
}

PATCH_VERSIONS = {
    "7.37e": 178,
    "7.38": 179,
    "7.39c": 180,
}
CURRENT_PATCH = os.getenv('CURRENT_PATCH', list(PATCH_VERSIONS.keys())[-1])  # Default to last patch version if not specified

# API Configuration
STRATZ_API_URL = "https://api.stratz.com/graphql"
GAME_MODES = [1, 22]  # All Pick and Ranked All Pick
LOBBY_TYPES = [7]  # Ranked
MAX_HEROES_PER_POSITION = 7
MIN_WIN_RATE = 0.5
CONFIDENCE_LEVEL = 0.95

# Dota 2 Hero IDs (all heroes)
ALL_HERO_IDS = [
    1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20,
    21, 22, 23, 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, 119, 120, 121, 123, 126, 128, 129, 131, 135, 136,
    137, 138, 145
]

# Position mapping
POSITIONS = ["POSITION_1", "POSITION_2",
             "POSITION_3", "POSITION_4", "POSITION_5"]
POSITION_NAMES = {
    "POSITION_1": "Carry",
    "POSITION_2": "Mid",
    "POSITION_3": "Offlane",
    "POSITION_4": "Support",
    "POSITION_5": "Hard Support"
}


def validate_steam_id(steam_id: str) -> bool:
    """
    Validate Steam ID format.
    
    Args:
        steam_id: Steam ID to validate.
        
    Returns:
        True if valid, False otherwise.
    """
    return steam_id is not None and steam_id.isdigit() and len(steam_id) > 0


def wilson_score(positive: int, total: int, confidence: float = CONFIDENCE_LEVEL) -> float:
    """
    Calculate the Wilson score lower bound for a proportion with a given confidence level.

    Args:
        positive: Number of positive outcomes (e.g., wins).
        total: Total number of trials (e.g., matches).
        confidence: Confidence level (default is 0.95 for 95% confidence).

    Returns:
        Wilson score lower bound.
    """
    if total == 0:
        return 0.0

    z = norm.ppf(1 - (1 - confidence) / 2)
    p_hat = positive / total
    denominator = 1 + z**2 / total
    centre_adj = p_hat + z**2 / (2 * total)
    adj_se = z * math.sqrt((p_hat * (1 - p_hat) + z**2 / (4 * total)) / total)

    lower_bound = (centre_adj - adj_se) / denominator
    return max(0, lower_bound)


def validate_configuration() -> None:
    """Validate that required environment variables are set."""
    if not STRATZ_TOKEN:
        raise ValueError("STRATZ_TOKEN environment variable is required")

    if not validate_steam_id(STEAM_ID):
        raise ValueError("STEAM_ID environment variable is required and must be a valid Steam ID")

    if not os.path.exists(STEAM_USERDATA_PATH):
        raise ValueError(
            f"Steam userdata path does not exist: {STEAM_USERDATA_PATH}")


def create_hero_grid_config(filtered_heroes: List[Dict[str, Any]], config_name: str) -> Dict[str, Any]:
    """
    Create hero grid configuration from filtered heroes data.

    Args:
        filtered_heroes: List of hero data with position and rating information.
        config_name: Name of the configuration.

    Returns:
        Hero grid configuration dictionary.
    """
    # Group heroes by position
    categories = {position: [] for position in POSITIONS}
    chosen_hero_ids = []

    for hero in filtered_heroes:
        position = hero.get("position")
        if position in categories and len(categories[position]) < MAX_HEROES_PER_POSITION:
            categories[position].append(hero["heroId"])
            chosen_hero_ids.append(hero["heroId"])

    # Initialize hero grid structure
    hero_grid = {
        "config_name": config_name,
        "categories": []
    }

    # Define the layout structure using constants
    layout_config = [
        {
            "category_name": POSITION_NAMES["POSITION_1"],
            "x_position": 0,
            "y_position": 0,
            "width": LAYOUT_CONFIG["category_width"],
            "height": LAYOUT_CONFIG["category_height"],
            "position": "POSITION_1"
        },
        {
            "category_name": POSITION_NAMES["POSITION_2"],
            "x_position": 0,
            "y_position": LAYOUT_CONFIG["category_height"] + LAYOUT_CONFIG["category_spacing"],
            "width": LAYOUT_CONFIG["category_width"],
            "height": LAYOUT_CONFIG["category_height"],
            "position": "POSITION_2"
        },
        {
            "category_name": POSITION_NAMES["POSITION_3"],
            "x_position": 0,
            "y_position": 2 * (LAYOUT_CONFIG["category_height"] + LAYOUT_CONFIG["category_spacing"]),
            "width": LAYOUT_CONFIG["category_width"],
            "height": LAYOUT_CONFIG["category_height"],
            "position": "POSITION_3"
        },
        {
            "category_name": POSITION_NAMES["POSITION_4"],
            "x_position": 0,
            "y_position": 3 * (LAYOUT_CONFIG["category_height"] + LAYOUT_CONFIG["category_spacing"]),
            "width": LAYOUT_CONFIG["category_width"],
            "height": LAYOUT_CONFIG["category_height"],
            "position": "POSITION_4"
        },
        {
            "category_name": POSITION_NAMES["POSITION_5"],
            "x_position": 0,
            "y_position": 4 * (LAYOUT_CONFIG["category_height"] + LAYOUT_CONFIG["category_spacing"]),
            "width": LAYOUT_CONFIG["category_width"],
            "height": LAYOUT_CONFIG["category_height"],
            "position": "POSITION_5"
        },
    ]

    # Populate categories using the layout configuration
    for layout_item in layout_config:
        position = layout_item["position"]
        cat_info = {
            "category_name": layout_item["category_name"],
            "x_position": layout_item["x_position"],
            "y_position": layout_item["y_position"],
            "width": layout_item["width"],
            "height": layout_item["height"],
            "hero_ids": categories[position]
        }
        hero_grid["categories"].append(cat_info)

    # Add remaining heroes to "All Heroes" category
    rest_hero_ids = [
        hero_id for hero_id in ALL_HERO_IDS if hero_id not in chosen_hero_ids]
    hero_grid["categories"].append({
        "category_name": "All Heroes",
        "x_position": LAYOUT_CONFIG["all_heroes_x_offset"],
        "y_position": 0,
        "width": LAYOUT_CONFIG["all_heroes_width"],
        "height": LAYOUT_CONFIG["all_heroes_height"],
        "hero_ids": rest_hero_ids
    })

    return {
        "version": 3,
        "configs": [hero_grid]
    }


def validate_api_response(data: Dict[str, Any]) -> bool:
    """
    Validate API response structure.
    
    Args:
        data: API response data.
        
    Returns:
        True if valid, False otherwise.
    """
    try:
        return (data is not None and 
                "data" in data and 
                data["data"] is not None and
                "player" in data["data"] and
                data["data"]["player"] is not None and
                "heroesGroupBy" in data["data"]["player"])
    except (KeyError, TypeError):
        return False


def fetch_hero_data(steam_id: str, position: str, min_game_version_id: Optional[int] = None) -> List[Dict[str, Any]]:
    """
    Fetch hero data for a specific position from Stratz API.

    Args:
        steam_id: Steam account ID.
        position: Position to fetch data for.
        min_game_version_id: Optional minimum game version ID filter.

    Returns:
        List of hero data for the position.

    Raises:
        requests.RequestException: If API request fails.
        json.JSONDecodeError: If response is not valid JSON.
        ValueError: If API response is invalid.
    """
    query = """
    query GetPlayerHeroesStats($steamId: Long!, $heroesGroupByRequest: PlayerMatchesGroupByRequestType!) {
      player(steamAccountId: $steamId) {
        heroesGroupBy: matchesGroupBy(request: $heroesGroupByRequest) {
          ... on MatchGroupByHeroType {
            heroId
            matchCount
            winCount
            avgGoldPerMinute
            avgExperiencePerMinute
            lastMatchDateTime
            avgAssists
            avgKills
            avgDeaths
          }
        }
      }
    }
    """

    variables = {
        "steamId": int(steam_id),
        "heroesGroupByRequest": {
            "positionIds": [position],
            "gameModeIds": GAME_MODES,
            "lobbyTypeIds": LOBBY_TYPES,
            "groupBy": "HERO",
            "playerList": "SINGLE",
            "skip": 0,
            "take": 10000,
        }
    }
    
    # Add minGameVersionId only if provided
    if min_game_version_id is not None:
        variables["heroesGroupByRequest"]["minGameVersionId"] = min_game_version_id

    headers = {
        "Content-Type": "application/json",
        "Authorization": f"bearer {STRATZ_TOKEN}",
        "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36"
    }

    payload = {"query": query, "variables": variables}
    response = requests.post(STRATZ_API_URL, json=payload, headers=headers)
    response.raise_for_status()

    data = response.json()
    
    # Validate API response
    if not validate_api_response(data):
        raise ValueError("Invalid API response structure")
    
    heroes_data = data["data"]["player"]["heroesGroupBy"]

    # Add position information to each hero
    for hero in heroes_data:
        hero["position"] = position

    return heroes_data


def filter_heroes_by_win_rate(heroes_data: List[Dict[str, Any]], min_win_rate: float = MIN_WIN_RATE) -> List[Dict[str, Any]]:
    """
    Filter heroes based on win rate and calculate Wilson score ratings.

    Args:
        heroes_data: Raw hero data from API.
        min_win_rate: Minimum win rate threshold.

    Returns:
        Filtered heroes with Wilson score ratings.
    """
    filtered_heroes = []

    for hero in heroes_data:
        if hero["matchCount"] > 0:
            win_rate = hero["winCount"] / hero["matchCount"]
            if win_rate >= min_win_rate:
                hero["rating"] = wilson_score(
                    hero["winCount"], hero["matchCount"])
                filtered_heroes.append(hero)

    return filtered_heroes


def process_hero_data(steam_id: str, patch_name: str, min_game_version_id: Optional[int] = None) -> List[Dict[str, Any]]:
    """
    Common logic for processing hero data for a specific patch.
    
    Args:
        steam_id: Steam account ID.
        patch_name: Name of the patch for logging.
        min_game_version_id: Optional minimum game version ID filter.
        
    Returns:
        List of processed hero data.
    """
    all_heroes_data = []
    
    for position in POSITIONS:
        logger.info(f"Fetching {patch_name} data for {POSITION_NAMES[position]}...")
        try:
            heroes_data = fetch_hero_data(steam_id, position, min_game_version_id)
            all_heroes_data.extend(heroes_data)
            logger.info(f"  Found {len(heroes_data)} heroes for {POSITION_NAMES[position]}")
        except requests.RequestException as e:
            logger.error(f"  Error fetching {patch_name} data for {POSITION_NAMES[position]}: {e}")
            continue

    if not all_heroes_data:
        logger.warning(f"No hero data found for {patch_name}")
        return []

    # Filter heroes by win rate
    logger.info(f"Filtering {patch_name} heroes by win rate...")
    filtered_heroes = filter_heroes_by_win_rate(all_heroes_data)
    logger.info(f"Found {len(filtered_heroes)} {patch_name} heroes with win rate >= {int(MIN_WIN_RATE * 100)}%")

    # Sort by most recent match
    sorted_hero_data = sorted(
        filtered_heroes,
        key=lambda x: x["rating"],
        reverse=True
    )

    return sorted_hero_data


def save_grid_config(grid_config: Dict[str, Any], output_path: str) -> None:
    """
    Save grid configuration to JSON file.

    Args:
        grid_config: Grid configuration dictionary.
        output_path: Path to save the configuration file.
    """
    # Ensure directory exists
    os.makedirs(os.path.dirname(output_path), exist_ok=True)

    with open(output_path, "w") as json_file:
        json.dump(grid_config, json_file, indent=4)


def main() -> None:
    """Main function to generate hero grid configuration."""
    try:
        # Validate configuration
        validate_configuration()

        logger.info(f"Fetching hero data for Steam ID: {STEAM_ID}")

        # Create both grid configurations
        all_configs = []
        
        # 1. Current Patch Grid Configuration
        logger.info(f"Creating {CURRENT_PATCH} grid configuration...")
        sorted_hero_data_current_patch = process_hero_data(
            STEAM_ID, 
            CURRENT_PATCH, 
            min_game_version_id=PATCH_VERSIONS[CURRENT_PATCH]
        )

        if sorted_hero_data_current_patch:
            # Create current patch grid configuration
            grid_config_current_patch = create_hero_grid_config(
                sorted_hero_data_current_patch, 
                f"{CURRENT_PATCH} - High Win Rate Heroes (>= {int(MIN_WIN_RATE * 100)}%)"
            )
            all_configs.append(grid_config_current_patch["configs"][0])

        # 2. All-time Grid Configuration
        logger.info("Creating all-time grid configuration...")
        sorted_hero_data_all_time = process_hero_data(STEAM_ID, "all-time")  # No minGameVersionId

        if sorted_hero_data_all_time:
            # Create all-time grid configuration
            grid_config_all_time = create_hero_grid_config(
                sorted_hero_data_all_time, 
                f"All time - High Win Rate Heroes (>= {int(MIN_WIN_RATE * 100)}%)"
            )
            all_configs.append(grid_config_all_time["configs"][0])

        if not all_configs:
            logger.warning("No hero data found. Exiting.")
            return

        # Create final configuration with both grids
        final_config = {
            "version": 3,
            "configs": all_configs
        }

        # Save configuration
        output_path = f"{DOTA_CFG_PATH}/hero_grid_config.json"
        save_grid_config(final_config, output_path)
        logger.info(f"Grid configuration saved to: {output_path}")
        logger.info(f"Created {len(all_configs)} grid configurations")

    except ValueError as e:
        logger.error(f"Configuration error: {e}")
    except requests.exceptions.HTTPError as http_err:
        logger.error(f"HTTP error occurred: {http_err}")
        if hasattr(http_err, 'response') and http_err.response:
            logger.error(f"Response text: {http_err.response.text}")
    except requests.exceptions.RequestException as err:
        logger.error(f"Request error occurred: {err}")
    except json.JSONDecodeError as e:
        logger.error(f"JSON decode error: {e}")
    except Exception as e:
        logger.error(f"Unexpected error: {e}")


if __name__ == "__main__":
    main()