app.py

import gradio as gr
import requests
import json
import geoutil
from shapely.geometry import Polygon, MultiPoint, mapping
import re
import geopandas as gpd
import geo_level1
from openai import OpenAI
import numpy as np
import os

api_key = os.getenv('api_key')
client = OpenAI(
    api_key=api_key
)

model = "gpt-4o"

north = ["north", "N'", "North", "NORTH"]
south = ["south", "S'", "South", "SOUTH"]
east = ["east", "E'", "East", "EAST"]
west = ["west", "W'", "West", "WEST"]
northeast = ["north-east", "NE'", "north east", "NORTH-EAST", "North East", "NORTH EAST"]
southeast = ["south-east", "SE'", "south east", "SOUTH-EAST", "South East", "SOUTH EAST"]
northwest = ["north-west", "NW'", "north west", "NORTH-WEST", "North West", "NORTH WEST"]
southwest = ["south-west", "SW'", "south west", "SOUTH-WEST", "South West", "SOUTH WEST"]
center = ["center","central", "downtown","midtown"]




def to_standard_2d_list(data):
    arr = np.array(data)

    # 强制变成一维后 reshape，前提是元素总数是2的倍数
    flat = arr.flatten()
    if flat.size % 2 != 0:
        raise ValueError("元素个数不是2的倍数，不能 reshape 成 [N, 2] 格式")

    return flat.reshape(-1, 2).tolist()


def get_geojson(ent, arr, centroid):
    poly_json = {}
    poly_json['type'] = 'FeatureCollection'
    poly_json['features'] = []
    coordinates= []
    coordinates.append(arr)
    poly_json['features'].append({
    'type':'Feature',
    'id': ent,
    'properties': {
        'centroid': centroid
        },
    'geometry': {
        'type':'Polygon',
        'coordinates': coordinates
        }
    })
    return poly_json


def get_coordinates(ent):
    request_url = 'https://nominatim.openstreetmap.org/search.php?q= ' +ent +'&polygon_geojson=1&accept-language=en&format=jsonv2'
    headers = {
        "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/18.3 Safari/605.1.15"
    }
    page = requests.get(request_url, headers=headers, verify=False)
    json_content = json.loads(page.content)
    all_coordinates = json_content[0]['geojson']['coordinates'][0]
    centroid = (float(json_content[0]['lon']), float(json_content[0]['lat']))
    for p in all_coordinates:
        p2 = (p[0], p[1])
        angle = geoutil.calculate_bearing(centroid, p2)
        p.append(angle)

    geojson = get_geojson(ent, all_coordinates, centroid)

    return geojson['features'][0]['geometry']['coordinates'][0], geojson['features'][0]['properties']['centroid']

def get_coordinates(location):
    request_url = f'https://nominatim.openstreetmap.org/search.php?q={location}&polygon_geojson=1&accept-language=en&format=jsonv2'

    print(request_url)
    headers = {"User-Agent": "Mozilla/5.0"}
    response = requests.get(request_url, headers=headers, verify=False)
    json_content = json.loads(response.content)
    # print(json_content)
    if json_content[0]['geojson']['type'] == 'Polygon':
        coordinates = json_content[0]['geojson']['coordinates'][0]
    elif json_content[0]['geojson']['type'] == 'Point':
        coordinates = json_content[0]['geojson']['coordinates']
    elif json_content[0]['geojson']['type'] == 'MultiPolygon':
        coordinates = json_content[0]['geojson']['coordinates'][0][0]
    else:
        coordinates = get_inner(json_content[0]['geojson']['coordinates'])
        print(json_content[0]['geojson']['type'], 'refref')
    centroid = (float(json_content[0]['lon']), float(json_content[0]['lat']))
    return (coordinates, centroid)


# level3
def get_directional_coordinates_by_angle(coordinates, centroid, direction, minimum, maximum):
    # minimum = 157
    # maximum = 202

    direction_coordinates = []
    for p in coordinates:
        angle = geoutil.calculate_bearing(centroid, p)
        p2 = (p[0], p[1], angle)
        if direction in geo_level1.east:
            if angle >= minimum or angle <= maximum:
                direction_coordinates.append(p2)

        else:
            if angle >= minimum and angle <= maximum:
                direction_coordinates.append(p2)
    # print(type(direction_coordinates[0]))
    # if(direction in geo_level1.west):
    #    direction_coordinates.sort(key=lambda k: k[2], reverse=True)

    return direction_coordinates
