Skip to content

hyperion.domain.geo

hyperion.domain.geo

Pure-domain geographical primitives.

Split out of :mod:hyperion.infrastructure.geo.location (F2 / DDD refactor Step 3). :class:Location, :class:NamedLocation, :class:SpatialKMeans and the distance/conversion helpers are pure data + math: they depend only on the stdlib, haversine (lite core) and lite-core :mod:hyperion.log.

There is no cache reference and no singleton here -- the old Location._cache class variable (which pulled :class:hyperion.ports.cache.Cache and, via Cache.from_config(), boto3) is gone. get_distance simply computes the haversine (or, when approximate, the equirectangular) distance; callers that want memoisation wrap it themselves (e.g. :func:functools.lru_cache).

numpy is imported lazily inside :class:SpatialKMeans, so this module never requires numpy itself; only SpatialKMeans use does (it raises a clear ImportError without it). haversine does its own optional import numpy at load iff numpy is installed (a vector fast-path -- not a declared haversine dependency) and falls back to a pure-math scalar path when absent, which is the path :meth:Location.get_distance uses. numpy is therefore safely gated under the [data] extra; the dependency move itself lands in DDD refactor Step 10.

SpatialKMeans

SpatialKMeans(locations)

Bases: Generic[AnyLocation]

K-means clustering for geographical locations.

Initialize the K-means clustering with the given locations.

Parameters:

Name Type Description Default
locations Iterable[Location]

The locations to cluster.

required
Source code in hyperion/domain/geo.py
def __init__(self, locations: Iterable[AnyLocation]) -> None:
    """Initialize the K-means clustering with the given locations.

    Args:
        locations (Iterable[Location]): The locations to cluster.
    """
    import numpy as np

    self.locations = list(locations)
    self.locations_array = np.array(
        [[location.latitude, location.longitude] for location in self.locations],
    )

fit

fit(k, max_iters=100)

Fit the K-means model with k clusters and return the clusters.

Parameters:

Name Type Description Default
k int

The desired number of clusters (must be less than the number of locations).

required
max_iters int

The maximum number of iterations. Defaults to 100.

100

Returns:

Type Description
dict[Location, list[AnyLocation]]

dict[Location, list[Location]]: The clusters with the centroids as keys.

Source code in hyperion/domain/geo.py
def fit(self, k: int, max_iters: int = 100) -> dict[Location, list[AnyLocation]]:
    """Fit the K-means model with k clusters and return the clusters.

    Args:
        k (int): The desired number of clusters (must be less than the number of locations).
        max_iters (int, optional): The maximum number of iterations. Defaults to 100.

    Returns:
        dict[Location, list[Location]]: The clusters with the centroids as keys.
    """
    import numpy as np

    centroids = self.locations_array[np.random.choice(len(self.locations_array), k, replace=False), :2]

    for _ in range(max_iters):
        distances = self._get_distances_from_centroids(centroids)

        cluster_assignments = np.argmin(distances, axis=1)

        new_centroids_list = []
        for cluster_idx in range(k):
            cluster_points = self.locations_array[cluster_assignments == cluster_idx, :2]
            if len(cluster_points) > 0:
                new_centroids_list.append(cluster_points.mean(axis=0))
            else:
                new_centroids_list.append(centroids[cluster_idx])  # Keep old centroid for empty clusters
        new_centroids = np.array(new_centroids_list)

        if np.allclose(centroids, new_centroids):
            break
        centroids = new_centroids

    clusters: dict[Location, list[AnyLocation]] = defaultdict(list)
    centroid_locations = [Location(float(lat), float(lon)) for lat, lon in centroids]

    for location_id, centroid_id in enumerate(cluster_assignments):
        clusters[centroid_locations[centroid_id]].append(self.locations[location_id])
    return dict(clusters)

Location dataclass

Location(latitude, longitude, title=None, address=None)

A geographical location.

get_distance

get_distance(other, approximate=False)

Get the distance to another location in meters. If approximate is False (default), will use haversine. Otherwise uses Euclidean to approximate the distance.

Parameters:

Name Type Description Default
other Location

The other location.

required
approximate bool

Whether to approximate the distance. Defaults to False.

False

Returns:

Name Type Description
float float

The distance in meters.

Source code in hyperion/domain/geo.py
def get_distance(self, other: Location, approximate: bool = False) -> float:
    """Get the distance to another location in meters.
    If approximate is False (default), will use haversine. Otherwise uses Euclidean to approximate the distance.

    Args:
        other (Location): The other location.
        approximate (bool, optional): Whether to approximate the distance. Defaults to False.

    Returns:
        float: The distance in meters.
    """
    if approximate:
        return self._get_distance_euclidean(other)
    return self._get_distance_haversine(other)

get_nearest

get_nearest(others, threshold=None, approximate=False)

Get the closest location from the iterable of locations.

If threshold is given and all locations are further than the threshold in meters, an ValueError is raised.

Parameters:

Name Type Description Default
others Iterable[Location]

The other locations.

required
threshold float

The maximum distance in meters. Defaults to None.

None
approximate bool

Whether to approximate the distance. Defaults to False.

False

Returns:

Name Type Description
Location AnyLocation

The nearest location.

Source code in hyperion/domain/geo.py
def get_nearest(
    self, others: Iterable[AnyLocation], threshold: float | None = None, approximate: bool = False
) -> AnyLocation:
    """Get the closest location from the iterable of locations.

    If threshold is given and all locations are further than the threshold in meters, an ValueError is raised.

    Args:
        others (Iterable[Location]): The other locations.
        threshold (float, optional): The maximum distance in meters. Defaults to None.
        approximate (bool, optional): Whether to approximate the distance. Defaults to False.

    Returns:
        Location: The nearest location.
    """
    nearest: tuple[AnyLocation, float] | None = None
    for other in others:
        distance = self.get_distance(other, approximate=approximate)
        if threshold is not None and distance > threshold:
            continue
        if nearest is None or nearest[1] > distance:
            nearest = (other, distance)
    if nearest is None:
        raise ValueError(f"None of the given locations is close enough to {self!r}.")
    logger.debug("Found nearest location.", this=self, other=nearest[0], distance=nearest[1])
    return nearest[0]

meters_to_degrees

meters_to_degrees(meters, at_latitude)

Convert meters to degrees at a given latitude.

Parameters:

Name Type Description Default
meters float

The distance in meters.

required
at_latitude float

The latitude at which the conversion should be done.

required

Returns:

Type Description
tuple[float, float]

tuple[float, float]: The distance in degrees for latitude and longitude.

Raises:

Type Description
ValueError

If at_latitude is ±90 (the poles), where longitude degrees are geometrically undefined (cos(±90°) = 0).

Source code in hyperion/domain/geo.py
def meters_to_degrees(meters: float, at_latitude: float) -> tuple[float, float]:
    """Convert meters to degrees at a given latitude.

    Args:
        meters (float): The distance in meters.
        at_latitude (float): The latitude at which the conversion should be done.

    Returns:
        tuple[float, float]: The distance in degrees for latitude and longitude.

    Raises:
        ValueError: If ``at_latitude`` is ±90 (the poles), where longitude
            degrees are geometrically undefined (cos(±90°) = 0).
    """
    if abs(at_latitude) >= 90:
        raise ValueError(
            f"Cannot convert meters to longitude degrees at latitude {at_latitude}: "
            "longitude degrees are undefined at the poles (|latitude| >= 90)."
        )
    return meters / LATITUDE_DEGREE_TO_METERS, meters / (
        LATITUDE_DEGREE_TO_METERS * math.cos(math.radians(at_latitude))
    )