Coverage for paintmaskanimator/auto_routing.py: 100%

1"""転送・集計・結果回収まで含めたCPU/GPUの実測に基づく振り分け。""" 

2 

3 

4class PipelinePolicy: 

5 def __init__(self, budget=8): 

6 self.records = {} 

7 self.cpu_seconds = 0.0 

8 self.gpu_seconds = 0.0 

9 self.budget = budget 

10 

11 def choose(self, key): 

12 record = self.records.get(key) 

13 if record is not None: 

14 return record[0] 

15 related = [(candidate, item) for candidate, item in self.records.items() if candidate[0] == key[0]] 

16 if len(related) < self.budget: 

17 return None 

18 # 新しい寸法のたびに二重計算を続けない。同種・近い仕事量だけ実測を引き継ぐ。 

19 def distance(candidate): 

20 if key[0] == 'alignment': 

21 if candidate[1] != key[1]: 

22 return 100 

23 return abs(candidate[2] - key[2]) 

24 if key[0] == 'sample' and candidate[3:] != key[3:]: 

25 return 100 

26 return sum(abs(a-b) for a, b in zip(candidate[1:3], key[1:3])) 

27 closest, result = min(related, key=lambda pair: distance(pair[0])) 

28 return result[0] if distance(closest) <= 1 else False 

29 

30 def record(self, key, cpu_seconds, gpu_seconds, *, valid=True): 

31 # わずかな差や計測誤差でGPUへ移さない。CPUの並列実行時間と比較する。 

32 use_gpu = valid and gpu_seconds < cpu_seconds * 0.9 

33 self.records[key] = use_gpu, cpu_seconds, gpu_seconds 

34 self.cpu_seconds += cpu_seconds 

35 self.gpu_seconds += gpu_seconds 

36 return use_gpu