def get_level3(level3):
    digits = re.findall('[0-9]+', level3)[0]
    unit = re.findall('[A-Za-z]+', level3)[0]
    return digits, unit

def get_direction_coordinates(coordinates, centroid, level1):
    min_max = geo_level1.get_min_max(level1)
    if min_max is not None:
        coord = get_directional_coordinates_by_angle(coordinates, centroid, level1, min_max[0], min_max[1])
        return coord
    return coordinates
def sort_west(poly1, poly2, centroid):
    coords1 = mapping(poly1)["features"][0]["geometry"]["coordinates"]
    coords2 = mapping(poly2)["features"][0]["geometry"]["coordinates"]
    coord1 = []
    coord2 = []
    coord = []
    for c in coords1:
        pol = list(c[::-1])
        coord1.extend(pol)
    for c in coords2:
        pol = list(c[::-1])
        coord2.extend(pol)
    coo1 = []
    coo2 = []
    for p in coord1:
        angle = geoutil.calculate_bearing(centroid, p)
        if angle >= 157 and angle <= 202:
            coo1.append((p[0], p[1], angle))
    for p in coord2:
        angle = geoutil.calculate_bearing(centroid, p)
        if angle >= 157 and angle <= 202:
            coo2.append((p[0], p[1], angle))
    coo1.extend(coo2)
    return coo1


def get_level3_coordinates(coordinates, level_3, level1):
    distance, unit = get_level3(level_3)
    kms = geoutil.get_kilometers(distance, unit)
    coord = []

    coords0, center = coordinates

    if not isinstance(coords0, list) or len(coords0) < 3:

        # 从原始点出发，根据方向移动距离 kms 得到新圆心
        lat_km = 111.32
        lon_km = 111.32 * np.cos(np.radians(center[1]))

        dx = dy = 0

        if level1 is not None:
            if level1 in geo_level1.east:
                dx = kms / lon_km
            elif level1 in geo_level1.west:
                dx = -kms / lon_km
            elif level1 in geo_level1.north:
                dy = kms / lat_km
            elif level1 in geo_level1.south:
                dy = -kms / lat_km
            # 你也可以支持 northeast、southwest 等复合方向

        new_center = (center[0] + dx, center[1] + dy)

        # 用固定半径画个圆（例如半径2km）
        r_km = 1  # 半径设为1km，你也可以设为其他值

        circle_points = []
        for theta in np.linspace(0, 360, num=100):
            theta_rad = np.radians(theta)
            d_lat = (np.sin(theta_rad) * r_km) / lat_km
            d_lon = (np.cos(theta_rad) * r_km) / lon_km
            circle_points.append((new_center[0] + d_lon, new_center[1] + d_lat))

        # 输出中心（使用新圆心）
        if circle_points:
            center_point = MultiPoint(circle_points).centroid
            center = (center_point.x, center_point.y)
        else:
            center = new_center

        return circle_points, center

    # 正常 polygon 流程
    poly1 = Polygon(coords0)
    polygon1 = gpd.GeoSeries(poly1)

    # 生成环形区域
    poly2 = polygon1.buffer(0.0095 * kms, join_style=2)
    poly3 = polygon1.buffer(0.013 * kms, join_style=2)
    poly = poly3.difference(poly2)

    # 获取坐标
    coords = mapping(poly)["features"][0]["geometry"]["coordinates"]
    for c in coords:
        pol = list(c[::-1])
        coord.extend(pol)

    # 方向裁剪
    if level1 is not None:
        coord = get_direction_coordinates(coord, coordinates[1], level1)
        if level1 in geo_level1.west:
            coord = sort_west(poly3, poly2, coordinates[1])

    # 计算质心
    if coord:
        center_point = MultiPoint(coord).centroid
        center = (center_point.x, center_point.y)
    else:
        center = coordinates[1]

    return coord, center
# level 3 end

