skyeyesystem/backend/Skyeye-sys-dev/skyeye-service-py/klkx/planning/geohash/geohash.py

113 lines
3.8 KiB
Python
Raw Permalink Normal View History

BASE32 = "0123456789bcdefghjkmnpqrstuvwxyz"
# 邻居查找表(纬度和经度交替变化)
NEIGHBORS = {
"even": {
"N": "p0r21436x8zb9dcf5h7kjnmqesgutwvy",
"S": "14365h7k9dcfesgujnmqp0r2twvyx8zb",
"E": "bc01fg45238967deuvhjyznpkmstqrwx",
"W": "238967debc01fg45kmstqrwxuvhjyznp"
},
"odd": {
"N": "bc01fg45238967deuvhjyznpkmstqrwx",
"S": "238967debc01fg45kmstqrwxuvhjyznp",
"E": "p0r21436x8zb9dcf5h7kjnmqesgutwvy",
"W": "14365h7k9dcfesgujnmqp0r2twvyx8zb"
}
}
# 边界字符(用于计算超出边界的情况)
BORDERS = {
"even": {"N": "prxz", "S": "028b", "E": "bcfguvyz", "W": "0145hjnp"},
"odd": {"N": "bcfguvyz", "S": "0145hjnp", "E": "prxz", "W": "028b"}
}
def encode_geohash(lat, lng, precision=7):
"""手动实现 GeoHash 编码"""
lat_range = [-90.0, 90.0]
lng_range = [-180.0, 180.0]
binary = ""
is_even = True # 纬度和经度交替编码
while len(binary) < precision * 5:
mid = (lng_range[0] + lng_range[1]) / 2 if is_even else (lat_range[0] + lat_range[1]) / 2
if (lng if is_even else lat) >= mid:
binary += "1"
if is_even:
lng_range[0] = mid
else:
lat_range[0] = mid
else:
binary += "0"
if is_even:
lng_range[1] = mid
else:
lat_range[1] = mid
is_even = not is_even
# 5 位二进制转 Base32
return "".join(BASE32[int(binary[i:i+5], 2)] for i in range(0, len(binary), 5))
def decode_geohash(geohash):
"""解码 GeoHash 到 (lat, lng)"""
lat_range = [-90.0, 90.0]
lng_range = [-180.0, 180.0]
binary = "".join(f"{BASE32.index(c):05b}" for c in geohash)
is_even = True
for bit in binary:
if is_even:
mid = (lng_range[0] + lng_range[1]) / 2
lng_range = [mid, lng_range[1]] if bit == "1" else [lng_range[0], mid]
else:
mid = (lat_range[0] + lat_range[1]) / 2
lat_range = [mid, lat_range[1]] if bit == "1" else [lat_range[0], mid]
is_even = not is_even
return (sum(lat_range) / 2, sum(lng_range) / 2)
def get_neighbor(geohash, direction):
"""计算 GeoHash 的单个方向邻居"""
if not geohash:
return None
precision = len(geohash)
last_char = geohash[-1] # 取最后一个字符
geohash_prefix = geohash[:-1] # 除去最后一个字符
is_even = (precision % 2) == 0 # 判断奇偶精度
type_key = "even" if is_even else "odd"
# 如果当前 GeoHash 位于边界,需要递归调整前缀
if last_char in BORDERS[type_key][direction] and geohash_prefix:
geohash_prefix = get_neighbor(geohash_prefix, direction)
# 计算相邻 GeoHash
new_last_char = BASE32[NEIGHBORS[type_key][direction].index(last_char)]
return geohash_prefix + new_last_char
def get_all_neighbors(geohash):
"""计算 GeoHash 的 8 个相邻格子"""
return {
"N": get_neighbor(geohash, "N"),
"S": get_neighbor(geohash, "S"),
"E": get_neighbor(geohash, "E"),
"W": get_neighbor(geohash, "W"),
"NE": get_neighbor(get_neighbor(geohash, "N"), "E"),
"NW": get_neighbor(get_neighbor(geohash, "N"), "W"),
"SE": get_neighbor(get_neighbor(geohash, "S"), "E"),
"SW": get_neighbor(get_neighbor(geohash, "S"), "W")
}
if __name__ == '__main__':
# 示例计算北京天安门广场39.9087, 116.3975)的邻居
gh = encode_geohash(39.9087, 116.3975, precision=7)
neighbors = get_all_neighbors(gh)
print(f"中心 GeoHash: {gh}, {decode_geohash(gh)}")
print("相邻 GeoHash:")
for direction, neighbor in neighbors.items():
print(f"{direction}: {neighbor}, {decode_geohash(neighbor)}")