Best Python code snippet using molotov_python
experiment_runner.py
Source:experiment_runner.py
...105 polygon_9_gen = PolygonGenerator(radius, 9)106 107 scenario_id = 1108 109 def _scenario(**kwargs) -> TestScenarioSpec:110 nonlocal scenario_id111 scenario = TestScenarioSpec(112 id=scenario_id, common=common_spec, **kwargs113 )114 scenario_id += 1115 return scenario116 117 std_shape = dict(shape_gen=rectangle_gen, n_groups=6)118 all_affine = dict(use_translation=True, use_rotation=True, use_scale=True)119 120 scenarios = (121 _scenario(**std_shape),122 123 _scenario(use_translation=True, **std_shape),124 _scenario(use_rotation=True, **std_shape),125 _scenario(use_scale=True, **std_shape),126 127 _scenario(use_noise=False, **std_shape, **all_affine),128 _scenario(use_noise=True, **std_shape, **all_affine),129 130 _scenario(131 use_noise=True, shape_gen=polygon_5_gen, n_groups=6, **all_affine132 ),133 _scenario(134 use_noise=True, shape_gen=polygon_7_gen, n_groups=6, **all_affine135 ),136 _scenario(137 use_noise=True, shape_gen=polygon_9_gen, n_groups=6, **all_affine138 ),139140 _scenario(141 use_noise=True, shape_gen=rectangle_gen, n_groups=3, **all_affine142 ),143 _scenario(144 use_noise=True, shape_gen=rectangle_gen, n_groups=5, **all_affine145 ),146 _scenario(147 use_noise=True, shape_gen=rectangle_gen, n_groups=7, **all_affine148 ),149 _scenario(150 use_noise=True, shape_gen=rectangle_gen, n_groups=9, **all_affine151 ),152 )153 154 run_experiments(155 scenarios, config.RESULTS_FILE_PATH, config.N_INSTANCES_PER_SCENARIO156 )157 158 return 0159160161if __name__ == '__main__':
...
plot_fitCanvas.py
Source:plot_fitCanvas.py
1# script that prints all fit_canvas that can be found in all the passed root files and stores them with2# hopefully reasonable names3#4# TODO: this works quite nicely for the vtx and eta scenario already, however the pt_abseta scenario is5# currently a bit of a mess at the moment.6import os7import subprocess8import glob9import re10def saveCanvasOfFile(_file, _regex="fit_canvas", _extension="pdf"):11 """12 Saves all the TCanvas found in the file matching the passed regex via a call to the saveCanvas exectuable13 """14 path_to_exe="/afs/hephy.at/work/t/tmadlener/CMSSW_8_0_12/src/TnPfromJPsi/PlotEfficiency/utils/saveCanvas"15 DEVNULL = open(os.devnull, 'w') # needed to dump the output of TCanvas::Save into /dev/null16 print("Saving TCanvas matching \'{}\' from file \'{}\'".format(_regex, _file))17 status = subprocess.call([path_to_exe, _file, _regex, _extension], stdout=DEVNULL, stderr=subprocess.STDOUT)18 return status19def commonFileNameCleanup(_filename, _ID, _scenario, _trigRgx):20 """21 Do the cleanup that is common to both mvAndRename functions below:22 Remove trigger stuff and the TDirectory information from the pdf filename23 """24 fn = _trigRgx.sub('', _filename)25 fn = fn.replace(":tpTree:", "").replace("{}_{}:".format(_ID, _scenario), "")26 fn = fn.replace("_pair_drM1_bin0_", "").replace("_pair_probeMultiplicity_bin0_", "")27 return fn28def mvAndRename(_ID, _scenario, _targetdir="fitCanvasPdfs", _extension="pdf"):29 """30 The output of saveCanvas is not that nice and has to be cleaned up.31 Mainly doing some replacing of the trigger parts and moving the created files to a separate directory32 """33 triggerRegex = re.compile('(_tag)?_Mu7p5_Track2_Jpsi(_(TK|MU)_pass_)?')34 for filename in glob.glob(":tpTree:{}_{}*:*.{}".format(_ID, _scenario, _extension)):35 filenameTo = commonFileNameCleanup(filename, _ID, _scenario, triggerRegex)36 dirTo = "{}/{}_{}".format(_targetdir, _ID, _scenario)37 if not os.path.exists(dirTo):38 os.makedirs(dirTo)39 filenameTo = "/".join([dirTo, filenameTo])40 os.rename(filename, filenameTo)41def mvAndRenamePt(_ID, _scenario, _file, _targetdir="fitCanvasPdfs", _extension="pdf"):42 """43 The output of saveCanvas is not that nice and has to be cleaned up.44 Mainly doing some replacing of the trigger parts and moving the created files to a separate directory.45 For pt we currently run into a memory problem and have to split the input into abseta different bins.46 However this is not reflected in the names of the produced pdfs (since the .root files do not "know"47 about this splitting, so we have to define a special version for renaming the pt scenario that uses48 some more information that can be obtained from the root file (_file)49 """50 # compute some additional info from the filename (that follows at least for the moment a common pattern)51 addInfo = _file.replace(".root", "")52 addInfo = addInfo.replace("{}_{}_".format(_ID, _scenario), "")53 addInfo = addInfo.replace("TnP_MuonID_","").replace("_data_all_", "").replace("_signal_mc_", "")54 triggerRegex = re.compile('(_tag)?_Mu7p5_Track2_Jpsi(_(TK|MU)_pass_)?')55 for filename in glob.glob(":tpTree:{}_{}*:*.{}".format(_ID, _scenario, _extension)):56 filenameTo = commonFileNameCleanup(filename, _ID, _scenario, triggerRegex)57 dirTo = "{}/{}_{}_{}".format(_targetdir, _ID, _scenario, addInfo)58 if not os.path.exists(dirTo):59 os.makedirs(dirTo)60 filenameTo = "/".join([dirTo, filenameTo])61 os.rename(filename, filenameTo)62def processAllFiles(_dir, _ID, _scenario,63 _targetdir="fitCanvasPdfs",64 _canvasRegex="fit_canvas",65 _extension="pdf"):66 """67 Process all .root files matching the _ID AND _scenario in _dir.68 """69 os.chdir(_dir)70 for f in glob.glob("TnP_MuonID_*_{0}_{1}*.root".format(_ID, _scenario)):71 saveCanvasOfFile(f, _canvasRegex, _extension)72 if "pt_abseta" in _scenario:73 mvAndRenamePt(_ID, _scenario, f, _targetdir, _extension)74 else:75 mvAndRename(_ID, _scenario, _targetdir, _extension)76# Define on what to run77IDs = ["Loose2016", "Medium2016", "Tight2016", "Soft2016"]78scenarios = ["eta", "vtx", "pt_abseta"]79basedir="/afs/hephy.at/work/t/tmadlener/CMSSW_8_0_12/src/data_rootfiles"80targetdir="/afs/hephy.at/work/t/tmadlener/CMSSW_8_0_12/src/outputfiles/Figures/FitCanvasPdfs/data"81# pdf, ps and svg are currently working, png should too according to the ROOT documentation but doesn't at the moment82# Workaround to get png is to save the plots as ps and convert them to png using gs83plotformat="pdf"84for ID in IDs:85 for scen in scenarios:86 print("Currently processing ID: {}, scenario: {}".format(ID, scen))...
test_formatter_rerun.py
Source:test_formatter_rerun.py
...11 formatter_name = "rerun"12 def test_feature_with_two_passing_scenarios(self):13 p = self._formatter(_tf(), self.config)14 f = self._feature()15 scenarios = [ self._scenario(), self._scenario() ]16 for scenario in scenarios:17 f.add_scenario(scenario)18 # -- FORMATTER CALLBACKS:19 p.feature(f)20 for scenario in f.scenarios:21 p.scenario(scenario)22 assert scenario.status == "passed"23 p.eof()24 eq_([], p.failed_scenarios)25 # -- EMIT REPORT:26 p.close()27 def test_feature_with_one_passing_one_failing_scenario(self):28 p = self._formatter(_tf(), self.config)29 f = self._feature()30 passing_scenario = self._scenario()31 failing_scenario = self._scenario()32 failing_scenario.steps.append(self._step())33 scenarios = [ passing_scenario, failing_scenario ]34 for scenario in scenarios:35 f.add_scenario(scenario)36 # -- FORMATTER CALLBACKS:37 p.feature(f)38 for scenario in f.scenarios:39 p.scenario(scenario)40 failing_scenario.steps[0].status = "failed"41 assert scenarios[0].status == "passed"42 assert scenarios[1].status == "failed"43 p.eof()44 eq_([ failing_scenario ], p.failed_scenarios)45 # -- EMIT REPORT:46 p.close()47 def test_feature_with_one_passing_two_failing_scenario(self):48 p = self._formatter(_tf(), self.config)49 f = self._feature()50 passing_scenario = self._scenario()51 failing_scenario1 = self._scenario()52 failing_scenario1.steps.append(self._step())53 failing_scenario2 = self._scenario()54 failing_scenario2.steps.append(self._step())55 scenarios = [ failing_scenario1, passing_scenario, failing_scenario2 ]56 for scenario in scenarios:57 f.add_scenario(scenario)58 # -- FORMATTER CALLBACKS:59 p.feature(f)60 for scenario in f.scenarios:61 p.scenario(scenario)62 failing_scenario1.steps[0].status = "failed"63 failing_scenario2.steps[0].status = "failed"64 assert scenarios[0].status == "failed"65 assert scenarios[1].status == "passed"66 assert scenarios[2].status == "failed"67 p.eof()68 eq_([ failing_scenario1, failing_scenario2 ], p.failed_scenarios)69 # -- EMIT REPORT:70 p.close()71class TestRerunAndPrettyFormatters(MultipleFormattersTest):...
multi_agent_race.py
Source:multi_agent_race.py
1import math2from typing import Dict, Union3import gym4from .scenarios import MultiAgentScenario5from ..core.definitions import Pose, Velocity6class MultiAgentRaceEnv(gym.Env):7 metadata = {'render.modes': ['follow', 'birds_eye', 'lidar']}8 def __init__(self, scenario: MultiAgentScenario):9 self._scenario = scenario10 self._initialized = False11 self._time = 0.012 self.observation_space = gym.spaces.Dict([(k, a.observation_space) for k, a in scenario.agents.items()])13 self.action_space = gym.spaces.Dict([(k, a.action_space) for k, a in scenario.agents.items()])14 @property15 def scenario(self):16 return self._scenario17 def step(self, action: Dict):18 assert self._initialized, 'Reset before calling step'19 observations = {}20 dones = {}21 rewards = {}22 state = self._scenario.world.state()23 for id, agent in self._scenario.agents.items():24 observation, info = agent.step(action=action[id])25 state[id]['observations'] = observation26 done = agent.done(state)27 reward = agent.reward(state, action[id])28 observations[id] = observation29 dones[id] = done30 rewards[id] = reward31 self._time = self._scenario.world.update()32 return observations, rewards, dones, state33 def set_state(self, agent: str, pose: Pose):34 self._scenario.agents[agent].reset(pose=pose)35 def reset(self, mode: str = 'grid'):36 if not self._initialized:37 self._scenario.world.init()38 self._initialized = True39 observations = {}40 for agent in self._scenario.agents.values():41 obs = agent.reset(self._scenario.world.get_starting_position(agent=agent, mode=mode))42 observations[agent.id] = obs43 self._scenario.world.reset()44 self._scenario.world.update()45 return observations46 def render(self, mode='follow', agent: str = None, **kwargs):47 if mode not in MultiAgentRaceEnv.metadata['render.modes']:48 raise NotImplementedError(f'"{mode}" is no supported render mode. Available render modes: {MultiAgentRaceEnv.metadata["render.modes"]}')49 if agent is None:50 agent = list(self._scenario.agents.keys())[0]...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!