# between
def get_between_coordinates(coordinates1, coordinates2):
    """
    计算两个区域之间的中间点，并生成一个等面积的圆形区域。
    如果某个输入仅为点（坐标长度 < 3），则其面积设为 0；
    如果两个输入都是点，则默认半径为 2km。
    :param coordinates1: 第一个区域的边界坐标和中心点
    :param coordinates2: 第二个区域的边界坐标和中心点
    :return: 圆形区域的坐标集和圆心
    """

    def is_valid_polygon(coords):
        return isinstance(coords, list) and len(coords) >= 3

    coords1, center1 = coordinates1
    coords2, center2 = coordinates2

    # 判断输入是否为合法多边形（>=3个点）
    if is_valid_polygon(coords1):
        poly1 = Polygon(coords1)
        area1 = poly1.area
    else:
        area1 = 0

    if is_valid_polygon(coords2):
        poly2 = Polygon(coords2)
        area2 = poly2.area
    else:
        area2 = 0

    # 计算中心点（两个中心的中点）
    midpoint = (
        (center1[0] + center2[0]) / 2,
        (center1[1] + center2[1]) / 2
    )

    # 如果两个区域都是点，则使用默认半径 2km
    if area1 == 0 and area2 == 0:
        r_km = 2
    else:
        avg_area = (area1 + area2) / 2
        r_km = np.sqrt(avg_area / np.pi) * 111.32  # 近似 km 半径

    # 经纬度距离换算因子
    lat_km = 111.32
    lon_km = 111.32 * np.cos(np.radians(midpoint[1]))

    # 生成圆形区域坐标（100个点）
    circle_points = []
    for theta in np.linspace(0, 360, num=100):
        theta_rad = np.radians(theta)
        d_lat = (np.sin(theta_rad) * r_km) / lat_km
        d_lon = (np.cos(theta_rad) * r_km) / lon_km
        circle_points.append((midpoint[0] + d_lon, midpoint[1] + d_lat))

    return circle_points, midpoint
# between end


def llmapi(text):
    system_prompt = (
        "你是一个资深的地理学家，你的任务是通过给定的一段自然语言，来选择正确的定位函数顺序以及他们的输入。\n"
        "你能选择的定位函数有：\n"
        "1. 相对定位（Relative Positioning）：输入为地点坐标，方位，距离。输出为距离‘距离’输入的地点坐标的‘方位’的坐标。\n"
        "2. 中间定位（Between Positioning）：输入为两个地点的坐标，输出为两个地点坐标的中点。\n"
        "请先进行思维链（CoT）推理，并最终用 JSON 格式输出你的答案，用 `<<<JSON>>>` 和 `<<<END>>>` 包裹起来。\n"
        "请确保所有输入仅包含：地点名称（字符串）、索引（整数）、方位（字符串，必须是英文）或距离（字符串，带单位），不允许返回诸如 'Chatswood 南4 km的坐标' 这样的内容。\n"
        "每个步骤编号都有 id 记录，然后如果某个输入是之前步骤的输出，那么输入对应步骤的 id。\n"
        "所有方向必须使用英文（如 south, west, northeast, etc.）。\n"
        "示例输出：\n"
        "<<<JSON>>>\n"
        "[{\"id\": 1, \"function\": \"Relative\", \"inputs\": [\"Chatswood\", \"south\", \"4 km\"]},"
        "{\"id\": 2, \"function\": \"Relative\", \"inputs\": [\"North Sydney\", \"west\", \"2 km\"]},"
        "{\"id\": 3, \"function\": \"Between\", \"inputs\": [1, 2]},"
        "{\"id\": 4, \"function\": \"Relative\", \"inputs\": [3, \"southwest\", \"5 km\"]}]\n"
        "<<<END>>>")

    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": text},
    ]

    chat_completion = client.chat.completions.create(
        messages=messages,
        model=model,
    )

    result = chat_completion.choices[0].message.content
    json_match = re.search(r'<<<JSON>>>\n(.*?)\n<<<END>>>', result, re.DOTALL)

    if json_match:
        # print(json.loads(json_match.group(1)))
        return json.loads(json_match.group(1))
    else:
        raise ValueError("LLM 输出未包含预期的 JSON 格式数据。")
