Spaces:
Running
Running
File size: 7,532 Bytes
1999a98 |
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 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 |
# Ultralytics 🚀 AGPL-3.0 License - https://ultralytics.com/license
import concurrent.futures
import statistics
import time
from typing import List, Optional, Tuple
import requests
class GCPRegions:
"""
A class for managing and analyzing Google Cloud Platform (GCP) regions.
This class provides functionality to initialize, categorize, and analyze GCP regions based on their
geographical location, tier classification, and network latency.
Attributes:
regions (Dict[str, Tuple[int, str, str]]): A dictionary of GCP regions with their tier, city, and country.
Methods:
tier1: Returns a list of tier 1 GCP regions.
tier2: Returns a list of tier 2 GCP regions.
lowest_latency: Determines the GCP region(s) with the lowest network latency.
Examples:
>>> from ultralytics.hub.google import GCPRegions
>>> regions = GCPRegions()
>>> lowest_latency_region = regions.lowest_latency(verbose=True, attempts=3)
>>> print(f"Lowest latency region: {lowest_latency_region[0][0]}")
"""
def __init__(self):
"""Initializes the GCPRegions class with predefined Google Cloud Platform regions and their details."""
self.regions = {
"asia-east1": (1, "Taiwan", "China"),
"asia-east2": (2, "Hong Kong", "China"),
"asia-northeast1": (1, "Tokyo", "Japan"),
"asia-northeast2": (1, "Osaka", "Japan"),
"asia-northeast3": (2, "Seoul", "South Korea"),
"asia-south1": (2, "Mumbai", "India"),
"asia-south2": (2, "Delhi", "India"),
"asia-southeast1": (2, "Jurong West", "Singapore"),
"asia-southeast2": (2, "Jakarta", "Indonesia"),
"australia-southeast1": (2, "Sydney", "Australia"),
"australia-southeast2": (2, "Melbourne", "Australia"),
"europe-central2": (2, "Warsaw", "Poland"),
"europe-north1": (1, "Hamina", "Finland"),
"europe-southwest1": (1, "Madrid", "Spain"),
"europe-west1": (1, "St. Ghislain", "Belgium"),
"europe-west10": (2, "Berlin", "Germany"),
"europe-west12": (2, "Turin", "Italy"),
"europe-west2": (2, "London", "United Kingdom"),
"europe-west3": (2, "Frankfurt", "Germany"),
"europe-west4": (1, "Eemshaven", "Netherlands"),
"europe-west6": (2, "Zurich", "Switzerland"),
"europe-west8": (1, "Milan", "Italy"),
"europe-west9": (1, "Paris", "France"),
"me-central1": (2, "Doha", "Qatar"),
"me-west1": (1, "Tel Aviv", "Israel"),
"northamerica-northeast1": (2, "Montreal", "Canada"),
"northamerica-northeast2": (2, "Toronto", "Canada"),
"southamerica-east1": (2, "São Paulo", "Brazil"),
"southamerica-west1": (2, "Santiago", "Chile"),
"us-central1": (1, "Iowa", "United States"),
"us-east1": (1, "South Carolina", "United States"),
"us-east4": (1, "Northern Virginia", "United States"),
"us-east5": (1, "Columbus", "United States"),
"us-south1": (1, "Dallas", "United States"),
"us-west1": (1, "Oregon", "United States"),
"us-west2": (2, "Los Angeles", "United States"),
"us-west3": (2, "Salt Lake City", "United States"),
"us-west4": (2, "Las Vegas", "United States"),
}
def tier1(self) -> List[str]:
"""Returns a list of GCP regions classified as tier 1 based on predefined criteria."""
return [region for region, info in self.regions.items() if info[0] == 1]
def tier2(self) -> List[str]:
"""Returns a list of GCP regions classified as tier 2 based on predefined criteria."""
return [region for region, info in self.regions.items() if info[0] == 2]
@staticmethod
def _ping_region(region: str, attempts: int = 1) -> Tuple[str, float, float, float, float]:
"""Pings a specified GCP region and returns latency statistics: mean, min, max, and standard deviation."""
url = f"https://{region}-docker.pkg.dev"
latencies = []
for _ in range(attempts):
try:
start_time = time.time()
_ = requests.head(url, timeout=5)
latency = (time.time() - start_time) * 1000 # convert latency to milliseconds
if latency != float("inf"):
latencies.append(latency)
except requests.RequestException:
pass
if not latencies:
return region, float("inf"), float("inf"), float("inf"), float("inf")
std_dev = statistics.stdev(latencies) if len(latencies) > 1 else 0
return region, statistics.mean(latencies), std_dev, min(latencies), max(latencies)
def lowest_latency(
self,
top: int = 1,
verbose: bool = False,
tier: Optional[int] = None,
attempts: int = 1,
) -> List[Tuple[str, float, float, float, float]]:
"""
Determines the GCP regions with the lowest latency based on ping tests.
Args:
top (int): Number of top regions to return.
verbose (bool): If True, prints detailed latency information for all tested regions.
tier (int | None): Filter regions by tier (1 or 2). If None, all regions are tested.
attempts (int): Number of ping attempts per region.
Returns:
(List[Tuple[str, float, float, float, float]]): List of tuples containing region information and
latency statistics. Each tuple contains (region, mean_latency, std_dev, min_latency, max_latency).
Examples:
>>> regions = GCPRegions()
>>> results = regions.lowest_latency(top=3, verbose=True, tier=1, attempts=2)
>>> print(results[0][0]) # Print the name of the lowest latency region
"""
if verbose:
print(f"Testing GCP regions for latency (with {attempts} {'retry' if attempts == 1 else 'attempts'})...")
regions_to_test = [k for k, v in self.regions.items() if v[0] == tier] if tier else list(self.regions.keys())
with concurrent.futures.ThreadPoolExecutor(max_workers=50) as executor:
results = list(executor.map(lambda r: self._ping_region(r, attempts), regions_to_test))
sorted_results = sorted(results, key=lambda x: x[1])
if verbose:
print(f"{'Region':<25} {'Location':<35} {'Tier':<5} Latency (ms)")
for region, mean, std, min_, max_ in sorted_results:
tier, city, country = self.regions[region]
location = f"{city}, {country}"
if mean == float("inf"):
print(f"{region:<25} {location:<35} {tier:<5} Timeout")
else:
print(f"{region:<25} {location:<35} {tier:<5} {mean:.0f} ± {std:.0f} ({min_:.0f} - {max_:.0f})")
print(f"\nLowest latency region{'s' if top > 1 else ''}:")
for region, mean, std, min_, max_ in sorted_results[:top]:
tier, city, country = self.regions[region]
location = f"{city}, {country}"
print(f"{region} ({location}, {mean:.0f} ± {std:.0f} ms ({min_:.0f} - {max_:.0f}))")
return sorted_results[:top]
# Usage example
if __name__ == "__main__":
regions = GCPRegions()
top_3_latency_tier1 = regions.lowest_latency(top=3, verbose=True, tier=1, attempts=3)
|