合并完成

2026-09-10 21:37:28 +08:00
parent c57865edf5
commit 21e8a8bf52
14 changed files with 0 additions and 152 deletions
@@ -1,123 +0,0 @@
-"""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 sys
-import time
-import uuid
-from urllib.error import URLError
-from urllib.request import Request, urlopen
-
-sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
-from geometry 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)
-
-    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()