def llmapi(text):
    system_prompt = (
        "You are an experienced geographer. Your task is to determine the correct sequence of positioning functions and their inputs based on a given piece of natural language.\n"
        "The positioning functions you can choose from are:\n"
        "1. Relative Positioning: Inputs is (location coordinate or location name, direction, and distance). Outputs the coordinates that are in the given 'direction' and 'distance' from the input location.\n"
        "2. Between Positioning: Inputs is (location 1 coordinates or location 1 name, location 2 coordinates or location 2 name). Outputs the midpoint coordinate between the two locations.\n"
        "You can only use the given functions, and the inputs to the functions must obey the above properties. The given functions can be combined to solve complex situations."
        "First, perform chain-of-thought (CoT) reasoning, and finally output your answer in JSON format, wrapped between `<<<JSON>>>` and `<<<END>>>`.\n"
        "Make sure all inputs only include: location names (strings), step indices (integers), directions (strings, must be in English), or distances (strings with units). Do not return expressions like 'the coordinate 4 km south of Chatswood'.\n"
        "Each step must have an 'id'. If the input of a step is the output of a previous step, use that step’s 'id' as the input.\n"
        "All directions must be in English (e.g., south, west, northeast, etc.).\n"
        "Example output:\n"
        "<<<JSON>>>\n"
        "[{\"id\": 1, \"function\": \"Relative\", \"inputs\": [\"Chatswood\", \"south\", \"4 km\"]},"
        "{\"id\": 2, \"function\": \"Relative\", \"inputs\": [\"North Sydney\", \"west\", \"2 km\"]},"
        "{\"id\": 3, \"function\": \"Between\", \"inputs\": [1, 2]},"
        "{\"id\": 4, \"function\": \"Relative\", \"inputs\": [3, \"southwest\", \"5 km\"]}]\n"
        "<<<END>>>")

    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": text},
    ]

    chat_completion = client.chat.completions.create(
        messages=messages,
        model=model,
    )

    result = chat_completion.choices[0].message.content
    print(result)
    json_match = re.search(r'<<<JSON>>>\n(.*?)\n<<<END>>>', result, re.DOTALL)

    if json_match:
        return json.loads(json_match.group(1))
    else:
        raise ValueError("LLM 输出未包含预期的 JSON 格式数据。")





def execute_steps(steps):
    data = {}

    for step in steps:
        step_id = step['id']
        function = step['function']
        inputs = step['inputs']
        # print('-' * 50)
        # print(function)
        # print(inputs)


        resolved_inputs = []
        for inp in inputs:
            if isinstance(inp, int):
                resolved_inputs.append(data[inp])
            else:
                resolved_inputs.append(inp)
        if function == "Relative":
            location, direction, distance = resolved_inputs
            if isinstance(location, str):
                location = get_coordinates(location)

            location = [to_standard_2d_list(location[0])] + list(location[1:])
            location = [[[151.214901,-33.859175]], (151.214901,-33.859175)]
            result = get_level3_coordinates(location, distance, direction)
            data[step_id] = result

        elif function == "Between":


            location1, location2 = resolved_inputs
            # print(location1)
            # print(111)
            # print(location2)
            if isinstance(location1, str):
                location1 = get_coordinates(location1)

                location1 = [to_standard_2d_list(location1[0])] + list(location1[1:])
            if isinstance(location2, str):

                location2 = get_coordinates(location2)
                location2 = [to_standard_2d_list(location2[0])] + list(location2[1:])
            result = get_between_coordinates(location1, location2)

            data[step_id] = result

    return data


def process_api(input_text):
    # 这里编写实际的后端处理逻辑

    # return {
    #     "status": "success",
    #     # "result": f"Processed: {input_text.upper()}",
    #     "result": f"Processed: {nlp(input_text).to_json()}",
    #     "timestamp": time.time()
    # }
    parsed_steps = llmapi(input_text)
    result = execute_steps(parsed_steps)
    coords = result[(max(result.keys()))]

    geojson = get_geojson(None, coords[0], coords[1])
    return geojson

def process_api(input_text):
    
    return get_coordinates(input_text)

# get_coordinates(location)

request_url = 'https://nominatim.openstreetmap.org/search.php?q=Glebe&polygon_geojson=1&accept-language=en&format=jsonv2'
headers = {
    "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/18.3 Safari/605.1.15"
}
page1 = requests.get(request_url, headers=headers, verify=False)
cont = page1.content

# 设置API格式为JSON
gr.Interface(
    fn=process_api,
    # fn=cont,
    inputs="text",
    outputs="json",
    title="Backend API",
    allow_flagging="never"
).launch(debug=True)