find an answer for task3

A/3/dragon.py

@@ -27,7 +27,7 @@ class Dragon:
     return (self.kAlpha * theta * mp.cos(theta), self.kAlpha * theta * mp.sin(theta))
 
   def GenerateFirstNodeTheta(self, delta_theta):
-    return kInnerCircleRadius/self.kAlpha
+    return kInnerCircleRadius/self.kAlpha + delta_theta
 
   def GenerateFollowNodeTheta(self, cur_node_theta, expected_distance):
     cur_node_dot = self.Theta2Dot(cur_node_theta)

A/3/full_valid_test.py

@@ -0,0 +1,30 @@
+from dragon import *
+kPitchToTest=0.450338
+kDeltaThetaBeg=0
+kDeltaThetaEnd=2*2*3.1415926
+kTotalSteps=100000
+kStepDeltaTheta=(kDeltaThetaEnd-kDeltaThetaBeg)/kTotalSteps
+kParallelNum=24
+tasks_list=[i for i in np.arange(kDeltaThetaBeg, kDeltaThetaEnd, kStepDeltaTheta)]
+task_list_per_process=[tasks_list[i::kParallelNum] for i in range(kParallelNum)]
+print(f"len(task_list_per_thread)={len(task_list_per_process)}",file=sys.stderr)
+def ProcessEntryPoint(arg):
+  dragen = Dragon(mp.mpf(kPitchToTest)/(2*mp.pi))
+  delta_theta_list, process_id = arg
+  logf=open(f"sufficiency_test_{process_id}.log","w")
+  print(f"calculating delta_theta_list={delta_theta_list} with process_id={process_id}",file=logf)
+  for delta_theta in delta_theta_list:
+    cur_status=CheckCollision(status2blocks(dragen.CalcMoveList(delta_theta)))
+    if cur_status>=0:
+      return False # Error
+  return True # OK
+
+if __name__ == "__main__":
+  manager = multiprocessing.Manager()
+  task_args_list = [(task_list_per_process[i], i) for i in range(kParallelNum)]
+  with multiprocessing.Pool(processes=kParallelNum) as pool:
+    results=pool.map(ProcessEntryPoint, task_args_list)
+  if False in results:
+    print("Error")
+  else:
+    print("OK")

A/3/low_bound_cal.py

@@ -3,9 +3,9 @@ from dragon import *
 kBegPitch = 0.3
 kEndPitch = 0.55
 kTotalSteps = 10000
-kStepAlpha = (kEndPitch - kBegPitch) / kTotalSteps
+kStepPitch = (kEndPitch - kBegPitch) / kTotalSteps
 kParallelNum=24
-tasks_list = [kBegPitch + kStepAlpha * i for i in range(kTotalSteps)]
+tasks_list = [kBegPitch + kStepPitch * i for i in range(kTotalSteps)]
 task_list_per_process=[tasks_list[i::kParallelNum] for i in range(kParallelNum)]
 print(f"len(task_list_per_thread)={len(task_list_per_process)}",file=sys.stderr)
 def ProcessEntryPoint(arg):

A/3/sufficiency_test.py

@@ -1 +1,50 @@
-from dragon import *
+from dragon import *
+kBegPitch = 0.4503
+kEndPitch = 0.4504
+kTotalSteps = 100
+kStepPitch = (kEndPitch - kBegPitch) / kTotalSteps
+kParallelNum=24
+tasks_list = [kBegPitch + kStepPitch * i for i in range(kTotalSteps)]
+task_list_per_process=[tasks_list[i::kParallelNum] for i in range(kParallelNum)]
+
+kDeltaThetaBeg=0
+kDeltaThetaEnd=5*2*3.1415926
+kStepDeltaTheta=(kDeltaThetaEnd-kDeltaThetaBeg)/1000
+print(f"len(task_list_per_thread)={len(task_list_per_process)}",file=sys.stderr)
+def ProcessEntryPoint(arg):
+  pitch_list, process_id, shared_dict, lock = arg
+  logf=open(f"sufficiency_test_{process_id}.log","w")
+  print(f"calculating pitch_list={pitch_list} with process_id={process_id}",file=logf)
+  cnt = 0
+  tmp_res={}
+  for pitch in pitch_list:
+    dragen = Dragon(mp.mpf(pitch)/(2*mp.pi))
+    status = 1 # OK
+    for delta_theta in np.arange(kDeltaThetaBeg, kDeltaThetaEnd, kStepDeltaTheta):
+      cur_status=CheckCollision(status2blocks(dragen.CalcMoveList(delta_theta)))
+      if cur_status>=0:
+        status=-1
+        break
+    tmp_res[pitch]=status
+    print(f"res={status}",file=logf)
+    if status == 1:
+      break
+  with lock:  # 添加锁保护对共享字典的操作
+    shared_dict.update(tmp_res)
+
+if __name__ == "__main__":
+  manager = multiprocessing.Manager()
+  shared_dict = manager.dict()  # 创建一个共享字典
+  lock = manager.Lock()  # 创建一个共享的锁
+  task_args_list = [(task_list_per_process[i], i, shared_dict, lock) for i in range(kParallelNum)]
+  with multiprocessing.Pool(processes=kParallelNum) as pool:
+    pool.map(ProcessEntryPoint, task_args_list)
+  print(f"\nFinal Results: \n")
+  res_array=sorted(shared_dict.items())
+  for item in res_array:
+    time_point, status = item
+    print(f"Pitch={time_point}",end=" ")
+    if status==1:
+      print("OK")
+    else:
+      print("Error")