Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use task_start method in autotest
Best Python code snippet using autotest_python
Temps.py
Source:Temps.py 
1import pandas as pd2import numpy as np3def work_schedule(nb_days = 365, day_start = 8, day_end = 24, month_start = 1,work_on_weekend = False):4 """5 Paramètres:6 nb_days:7 Nombre de jour de la simulation.8 (int: 1 à 365)9 day_start:10 Heure de début de la journée de travail.11 (int: 0 à 24)12 day_end:13 Heure de fin de la journée de travail.14 (int: 0 à 24)15 month_start:16 Mois de début de simulation.17 (int: 1 à 12)18 19 Sortie:20 Dataframes sur les heures de travail (0/1), journées de travail (0/1) et mois (1 à 12).21 """22 23 columns = ["work_time", "work_day", "month"]24 df= pd.DataFrame(columns = columns, index = range(nb_days * 24))25 26 # Heures de travail27 df.work_time = list(range(1,25)) * nb_days28 df.loc[(df.work_time < day_start+1) | (df.work_time >= day_end+1), "work_time"] = False29 df.loc[df.work_time != False, "work_time"] = True30 df = df.replace({True: 1, False: 0})31 32 # Semaine et fin de semaine33 df.work_day = np.ceil((df.index+1)/24)34 if work_on_weekend :35 df["work_day"] = 136 else : 37 df.work_day = np.where((df.work_day % 7 == 0) | ((df.work_day + 1) % 7 == 0), 0, 1)38 df.loc[df.work_day == 0, "work_time"] = 039 40 # Mois41 month_order = list(range(month_start,13)) + list(range(1, month_start))42 month_dict = {1:31, 2:28, 3:31, 4:30, 5:31, 6:30, 7:31, 8:31, 9:30, 10:31, 11:30, 12:31}43 month_length = np.array([month_dict[m] for m in month_order])*2444 for i in range(len(month_order)):45 idx = df.loc[df.month.isna(), "month"].index[:month_length[i]]46 df.loc[idx, "month"] = month_order[i]47 df.month = df.month.astype(int)48 49 return df50def task_total_length(df, task_start, task_time):51 """52 Paramètres:53 df:54 Dataframe sur les horaires de travail.55 task_start:56 Heure de début de la tâche.57 (float: 0 à len(df))58 task_time:59 Durée de la tâche.60 (float)61 62 Sortie:63 Durée finale de la tâche avec les arrêts de travail.64 """65 66 # Ramener le calcul à partir de la première semaine pour être certain67 # de ne pas défoncé l'année... Permet de réutiliser directement le code déjà 68 # fonctionnel pour 1 an, mais pourrait être à revoir si la gestion de semaines/mois69 # apporteraient des nuances...70 if task_start > 7 * 24 : 71 return task_total_length(df,task_start - int(task_start / (7*24))*(7*24),task_time)72 73 74 length = pd.Index(df.loc[int(task_start):, "work_time"].cumsum()).get_loc(int(task_time))75 76 # length est un slice plutôt qu'un int quand la dernière heure de travail est avant un arrêt de travail77 # Si task_start est une valeur sans décimale, la tâche se termine avant l'arrêt78 # Sinon, elle se termine au retour79 if isinstance(length, slice):80 if (task_start == int(task_start)) and (task_time == int(task_time)):81 length = length.start + 182 else:83 length = length.stop84 else:85 length += 186 87 # Ajustement de décimale si la tâche débute quand il n'y a pas de travail88 if df.loc[int(task_start), "work_time"] == 0:89 length = length - (task_start - int(task_start))90 91 # Ajustement de décimale si la durée de la tâche a une décimale92 length = length + (task_time - int(task_time))93 return length94def GetInfosTemps(now) : 95 """96 Paramètres:97 now:98 Temps en heure (réel)99 100 Sortie:101 day_of_week : jour de la semaine (1 à 7)102 hour : heure de la journée (réel)103 """104 105 day_of_week = ((int(now / 24)) % 7) + 1 106 hour = now % 24 107 108 return day_of_week, hour109def HeuresProductives(df,debut,fin):110 """111 Paramètres:112 df:113 Dataframe sur les horaires de travail.114 debut:115 Heure de début du calcul116 fin:117 Fin de la tâche.118 119 Sortie:120 Durée en heures (réel) de travail productive entre debut et fin121 """122 # Si l'heure de début excéde l'heure de fin, il n'y a pas d'heures productives123 # (Cette situation pourrait survenir si on ne gère pas bien certaines transitions entre le 124 # régime transitoire et le permanent. Les indicateurs comptes à partir du régime permanent125 # et on se compare au now)126 if debut >= fin : 127 return 0128 # Ramener le calcul à partir de la première semaine pour être certain129 # de ne pas défoncé l'année... Permet de réutiliser directement le code déjà 130 # fonctionnel pour 1 an, mais pourrait être à revoir si la gestion de semaines/mois131 # apporteraient des nuances...132 if debut > 7 * 24 : 133 return HeuresProductives(df,debut - int(debut / (7*24))*(7*24),fin - int(debut / (7*24))*(7*24))134 # Retirer les semaines complètes pour ne pas défoncer l'année de calcul135 if (fin - debut) > 24 * 7 : 136 NbHeuresUneSemaine = sum(df[:168]["work_time"])137 NbSemainesComplètes = int((fin-debut) / (24*7))138 NbHeuresSemainesComplètes = NbSemainesComplètes * NbHeuresUneSemaine139 DureeSemIncomplete = HeuresProductives(df,debut,fin - NbSemainesComplètes*(7*24))140 return NbHeuresSemainesComplètes + DureeSemIncomplete141 142 nbHeures = sum(df[int(debut):int(fin)]["work_time"])143 MinutesDebut = (debut - int(debut)) * df.iloc[int(debut)]["work_time"]144 MinutesFin = (fin - int(fin)) * df.iloc[int(fin)]["work_time"]145 146 return nbHeures - MinutesDebut + MinutesFin 147 148if __name__ == "__main__":149 150 151 df = work_schedule(nb_days = 365, day_start = 0, day_end = 24, month_start = 1,work_on_weekend=False)152 task_length = task_total_length(df, task_start = 360*24, task_time = 23)153 #print(task_length)154 155 if 1 ==0 : 156 # Pour faciliter le développement, on s'assure d'avoir toujours le mêmes157 # nombres aléatoires d'une exécution à l'autre158 import random159 random.seed(1)160 161 now = 0162 for jour in range(1,5000) : 163 print(jour)164 for heure in range(1) : 165 for minutes in range(1) : 166 task_start = jour*24 + heure + minutes/60167 task_time = random.random() * 1500 + 1168 169 #bk = task_total_length_bk(df, task_start = jour*24 + heure + minutes/60, task_time = 50)170 new = task_total_length(df, task_start = task_start, task_time = task_time)171 end = new + task_start172 173 prod = HeuresProductives(df,task_start,end)174 175 176 if round(prod,4) != round(task_time,4) : 177 #print("PROBLÃME", task_start, task_time, new,prod)178 # task_start = jour*24 + heure + minutes/60179 180 # task_time = 50181 182 # print(jour,heure,minutes,"PROBLÃME")183 # print(bk)184 # print(new)185 print("task start", task_start, "task time", task_time)186 print("new",new)187 print("prod",prod)188 print("end",end)189 # print(task_start - int(task_start / (7*24))*(7*24))190 exit191 192 print("FINI") 193 # print(now, jour, heure)194 #print("now : ", now,"jour : ", jour)195 196 #month, day_of_week, hour = GetInfosTemps(now)197 198 #print(day_of_week)199 #print()200 201 #now += 1 * 24202 debut = 189.17756324661573 203 fin = debut-5204 print("Calculer",HeuresProductives(df,debut,fin))205 print("Bon 24/7", fin-debut)...
runner.py
Source:runner.py
1from copy import deepcopy2from cbs import CBSSolver3from prioritised import PrioritizedPlanningSolver4from visualize import Animation5from path_astar_one import get_sum_of_cost, path6import itertools7import heapq8def min_cost_task_assn(starts,tasks):9 _tasks = list(itertools.permutations(tasks))10 _len = len(starts)11 starts_set=[]12 tasks_set=[]13 for t in _tasks:14 starts_set.append(starts)15 tasks_set.append(list(t))16 paths_with_cost = []17 for i, j in zip(starts_set, tasks_set):18 cbs = CBSSolver(my_map, i, j)19 _paths = cbs.find_solution(True, CBSSolver.NORMAL)20 _cost = get_sum_of_cost(_paths)21 paths_with_cost.append((_paths, _cost))22 _min_path = min(paths_with_cost, key=lambda x: x[1])23 return _min_path[0],_min_path[1]24def read_input(filename):25 f = open(filename, 'r')26 line = f.readline()27 rows, columns = [int(x) for x in line.split(' ')]28 rows = int(rows)29 columns = int(columns)30 my_map = []31 for r in range(rows):32 line = f.readline()33 my_map.append([])34 for cell in line:35 if cell == '@' or cell == 'T':36 my_map[-1].append(True)37 elif cell == '.':38 my_map[-1].append(False)39 line = f.readline()40 num_agents = int(line)41 agent_loc=[]42 task_start=[]43 task_goal=[]44 #print(num_agents)45 for a in range(num_agents):46 line=f.readline()47 sx, sy = [int(x) for x in line.split(' ')]48 agent_loc.append((sx,sy))49 line = f.readline()50 num_tasks = int(line)51 for a in range(num_tasks):52 line = f.readline()53 sx, sy, gx, gy = [int(x) for x in line.split(' ')]54 task_start.append((sx,sy))55 task_goal.append((gx,gy))56 line = f.readline()57 aaa,ttt=[],[]58 if(int(line)>0):59 for i in range(int(line)):60 line = f.readline()61 x,y = [int(x) for x in line.split(' ')]62 aaa.append(x)63 ttt.append(y)64 f.close()65 return my_map,agent_loc,task_start,task_goal,aaa,ttt66print("Case 1: PickUp And Delivery of Single Task")67my_map, agent_loc,task_start, task_goal,a,t = read_input('./cases/test_1.txt')68print("Total Agents = {}".format(len(agent_loc)))69print("Total Tasks = {}".format(len(task_start)))70paths1,cost = min_cost_task_assn(agent_loc, task_start)71ttt=[]72for i in range(len(paths1)):73 ttt.append(paths1[i][-1])74cbs = CBSSolver(my_map, ttt, task_goal)75paths2 = cbs.find_solution(True, CBSSolver.NORMAL)76cost = get_sum_of_cost(paths2)77for i in range(len(paths2)):78 paths1[i]=paths1[i]+paths2[i]79print("***Test paths on a simulation***")80print('Tasks Completed')81animation = Animation(my_map, agent_loc, task_goal, paths1,task_start,task_goal)82animation.show()83print("Case 2: PickUp And Delivery Based on Weight of Task and Capacity of Agents")84my_map, agent_loc,task_start, task_goal,a,t = read_input('./cases/test_2.txt')85print("Total Agents = {}".format(len(agent_loc)))86print("Total Tasks = {}".format(len(task_start)))87prep_a=[]88for i in range(len(a)):89 prep_a.append((a[i],i))90prep_t=[]91for i in range(len(t)):92 prep_t.append((t[i],i))93heapq.heapify(prep_a)94heapq.heapify(prep_t)95fff,ggg,hhh=[],[],[]96aasn_task,rem_task=[],[]97for i in range(len(agent_loc)):98 zz=heapq.heappop(prep_a)99 yy=heapq.heappop(prep_t)100 if(zz[0]>=yy[0]):101 fff.append(agent_loc[zz[1]])102 ggg.append(task_start[yy[1]])103 hhh.append(task_goal[yy[1]])104 aasn_task.append(zz[1])105 else:106 break107for j in range(len(task_start)):108 if j not in aasn_task:109 rem_task.append(task_start[j])110print("Total Tasks Allocated Based on Agent Capacity= {}".format(len(fff)))111paths1,cost = min_cost_task_assn(fff, ggg)112cbs = CBSSolver(my_map, ggg, hhh)113paths2 = cbs.find_solution(True, CBSSolver.NORMAL)114cost = get_sum_of_cost(paths2)115for i in range(len(paths2)):116 paths1[i]=paths1[i]+paths2[i]117print("***Test paths on a simulation***")118print('Tasks Completed')119animation = Animation(my_map, fff, hhh, paths1,ggg,rem_task)120animation.show()121print("Case 3: PickUp And Delivery of Multiple Task")122my_map, agent_loc,task_start, task_goal,a,t = read_input('./cases/test_3.txt')123print("Total Agents = {}".format(len(agent_loc)))124print("Total Tasks = {}".format(len(task_start)))125cbs = CBSSolver(my_map, agent_loc, task_start[:len(agent_loc)])126paths1 = cbs.find_solution(True, CBSSolver.NORMAL)127cbs = CBSSolver(my_map, task_start[:len(agent_loc)], task_goal[:len(agent_loc)])128paths2 = cbs.find_solution(True, CBSSolver.NORMAL)129cost = get_sum_of_cost(paths2)130for i in range(len(paths2)):131 paths1[i]=paths1[i]+paths2[i]132time=len(task_start)//len(agent_loc)-1133temp=paths1134while(time>0):135 time-=1136 st=task_goal[:len(agent_loc)]137 tar=task_start[len(agent_loc):]138 cbs = CBSSolver(my_map, st, tar)139 paths1 = cbs.find_solution(True, CBSSolver.NORMAL)140 cbs = CBSSolver(my_map,task_start[len(agent_loc):],task_goal[len(agent_loc):])141 paths2 = cbs.find_solution(True, CBSSolver.NORMAL)142 cost = get_sum_of_cost(paths2)143 for i in range(len(paths2)):144 paths1[i]=paths1[i]+paths2[i] 145for i in range(len(paths1)):146 temp[i]=temp[i]+paths1[i]147print("***Test paths on a simulation***")148print('Tasks Completed')149animation = Animation(my_map, agent_loc, task_goal[len(agent_loc):], temp,task_start,task_goal[:len(agent_loc)])150animation.show()151print("Case 4: Miscellaneous Warehouse")152my_map, agent_loc,task_start, task_goal,a,t = read_input('./cases/misc.txt')153print("Total Agents = {}".format(len(agent_loc)))154print("Total Tasks = {}".format(len(task_start)))155paths1,cost = min_cost_task_assn(agent_loc, task_start)156ttt=[]157for i in range(len(paths1)):158 ttt.append(paths1[i][-1])159cbs = CBSSolver(my_map, ttt, task_goal)160paths2 = cbs.find_solution(True, CBSSolver.NORMAL)161cost = get_sum_of_cost(paths2)162for i in range(len(paths2)):163 paths1[i]=paths1[i]+paths2[i]164print("***Test paths on a simulation***")165print('Tasks Completed')166animation = Animation(my_map, agent_loc, task_goal, paths1,task_start,task_goal)167animation.show()168print("Case 5: Miscellaneous Warehouse 2 Using Prioritised")169my_map, agent_loc,task_start, task_goal,a,t = read_input('./cases/misc2.txt')170print("Total Agents = {}".format(len(agent_loc)))171print("Total Tasks = {}".format(len(task_start)))172solver = PrioritizedPlanningSolver(my_map, agent_loc, task_start)173paths1 = solver.find_solution()174solver = PrioritizedPlanningSolver(my_map, task_start, task_goal)175paths2= solver.find_solution()176for i in range(len(paths2)):177 paths1[i]=paths1[i]+paths2[i]178print("***Test paths on a simulation***")179print('Tasks Completed')180animation = Animation(my_map, agent_loc, task_goal, paths1,task_start,task_goal)...
run.py
Source:run.py
1import sys2from .mylib import *3from .settings import Settings4from .etcbc import Etcbc5from .laf import Laf6from .validate import Validate7from .transform import Transform8def init():9 global settings, val, et, lf, tr, prog_start, task_start10 settings = Settings()11 val = Validate(settings)12 et = Etcbc(settings)13 lf = Laf(settings, et, val)14 tr = Transform(settings, et, lf)15 prog_start = Timestamp()16 task_start = Timestamp()17def dotask(part): 18 print("INFO: Start Task {}".format(part))19 task_start = Timestamp()20 tr.transform(part)21 print("{} - {}".format(prog_start.elapsed(), task_start.elapsed()))22 print("INFO: End Task {}".format(part))23def final():24 task_start = Timestamp()25 lf.makeheaders()26 val.validate()27 val.report()28 lf.report()29 print("{} - {}".format(prog_start.elapsed(), task_start.elapsed()))30def processor():31 init()32 print("{} - {}".format(prog_start.elapsed(), task_start.elapsed()))33 print("INFO: Doing parts: {}".format(','.join(settings.given_parts)))34 for part in settings.given_parts: dotask(part)...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run autotest automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
