128 lines
5.5 KiB
Python
128 lines
5.5 KiB
Python
"""Collect a few bearings for Q1 from an ALREADY OPEN practice session.
|
|
|
|
There is deliberately no test-start API, formal mode, or /clear call here.
|
|
The documented robot API does not expose mode: verify the UI before running.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import math
|
|
from pathlib import Path
|
|
import time
|
|
import uuid
|
|
from urllib.error import URLError
|
|
from urllib.request import Request, urlopen
|
|
|
|
from mcm26b import Measurement, solve
|
|
|
|
|
|
class PracticeClient:
|
|
def __init__(self, robot_id, output, base_url='http://127.0.0.1:2026'):
|
|
self.robot_id = robot_id
|
|
self.output = output
|
|
self.base_url = base_url.rstrip('/')
|
|
self.prefix = uuid.uuid4().hex[:12]
|
|
self.count = 0
|
|
self.events = []
|
|
self.deadline = None
|
|
|
|
def action(self, path, **extra):
|
|
if path not in ('/enter', '/measure', '/exit'):
|
|
raise ValueError('采集器只允许进入、检测、退出')
|
|
if self.deadline is not None and path != '/exit' and time.monotonic() > self.deadline-15:
|
|
raise TimeoutError('剩余时间不足，预留退出时间')
|
|
self.count += 1
|
|
payload = dict(arena_id='default', robot_id=self.robot_id,
|
|
request_id=f'q1-{self.prefix}-{self.count}', **extra)
|
|
encoded = json.dumps(payload, allow_nan=False).encode('utf-8')
|
|
for attempt in range(3):
|
|
request = Request(self.base_url+path, data=encoded,
|
|
headers={'Content-Type':'application/json'}, method='POST')
|
|
try:
|
|
with urlopen(request, timeout=5) as response:
|
|
status = response.status
|
|
body = json.load(response)
|
|
# A retried action is recorded/consumed only once.
|
|
self.events.append(dict(path=path, request=payload, response=body,
|
|
retries=attempt))
|
|
self.output.write_text(
|
|
json.dumps(self.events,ensure_ascii=False,indent=2),encoding='utf-8')
|
|
if status != 200 or body.get('accepted') is not True:
|
|
raise RuntimeError(f'动作未被接受: {status}, {body}')
|
|
if path == '/enter':
|
|
self.deadline = time.monotonic()+body['remaining_real_duration_s']
|
|
return body
|
|
except URLError:
|
|
if attempt == 2:
|
|
raise
|
|
time.sleep(.25)
|
|
raise URLError('网络重试次数已用尽')
|
|
|
|
def measure(self, p, channel):
|
|
return self.action('/measure',position={'x':p[0],'y':p[1]},channel=channel)
|
|
|
|
|
|
def main():
|
|
parser=argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--robot-id',required=True)
|
|
parser.add_argument('--practice-confirmed',action='store_true',required=True,
|
|
help='仅在界面确认为问题3演练且接口已就绪时提供')
|
|
parser.add_argument('--case-code',required=True,help='界面显示的演练案例编码')
|
|
parser.add_argument('--output',type=Path,default=Path('output/q1/practice'))
|
|
args=parser.parse_args()
|
|
args.output.mkdir(parents=True,exist_ok=True)
|
|
log_path=args.output/f'{args.case_code}_actions.json'
|
|
if log_path.exists():
|
|
parser.error('同名采集日志已存在，请使用新的输出目录，避免覆盖证据')
|
|
client=PracticeClient(args.robot_id,log_path)
|
|
observations=[]
|
|
channel=None
|
|
first=None
|
|
entered=False
|
|
repeat=None
|
|
try:
|
|
client.action('/enter')
|
|
entered=True
|
|
for p in [(0,0),(800,0),(-800,0),(0,800),(0,-800)]:
|
|
for c in range(1,21):
|
|
response=client.measure(p,c)
|
|
if response['measure_result']=='direction':
|
|
first=(p,response['svd_deg'])
|
|
observations.append(Measurement(*p,response['svd_deg']))
|
|
channel=c
|
|
break
|
|
if channel is not None:
|
|
break
|
|
if channel is None:
|
|
raise RuntimeError('预设采样点未取得direction；本次不扩展为全域搜索')
|
|
p,theta=first
|
|
repeat=client.measure(p,channel)
|
|
a=math.radians(theta)
|
|
# For an omnidirectional source seen initially at <=1500m, these
|
|
# 700m-forward, +/-200m-lateral points remain within 1000m of its wedge.
|
|
for lateral in [200,-200,300,-300]:
|
|
q=(p[0]+700*math.cos(a)-lateral*math.sin(a),
|
|
p[1]+700*math.sin(a)+lateral*math.cos(a))
|
|
response=client.measure(q,channel)
|
|
if response['measure_result']=='direction':
|
|
observations.append(Measurement(*q,response['svd_deg']))
|
|
if len(observations)>=3:
|
|
break
|
|
data={'name':'官方问题3演练观测','source':'official_practice','case_code':args.case_code,
|
|
'channel':channel,'ground_truth':None,
|
|
'measurements':[m.__dict__ for m in observations],
|
|
'repeated_station_response':repeat,'result':solve(observations),
|
|
'scope':'仅采集问题一观测，不执行问题三全清除任务；没有调用clear'}
|
|
path=args.output/f'{args.case_code}_observations.json'
|
|
path.write_text(json.dumps(data,ensure_ascii=False,indent=2,allow_nan=False),
|
|
encoding='utf-8')
|
|
print(json.dumps({'observations':str(path),'count':len(observations),
|
|
'result':data['result']},ensure_ascii=False,indent=2))
|
|
finally:
|
|
if entered:
|
|
client.action('/exit')
|
|
|
|
|
|
if __name__=='__main__':
|
|
main()
|