2

下面是我的代码，我在位置变量中使用了经纬度坐标，并通过坐标字符串将其附加到 URL。因为我有 CSV 文件，其中包含许多位置的纬度和经度坐标，然后调用该 CSV 文件作为此 API 的输入（需要身份验证）。如何将 CSV 文件输入此代码而不是位置变量？

import requests
import pprint

locations = [(13.84, -12.57), (12.21, -14.69)]

coordinates_str = ','.join(map(lambda a: ' '.join(f'{f:.3f}' for f in a), locations))

# Replace "poi-settings" with the endpoint you would like to call.

URL = f'https://ubiconnect-eu.ubimet.com:8090/pinpoint-data?coordinates={coordinates_str}'
TOKEN = 'TOKEN KEY'

# Create session object that can be used for all requests.
session = requests.Session()
session.headers['Authorization'] = 'Token {token}'.format(token=TOKEN)

# Send GET request to UBIconnect.
res = session.get(URL)
res.raise_for_status()

# Decode JSON response.
poi_info = res.json()
pprint.pprint(poi_info, indent=2, compact=True)

然后我尝试了这种方式：代替了 coordinates_str 我这样做了

import requests
import pprint
import pandas as pd 

df = pd.read_csv(r'E:\route_points.csv')
print(df)


# Replace "poi-settings" with the endpoint you would like to call.
URL = f'https://ubiconnect-eu.ubimet.com:8090/pinpoint-data?'
TOKEN = 'API TOKEN'
params= {'coordinates':(df)}

# Create session object that can be used for all requests.
session = requests.Session()
session.headers['Authorization'] = 'Token {token}'.format(token=TOKEN)

# Send GET request to UBIconnect.
res = session.get(URL, params= params)
res.raise_for_status()

# Decode JSON response.
poi_info = res.json()
pprint.pprint(poi_info, indent=2, compact=True)

还是行不通。

从文档调用 API 所需的格式为：

# Replace "poi-settings" with the endpoint you would like to call.
URL = 'https://ubiconnect-eu.ubimet.com:8090/poi-settings'
TOKEN = '<YOUR TOKEN GOES HERE>'

所以我用 pinpoint-data 替换了 poi-settings

URL = 'https://ubiconnect-eu.ubimet.com:8090/pinpoint-data?coordinates=longitude<space<latitude'

例如：我将一个坐标集放入 API URL

URL = 'https://ubiconnect-eu.ubimet.com:8090/pinpoint-data?coordinates=132.85 12.84'

然后使用上面的 URL 获取该位置的天气数据。

4

2 回答 2

0

如果您只想从 CSV 文件一次提交一个坐标块，那么类似以下内容就足够了：

from itertools import islice
import requests
import pprint
import csv

def grouper(n, iterable):
    it = iter(iterable)
    return iter(lambda: tuple(islice(it, n)), ())


block_size = 10   # how many pairs to submit per request
TOKEN = 'TOKEN KEY'

# Create session object that can be used for all requests.
session = requests.Session()
session.headers['Authorization'] = 'Token {token}'.format(token=TOKEN)

with open('coordinates.csv', newline='') as f_input:
    csv_input = csv.reader(f_input)
    header = next(csv_input)        # skip the header
    
    for coords in grouper(block_size, csv_input):
        coordinates = ','.join(f'{float(long):.3f} {float(lat):.3f}' for long, lat in coords)
        print(coordinates)
        
        URL = f'https://ubiconnect-eu.ubimet.com:8090/pinpoint-data?coordinates={coordinates}'

        # Send GET request to UBIconnect.
        res = session.get(URL)
        res.raise_for_status()
        
        # Decode JSON response.
        poi_info = res.json()
        
        pprint.pprint(poi_info, indent=2, compact=True)

（显然这没有经过测试 - 没有令牌）。确保 CSV 文件中没有空行。

要输出到文件，请添加输出文件：

with open('coordinates.csv', newline='') as f_input, open('output.json', 'w', encoding='utf-8')  as f_output:

并在pprint()通话中使用它：

pprint.pprint(poi_info, f_output, indent=2, compact=True)
f_output.write('\n')    # add blank line if needed
于 2021-02-07T17:16:29.597 回答
0

希望这是您正在寻找的

import csv
locations = list()
with open("foo.csv") as csvf:
    csvreader = csv.DictReader(csvf)    
    for row in csvreader:
        locations.append((float(row["lat"]), float(row["long"])))
# now add your code
coordinates_str = ','.join(map(lambda a: ' '.join(f'{f:.3f}' for f in a), locations))
于 2021-02-07T18:11:18.560 回答