utils/llm_coding.py

import requests
import urllib3
import json
from utils import geoutil
import regex_spatial
from shapely.geometry import Polygon, MultiPoint, LineString, Point, mapping
import re
import geopandas as gpd
from geocoder import geo_level1
from openai import OpenAI
from utils.config import api_key
import numpy as np


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)
    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)
    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']
    else:
        print(json_content[0]['geojson']['type'])
    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):
    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)
    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:
        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
        new_center = (center[0] + dx, center[1] + dy)
        r_km = 1

        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

    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

# between
def get_between_coordinates(coordinates1, coordinates2):

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

    coords1, center1 = coordinates1
    coords2, center2 = coordinates2

    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
    )

    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]))

    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


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

    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("The LLM output does not contain the expected JSON formatted data. Please try again.")





def execute_steps(steps):
    data = {}
    locations_history = []
    for step in steps:
        step_id = step['id']
        function = step['function']
        inputs = step['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)
                locations_history.append(location)
            location = [to_standard_2d_list(location[0])] + list(location[1:])
            result = get_level3_coordinates(location, distance, direction)
            locations_history.append(result)
            data[step_id] = result
        elif function == "Between":
            location1, location2 = resolved_inputs

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

                location2 = get_coordinates(location2)
                locations_history.append(location2)
                location2 = [to_standard_2d_list(location2[0])] + list(location2[1:])
            result = get_between_coordinates(location1, location2)
            locations_history.append(result)
            data[step_id] = result

    return [data, locations_history]



if __name__ == '__main__':
    parsed_steps = []
    step_loc = execute_steps(parsed_steps)
    result = step_loc[0]