ATLAS Offline Software
Loading...
Searching...
No Matches
test_magFieldCondAlg.py
Go to the documentation of this file.
1#!/usr/bin/env python
2#
3# Copyright (C) 2002-2024 CERN for the benefit of the ATLAS collaboration
4#
5# Test magnetic field conditions algs with varying currents.
6#
7# Folder name
8import sys
9from MagFieldServices.createDBForTest import createDB
10folder = '/EXT/DCS/MAGNETS/SENSORDATA'
11sqlite = 'magfieldForTest.db'
12
13
14# Testing IOVs and currents: (since LB, solenoid, toroids)
15# Default test - should read both mag field files, and turn off fields for events 5 to 9, and back on for 10 to 14
16currents = [(0, 7730, 20400),
17 (5, 0, 0),
18 (10, 7730, 20400)]
19
20# Optional test: only toroid is on for whole run. Scale factor for solenoid will become 1 at event 5,
21# but solenoid field will still be off
22# currents = [(0, 0, 20400),
23# (5, 7730, 20400),
24# (10, 7730, 20400)]
25
26# Optional test: only solenoid is on for whole run. Scale factor for toroid will become 1 at event 5,
27# but toroid field will still be off
28# currents = [(0, 7730, 0),
29# (5, 7730, 20400),
30# (10, 7730, 20400)]
31
32
33# Create sqlite file with DCS currents
34createDB(folder, sqlite, currents)
35
36from MagFieldServices.MagFieldServicesConfig import AtlasFieldCacheCondAlgCfg
37from AthenaConfiguration.MainServicesConfig import MainEvgenServicesCfg
38from AthenaConfiguration.ComponentFactory import CompFactory
39from AthenaConfiguration.AllConfigFlags import initConfigFlags
40
41flags = initConfigFlags()
42flags.Input.Files = []
43flags.Concurrency.NumThreads = 1
44flags.Exec.MaxEvents = currents[-1][0]+5 # 5 events per IOV
45flags.IOVDb.GlobalTag = 'CONDBR2-BLKPA-RUN2-01'
46flags.IOVDb.SqliteInput = sqlite
47flags.IOVDb.SqliteFolders = (folder,)
48flags.lock()
49
50acc = MainEvgenServicesCfg(flags)
51acc.getService('EventSelector').EventsPerLB = 1
52
53acc.merge(AtlasFieldCacheCondAlgCfg(flags, LockMapCurrents=False))
54
55acc.addEventAlgo(CompFactory.MagField.CondReader('MagFieldCondReader'))
56
57sys.exit(acc.run().isFailure())