scipy usage survey

1_scipy_package_usage.csv

2_scipy_function_usage.csv

3_scipy_usage.py

# /// script
# requires-python = ">=3.12"
# dependencies = [
#     "pandas",
#     "requests",
# ]
# ///

"""
GitHub SciPy Usage Analysis Script using GraphQL API

This script uses GitHub's GraphQL API to analyze scipy usage patterns from GitHub repositories.
It searches for specific import patterns and function usage across repositories.

Setup:
1. Create a GitHub Personal Access Token with 'public_repo' scope
2. Set GITHUB_TOKEN environment variable or update the script
3. Run with `uv run --env-file <ENV_FILE> <THIS_SCRIPT>`

Disclaimer: This was almost completely vibe-coded.
"""

# ruff: noqa: T201

import base64
import json
import logging
import os
import re
import time
from collections import Counter
from datetime import datetime
from typing import Any

import pandas as pd
import requests

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


class GitHubSciPyAnalyzer:
    results: dict[str, Any]
    rate_limit_reset: int | None

    def __init__(self, github_token: str) -> None:
        """
        Initialize the GitHub SciPy usage analyzer.

        Args:
            github_token: GitHub Personal Access Token
        """
        self.github_token = github_token
        self.api_url = "https://api.github.com/graphql"
        self.headers = {
            "Authorization": f"Bearer {github_token}",
            "Content-Type": "application/json",
        }

        # Rate limiting
        self.requests_made = 0
        self.rate_limit_remaining = 5000
        self.rate_limit_reset = None

        # Define scipy modules and search patterns
        self.scipy_modules = [
            "stats",
            "linalg",
            "optimize",
            "integrate",
            "interpolate",
            "signal",
            "sparse",
            "spatial",
            "special",
            "ndimage",
            "io",
            "fft",
            "cluster",
            "constants",
            "misc",
            "odr",
        ]

        # Results storage
        self.results = {
            "module_usage": Counter(),
            "function_usage": Counter(),
            "import_patterns": Counter(),
            "repositories": set(),
            "files_analyzed": 0,
        }

    def wait_for_rate_limit(self) -> None:
        """Handle rate limiting by waiting if necessary."""
        logger.debug(
            "Rate limit check: remaining=%d, reset=%s",
            self.rate_limit_remaining,
            self.rate_limit_reset,
        )

        # Check if we're actually close to the rate limit
        if self.rate_limit_remaining <= 10:
            if self.rate_limit_reset:
                current_time = time.time()
                wait_time = max(0, self.rate_limit_reset - current_time) + 1
                logger.info(
                    "Rate limit low (remaining: %d), waiting %d seconds...",
                    self.rate_limit_remaining,
                    round(wait_time, 2),
                )
                time.sleep(wait_time)
                # Reset rate limit info after waiting
                self.rate_limit_remaining = 5000
            else:
                # If no reset time, wait a default amount
                logger.info("Rate limit low but no reset time, waiting 60 seconds...")
                time.sleep(60)
                self.rate_limit_remaining = 5000

    def make_graphql_request(
        self, query: str, variables: dict[str, Any] | None = None
    ) -> dict[str, Any] | None:
        """
        Make a GraphQL request to GitHub API.

        Args:
            query: GraphQL query string
            variables: Query variables

        Returns:
            Response data
        """
        self.wait_for_rate_limit()

        payload: dict[str, Any] = {"query": query}
        if variables:
            payload["variables"] = variables

        response = requests.post(self.api_url, json=payload, headers=self.headers)
        self.requests_made += 1

        # Update rate limit info
        self.rate_limit_remaining = int(
            response.headers.get("X-RateLimit-Remaining", 0)
        )
        reset_time = response.headers.get("X-RateLimit-Reset")
        if reset_time:
            self.rate_limit_reset = int(reset_time)

        if response.status_code != 200:
            logger.error("GraphQL request failed: %d", response.status_code)
            logger.error("%s", response.text)
            return None

        data: dict[str, Any] = response.json()
        if "errors" in data:
            logger.error("GraphQL errors: %s", data["errors"])
            return None

        return data

    def search_code_files(
        self, search_query: str, max_results: int = 100
    ) -> list[dict[str, Any]]:
        """
        Search for code files using GitHub's REST API.

        Args:
            search_query: Search query string
            max_results: Maximum number of results to return

        Returns:
            List of file information with content
        """
        results: list[dict[str, Any]] = []
        page = 1
        per_page = min(100, max_results)

        while len(results) < max_results:
            # Use REST API for code search
            url = "https://api.github.com/search/code"
            headers = {
                "Authorization": f"Bearer {self.github_token}",
                "Accept": "application/vnd.github.v3+json",
            }

            params = {"q": search_query, "page": page, "per_page": per_page}

            self.wait_for_rate_limit()

            response = requests.get(url, headers=headers, params=params)
            self.requests_made += 1

            # Update rate limit info
            self.rate_limit_remaining = int(
                response.headers.get("X-RateLimit-Remaining", 5000)
            )
            reset_time = response.headers.get("X-RateLimit-Reset")
            if reset_time:
                self.rate_limit_reset = int(reset_time)

            logger.debug(
                "After request: remaining=%d, reset=%s, status=%d",
                self.rate_limit_remaining,
                self.rate_limit_reset,
                response.status_code,
            )

            if response.status_code != 200:
                logger.error("REST API request failed: %d", response.status_code)
                logger.error("%s", response.text)
                if response.status_code == 403:
                    logger.error("Rate limit exceeded or forbidden")
                    # Wait longer for rate limit reset
                    if self.rate_limit_reset:
                        wait_time = max(0, self.rate_limit_reset - time.time()) + 1
                        logger.info(
                            "Waiting %d seconds for rate limit reset...",
                            round(wait_time, 2),
                        )
                        time.sleep(wait_time)
                        self.rate_limit_remaining = 5000
                        continue
                break

            data = response.json()

            for item in data.get("items", []):
                if len(results) >= max_results:
                    break

                # Get file content using raw URL
                content_url = item.get("git_url", "")
                if content_url:
                    content = self.get_file_content(content_url)
                    if content:
                        results.append({
                            "repo_name": item["repository"]["full_name"],
                            "content": content,
                            "fragment": item.get("text_matches", [{}])[0].get(
                                "fragment", ""
                            ),
                        })

            # Check if there are more pages
            if len(data.get("items", [])) < per_page:
                break

            page += 1

        return results[:max_results]

    def get_file_content(self, git_url: str) -> str | None:
        """
        Get file content from GitHub's git API.

        Args:
            git_url: Git URL for the file

        Returns:
            File content or None if failed
        """
        self.wait_for_rate_limit()

        headers = {
            "Authorization": f"Bearer {self.github_token}",
            "Accept": "application/vnd.github.v3+json",
        }

        response = requests.get(git_url, headers=headers)
        self.requests_made += 1

        # Update rate limit info
        self.rate_limit_remaining = int(
            response.headers.get("X-RateLimit-Remaining", 5000)
        )
        reset_time = response.headers.get("X-RateLimit-Reset")
        if reset_time:
            self.rate_limit_reset = int(reset_time)

        if response.status_code != 200:
            return None

        data = response.json()

        # Content is base64 encoded
        try:
            content = base64.b64decode(data["content"]).decode("utf-8")
            return content
        except Exception:
            return None

    def search_scipy_patterns(self, max_files_per_pattern: int = 200) -> dict[str, Any]:
        """
        Search for various scipy usage patterns.

        Args:
            max_files_per_pattern: Maximum files to analyze per pattern

        Returns:
            Dictionary of search results by pattern
        """
        search_patterns = [
            # Import patterns
            'language:python "from scipy.stats import"',
            'language:python "from scipy.linalg import"',
            'language:python "from scipy.optimize import"',
            'language:python "from scipy.integrate import"',
            'language:python "from scipy.interpolate import"',
            'language:python "from scipy.signal import"',
            'language:python "from scipy.sparse import"',
            'language:python "from scipy.spatial import"',
            'language:python "from scipy.special import"',
            'language:python "from scipy.ndimage import"',
            # Module imports
            'language:python "import scipy.stats"',
            'language:python "import scipy.linalg"',
            'language:python "import scipy.optimize"',
            'language:python "import scipy.signal"',
            # Function calls
            'language:python "scipy.stats."',
            'language:python "scipy.linalg."',
            'language:python "scipy.optimize."',
            'language:python "scipy.signal."',
        ]

        all_results: dict[str, Any] = {}

        for i, pattern in enumerate(search_patterns):
            logger.info(
                "Searching pattern %d/%d: %s", i + 1, len(search_patterns), pattern
            )

            try:
                results = self.search_code_files(pattern, max_files_per_pattern)
                all_results[pattern] = results
                logger.info("Found %d files for pattern: %s", len(results), pattern)

                # Small delay to be respectful to the API
                time.sleep(1)

            except Exception as e:
                logger.exception("Error searching pattern '%s': %s", pattern, e)
                continue

        return all_results

    def analyze_file_content(self, content: str, repo_name: str) -> None:
        """
        Analyze a single file's content for scipy usage.

        Args:
            content: File content
            repo_name: Repository name
        """
        if not content:
            return

        self.results["files_analyzed"] += 1
        self.results["repositories"].add(repo_name)

        # Patterns to match
        patterns = {
            "from_imports": r"from\s+scipy\.(\w+)\s+import\s+([^#\n]+)",
            "module_imports": r"import\s+scipy\.(\w+)(?:\s+as\s+\w+)?",
            "function_calls": r"scipy\.(\w+)\.(\w+)",
            "direct_scipy_import": r"from\s+scipy\s+import\s+([^#\n]+)",
        }

        # Analyze from imports
        for match in re.finditer(patterns["from_imports"], content):
            module = match.group(1)
            imports = match.group(2)

            self.results["module_usage"][f"scipy.{module}"] += 1
            self.results["import_patterns"][f"from scipy.{module} import"] += 1

            # Parse individual functions
            for func in imports.split(","):
                func = func.strip().split(" as ")[0].strip()
                if func and func != "*" and func.isidentifier():
                    self.results["function_usage"][f"scipy.{module}.{func}"] += 1

        # Analyze module imports
        for match in re.finditer(patterns["module_imports"], content):
            module = match.group(1)
            self.results["module_usage"][f"scipy.{module}"] += 1
            self.results["import_patterns"][f"import scipy.{module}"] += 1

        # Analyze function calls
        for match in re.finditer(patterns["function_calls"], content):
            module = match.group(1)
            function = match.group(2)

            self.results["module_usage"][f"scipy.{module}"] += 1
            self.results["function_usage"][f"scipy.{module}.{function}"] += 1

        # Analyze direct scipy imports
        for match in re.finditer(patterns["direct_scipy_import"], content):
            imports = match.group(1)
            self.results["import_patterns"]["from scipy import"] += 1

            for module in imports.split(","):
                module = module.strip().split(" as ")[0].strip()
                if module and module in self.scipy_modules:
                    self.results["module_usage"][f"scipy.{module}"] += 1

    def run_analysis(
        self, max_files_per_pattern: int = 200, output_dir: str = "scipy_usage_results"
    ) -> None:
        """
        Run the complete analysis pipeline.

        Args:
            max_files_per_pattern: Maximum files to analyze per search pattern
            output_dir: Directory to save results
        """
        logger.info("Starting SciPy usage analysis with GitHub GraphQL API...")

        # Search for scipy patterns
        search_results = self.search_scipy_patterns(max_files_per_pattern)

        # Analyze all found files
        total_files = sum(len(files) for files in search_results.values())
        logger.info("Analyzing %d files...", total_files)

        analyzed_files: set[str] = set()  # To avoid duplicates

        for pattern, files in search_results.items():
            logger.info("Processing %d files from pattern: %s", len(files), pattern)

            for file_info in files:
                file_key = f"{file_info['repo_name']}-{hash(file_info['content'])}"

                if file_key not in analyzed_files:
                    self.analyze_file_content(
                        file_info["content"], file_info["repo_name"]
                    )
                    analyzed_files.add(file_key)

        # Save results
        self.save_results(output_dir)
        self.print_summary()

        logger.info("Analysis complete!")

    def save_results(self, output_dir: str = "scipy_usage_results") -> None:
        """
        Save analysis results to files.

        Args:
            output_dir: Directory to save results
        """
        os.makedirs(output_dir, exist_ok=True)

        # Save module usage
        module_df = pd.DataFrame(
            [
                (module, count)
                for module, count in self.results["module_usage"].most_common()
            ],
            columns=["module", "usage_count"],
        )
        module_df.to_csv(f"{output_dir}/scipy_module_usage.csv", index=False)

        # Save function usage
        function_df = pd.DataFrame(
            [
                (func, count)
                for func, count in self.results["function_usage"].most_common()
            ],
            columns=["function", "usage_count"],
        )
        function_df.to_csv(f"{output_dir}/scipy_function_usage.csv", index=False)

        # Save import patterns
        import_df = pd.DataFrame(
            [
                (pattern, count)
                for pattern, count in self.results["import_patterns"].most_common()
            ],
            columns=["import_pattern", "usage_count"],
        )
        import_df.to_csv(f"{output_dir}/scipy_import_patterns.csv", index=False)

        # Save summary statistics
        summary = {
            "analysis_date": datetime.now().isoformat(),
            "total_repositories": len(self.results["repositories"]),
            "total_files_analyzed": self.results["files_analyzed"],
            "unique_modules": len(self.results["module_usage"]),
            "unique_functions": len(self.results["function_usage"]),
            "api_requests_made": self.requests_made,
            "top_10_modules": dict(self.results["module_usage"].most_common(10)),
            "top_20_functions": dict(self.results["function_usage"].most_common(20)),
            "top_import_patterns": dict(
                self.results["import_patterns"].most_common(10)
            ),
        }

        with open(f"{output_dir}/scipy_usage_summary.json", "w") as f:
            json.dump(summary, f, indent=2)

        logger.info("Results saved to %s/", output_dir)

    def print_summary(self) -> None:
        """Print a summary of the analysis results."""
        print("\n=== SciPy Usage Analysis Summary ===")
        print(f"Total repositories analyzed: {len(self.results['repositories']):,}")
        print(f"Total Python files analyzed: {self.results['files_analyzed']:,}")
        print(f"Unique scipy modules found: {len(self.results['module_usage'])}")
        print(f"Unique scipy functions found: {len(self.results['function_usage'])}")
        print(f"API requests made: {self.requests_made}")

        print("\n=== Top 10 Most Used SciPy Modules ===")
        for module, count in self.results["module_usage"].most_common(10):
            print(f"{module}: {count:,} occurrences")

        print("\n=== Top 20 Most Used SciPy Functions ===")
        for function, count in self.results["function_usage"].most_common(20):
            print(f"{function}: {count:,} occurrences")

        print("\n=== Most Common Import Patterns ===")
        for pattern, count in self.results["import_patterns"].most_common(10):
            print(f"{pattern}: {count:,} occurrences")


def main() -> None:
    """Main function to run the analysis."""

    # Configuration
    GITHUB_TOKEN = os.getenv("GITHUB_TOKEN")  # set your GitHub token
    MAX_FILES_PER_PATTERN = 200  # Adjust based on your needs
    OUTPUT_DIR = "scipy_usage_results"

    if not GITHUB_TOKEN:
        print("Error: GITHUB_TOKEN environment variable not set")
        print(
            "Please create a GitHub Personal Access Token and set it as an environment variable"
        )
        return

    # Initialize analyzer
    analyzer = GitHubSciPyAnalyzer(GITHUB_TOKEN)

    # Run analysis
    analyzer.run_analysis(MAX_FILES_PER_PATTERN, OUTPUT_DIR)


if __name__ == "__main__":
    main()