Source code for Gauss.Geometry
###############################################################################
# (c) Copyright 2000-2022 CERN for the benefit of the LHCb Collaboration #
# #
# This software is distributed under the terms of the GNU General Public #
# Licence version 3 (GPL Version 3), copied verbatim in the file "COPYING". #
# #
# In applying this licence, CERN does not waive the privileges and immunities #
# granted to it by virtue of its status as an Intergovernmental Organization #
# or submit itself to any jurisdiction. #
###############################################################################
__author__ = "Gloria Corti, Dominik Muller, and Michal Mazurek"
__email__ = "lhcb-simulation@cern.ch"
import logging
import os
from DDDB.Configuration import DDDBConf
from Gaudi.Configuration import log
from GaudiConf.SimConf import SimConf
from Gauss.Defaults import (
CONDITIONS_VERSIONS_DD4HEP,
DEFAULT_DD4HEP_BEAMPIPE_VERSION,
DEFAULT_DETECTORS,
GEOMETRY_VERSIONS_DD4HEP,
INCOMPATIBLE_DETECTORS,
NON_MONITORABLE_DETECTORS,
NON_SIMULATABLE_DETECTORS,
RUN3_DETECTORS_DETDESC,
RUN3_DETECTORS_DETDESC_NO_UT,
)
from Gaussino.Geometry import GaussinoGeometry
from Gaussino.Utilities import GaussinoConfigurable
# Configurables (do NOT use 'from Configurables' here)
from LHCbAlgs.Configuration import LHCbApp
[docs]class GaussGeometry(GaussinoConfigurable):
__required_configurables__ = [
"Gauss",
]
__used_configurables__ = [
LHCbApp,
SimConf,
# we have to make sure that DDDBConf
# does not overwrite DD4hep properties we set ourselves
DDDBConf,
]
GAUSSINO_GEOMETRY_OPTIONS = {
"ExportGDML": {},
"SensDetMap": {},
"ExtraGeoTools": [
"GiGaSetSimAttributes/GiGaSetSimAttributes",
"GiGaRegionsTool/GiGaRegionsTool",
],
"ExternalDetectorEmbedder": "",
}
__slots__ = {
"DetectorGeo": {},
"DetectorSim": {},
"DetectorMoni": {},
"Legacy": False, # True -> DetDesc, False -> DD4hep (default)
# DD4hep and DetDesc conditions and geometry version
"ConditionsLocation": "",
"ConditionsVersion": "",
"GeometryLocation": "",
"GeometryVersion": "",
**GAUSSINO_GEOMETRY_OPTIONS,
}
# internal options to be set by Gauss
datatype = False
run1or2 = False
run4or5 = False
only_generation_phase = False
# List of geometry objects which will be converted, it's content is used in
# GaussGeo or in GiGaInputStream if GiGaGeo is used for conversion
_listOfGeoObjects_ = []
# List of DD4hep XML includes
_listOfXMLIncludes_ = []
# TODO: refactor this
_det_pieces = {
"BeforeUpstreamRegion": [],
"UpstreamRegion": [],
"BeforeMagnetRegion": [],
"MagnetRegion": [],
"AfterMagnetRegion": [],
"DownstreamRegion": [],
"AfterDownstreamRegion": [],
}
_base_pieces = dict(_det_pieces)
_lhcb_dd4hep_det_map = {
# Common Detectors
"Magnet": "Magnet",
# Run3/4 Detectors
"VP": "VP",
"UT": "UT",
"FT": "FT",
"MP": "MP",
"Rich1Pmt": "Rich1",
"Rich2Pmt": "Rich2",
"Ecal": "Ecal",
"Hcal": "Hcal",
"Muon": "Muon",
# Run5 Detectors
"TV": "TV",
"UP": "UP",
}
[docs] def __apply_configuration__(self):
log.debug("Configuring GaussGeometry")
if GaussGeometry.only_generation_phase:
log.debug("-> Only the generation phase, skipping.")
return
self._check_geometry_service_compatibility()
self._check_detectors_compatiblity()
detectors_geo = self.getProp("DetectorGeo")["Detectors"]
detectors_sim = self.getProp("DetectorSim")["Detectors"]
detectors_moni = self.getProp("DetectorMoni")["Detectors"]
# Initial loop to mark all used detectors as active
from Gauss.Detectors.Helpers import getsubdetector
for det in detectors_geo:
det_conf = getsubdetector(det)
det_conf.Active = True
if not det_conf.isPropertySet("UseDD4hep"):
det_conf.setProp("UseDD4hep", not self.getProp("Legacy"))
for det in detectors_sim:
getsubdetector(det).Simulate = True
for det in detectors_moni:
getsubdetector(det).Monitor = True
from Configurables import UpdateManagerSvc
UpdateManagerSvc().WithoutBeginEvent = True
if self.getProp("Legacy"):
if "Version" in self.getProp("DetectorGeo"):
logging.warning(
"the Version feature for DetectorGeo only works for the DD4HEP backend. It is ignored when using DetDesc"
)
# for DetDesc, detector volumes depend on the beampipe volumes
# so we have to apply the beampipe BEFORE the other detectors
getsubdetector("BeamPipe").ApplyDetectorDetDesc(
self._base_pieces, self._det_pieces
)
for det in detectors_geo:
if self.getProp("Legacy"):
getsubdetector(det).ApplyDetectorDetDesc(
self._base_pieces, self._det_pieces
)
# Now convert all basePieces entries to fill list of geo objects
self._define_stream_items_geo()
# FIXME: Michal M.: to be removed
for det in detectors_geo:
getsubdetector(det).ApplyStream()
# No BP requested - therefore remove all elements from Geo.StreamItems
if "BeamPipeOff" == getsubdetector("BeamPipe").getProp("State"):
from Gauss.Detectors.BeamPipe import BeamPipe
BeamPipe.removeAllBeamPipeElements()
# Populate the list of geometry elements
# in the requested conversion service
self._set_geo_service()
# FIXME: This is all just temporary, especially the provided xml is only
# for Run1+2 MC
from Configurables import RegionsDefinitionSvc
svc = RegionsDefinitionSvc()
xmlfile = "$GAUSSROOT/xml/SimulationRICHAerogelOff.xml"
svc.RegionsDefinitionDbLocation = xmlfile
# Setup read-out algorithms and fields
for det in detectors_sim:
if det == "Magnet":
getsubdetector(det).ApplyMagneticField()
else:
getsubdetector(det).SetupExtraction()
for det in detectors_moni:
getsubdetector(det).SetupMonitor()
# propagate properties to other configurables
self.propagateProperties(self.GAUSSINO_GEOMETRY_OPTIONS, GaussinoGeometry())
self.propagateProperty("DetectorGeo", LHCbApp())
[docs] def _check_geometry_service_compatibility(self):
if not self.isPropertySet("Legacy"):
self.setProp("Legacy", GaussGeometry.run1or2)
if not self.getProp("Legacy") and GaussGeometry.run1or2:
msg = (
"Legacy (DetDesc) is the only supported "
"geometry service for Run 1 & 2!"
)
log.error(msg)
raise NotImplementedError(msg)
if self.getProp("Legacy") and GaussGeometry.run4or5:
msg = "DD4Hep is the only supported" "geometry service for Run 4 & 5!"
log.error(msg)
raise NotImplementedError(msg)
if not self.isPropertySet("DetectorGeo"):
log.debug(
"-> Setting default detectors for "
f"data type: {GaussGeometry.datatype}."
)
def_dets = DEFAULT_DETECTORS[GaussGeometry.datatype]
if not self.getProp("Legacy"):
if GaussGeometry.datatype == "2022":
def_dets = RUN3_DETECTORS_DETDESC_NO_UT
elif GaussGeometry.datatype in ["2023", "2024", "Run3"]:
def_dets = RUN3_DETECTORS_DETDESC
self.setProp("DetectorGeo", {"Detectors": def_dets})
if not self.isPropertySet("DetectorSim"):
sim_dets = [
det for det in def_dets if det not in NON_SIMULATABLE_DETECTORS
]
self.setProp("DetectorSim", {"Detectors": sim_dets})
if not self.isPropertySet("DetectorMoni"):
moni_dets = [
det for det in def_dets if det not in NON_MONITORABLE_DETECTORS
]
self.setProp("DetectorMoni", {"Detectors": moni_dets})
[docs] def _check_detectors_compatiblity(self):
for incompat_dets in INCOMPATIBLE_DETECTORS:
used_dets = self.getProp("DetectorGeo")["Detectors"]
conflicts = [det for det in used_dets if det in incompat_dets]
if len(conflicts) > 1:
msg = f"Incompatible detectors: {conflicts}."
log.error(msg)
raise ValueError(msg)
[docs] def _set_geo_service(self):
if self.getProp("Legacy"):
self._setup_detdesc()
else:
self._configure_dd4hep_svc()
[docs] def _setup_detdesc(self):
from Configurables import ApplicationMgr, DDDBConf
from Configurables import GaussGeo as detdesc_svc
from Configurables import LHCb__DetDesc__ReserveDetDescForEvent as detdesc_iov
from Configurables import LHCbApp
# Dispatch gometry backend
LHCbApp().GeometryBackend = "DetDesc"
DDDBConf().GeometryBackend = "DetDesc"
# GeometryVersion and ConditionsVersion properties are unique entries
def _check_propertyset(app_property, geometry_property):
if LHCbApp().isPropertySet(app_property):
log.warning(
f"LHCbApp().{app_property} property is ignored:: using GaussGeometry().{geometry_property}"
)
_check_propertyset("DDDBtag", "GeometryVersion")
_check_propertyset("CondDBtag", "ConditionsVersion")
def _raise_error(_property):
msg = f"GaussGeometry().{_property} must be set for DetDesc"
log.error(msg)
raise ValueError(msg)
if self.isPropertySet("GeometryVersion"):
# FIXME:
# check the syntax is compatible with DetDesc, i.e. it does not start with "runX/"
LHCbApp().DDDBtag = self.getProp("GeometryVersion")
else: # or could set it to "upgrade/master"
_raise_error("GeometryVersion")
if self.isPropertySet("ConditionsVersion"):
# FIXME:
LHCbApp().CondDBtag = self.getProp("ConditionsVersion")
else: # or could set it to "upgrade/master"
_raise_error("ConditionsVersion")
# Setup the Gauss geometry service for DetDesc
GaussinoGeometry().GeometryService = "GaussGeo"
detdesc_svc(
UseAlignment=True,
AlignAllDetectors=True,
)
for el in self._listOfGeoObjects_:
detdesc_svc().GeoItemsNames.append(el)
ApplicationMgr().TopAlg.append(
detdesc_iov(
"IOVProducer",
ODIN="DAQ/ODIN",
)
)
[docs] def _define_stream_items_geo(self):
for region, pieces in self._base_pieces.items():
path = f"/dd/Structure/LHCb/{region}/"
if not self._det_pieces[region]:
continue
# This should preserve order
for element in pieces + self._det_pieces[region]:
stream_item = path + element
if stream_item not in self._listOfGeoObjects_:
self._listOfGeoObjects_.append(stream_item)
[docs] def _configure_dd4hep_svc(self) -> None:
"""Sets up the properties of the main DD4hep geometry service: ``DD4hepSvc``
based on what datatype, sub-detectors, etc. were chosen in Gauss. In particular,
the following properties are modified:
- geometry properties,
- conditions properties,
- list of detectors.
Gaussino is also configured to use a dedicated geometry conversion service:
``LHCbDD4hepCnvSvc``.
DDDBConf is also configured and handles the dispatching of the Conditions
GitLab repository
.. todo ::
We still have to configure ``LHCbApp``. This can be removed
in the future.
"""
from Configurables import ApplicationMgr
from Configurables import LHCb__Det__LbDD4hep__DD4hepSvc as dd4hep_svc
from Configurables import LHCb__Det__LbDD4hep__IOVProducer as dd4hep_iov
# configure Gaussino
GaussinoGeometry().GeometryService = "LHCbDD4hepCnvSvc"
# geometry
geometry_location, geometry_version = self._configure_dd4hep_geometry()
dets_to_pass_to_lhcb = ["/world"]
for det in self.getProp("DetectorGeo")["Detectors"]:
lhcb_det = self._lhcb_dd4hep_det_map.get(det)
if lhcb_det:
dets_to_pass_to_lhcb.append(lhcb_det)
# conditions
conditions_location, conditions_version = self._configure_dd4hep_conditions()
dd4hep_svc(
GeometryLocation=geometry_location,
GeometryVersion=geometry_version,
GeometryMain="LHCb.xml",
DetectorList=dets_to_pass_to_lhcb,
ConditionsLocation=conditions_location,
ConditionsVersion=conditions_version,
)
# we have to set DD4hep backend of DDDBConf manually
# as we cannot rely on UseDD4Hep compile variable
DDDBConf(
GeometryLocation=geometry_location,
GeometryVersion=geometry_version,
GeometryMain="LHCb.xml",
DD4HepDetectorList=dets_to_pass_to_lhcb,
ConditionsLocation=conditions_location,
ConditionsVersion=conditions_version,
GeometryBackend="DD4Hep",
)
# FIXME:
# we have to set DD4hep values of LHCbApp manually
# as we cannot rely on UseDD4Hep compile variable
LHCbApp(
GeometryLocation=geometry_location,
GeometryVersion=geometry_version,
GeometryMain="LHCb.xml",
GeometryBackend="DD4Hep",
)
ApplicationMgr().TopAlg.append(
dd4hep_iov(
"IOVProducer",
ODIN="DAQ/ODIN",
)
)
[docs] def _configure_dd4hep_conditions(self) -> tuple[str, str]:
"""Sets the properties of the conditions database that will be used by Gauss.
By default:
- ``ConditionsLocation`` is set to where ``lhcb-conditions-database``
deployment area on CVMFS is,
- ``ConditionsVersion`` is set based on the provided ``DataType`` in Gauss.
Raises:
ValueError: when the default ``lhcb-conditions-database`` is not provided
by ``GIT_CONDDBS``.
NotADirectoryError: when the path to conditions database does not exist.
ValueError: when the default version for a given datatype was not found.
Returns:
tuple[str, str]: Parsed location and version of the conditions database.
"""
from DDDB.Configuration import GIT_CONDDBS
conds_loc = self.getProp("ConditionsLocation")
if not self.isPropertySet("ConditionsLocation"):
conds_loc = GIT_CONDDBS.get("lhcb-conditions-database")
if not conds_loc:
return ValueError(
"Could not find git repository: lhcb-conditions-database"
)
log.debug(f"-> Setting default DD4hep conditions location: '{conds_loc}'")
self.setProp("ConditionsLocation", conds_loc)
if not os.path.isdir(conds_loc):
raise NotADirectoryError(
"Provided directory of the conditions database"
f"'{conds_loc}' does not exist."
)
if conds_loc.endswith(".git"):
conds_loc = f"git:{conds_loc}"
else:
conds_loc = f"file:{conds_loc}/"
conds_version = self.getProp("ConditionsVersion")
if not self.isPropertySet("ConditionsVersion"):
conds_version = CONDITIONS_VERSIONS_DD4HEP.get(GaussGeometry.datatype)
if not conds_version:
raise ValueError(
"Failed to set default version of the DD4hep conditions database "
f"for datatype '{conds_version}'. Check the configuration again."
)
log.debug(
f"-> Setting default DD4hep conditions version: '{conds_version}'"
)
self.setProp("ConditionsVersion", conds_version)
return conds_loc, conds_version
[docs] def _configure_dd4hep_geometry(self) -> tuple[str, str]:
"""Sets the DD4hep geometry properties that will be used by Gauss.
In particular, Gauss has to control what is provided by the Detector project:
- XML files under ``GeometryLocation``/``GeometryVersion``,
- as well as detector element files implemented in C++.
By default:
- ``GeometryLocation`` is set to the same location as the Detector project
that Gauss was built against
- ``GeometryVersion`` is set based on the provided ``DataType`` in Gauss.
The main ``LHCb.xml`` file has to be rebuilt every single time you run Gauss
because in Gauss it should be possible to simulate full or partial LHCb geometry
and therefore we must be able to decide which sub-detectors' XML files
are included in the final `LHCb.xml` file. This new file is created at initialize
time and stored in `/tmp` directory.
Raises:
NotADirectoryError: when directory ``GeometryLocation`` does not exist
ValueError: when unable to set the default version of geometry
NotADirectoryError: when ``GeometryVersion`` is not found in ``GeometryLocation``
Returns:
tuple[str, str]: New location and version of the temporary `LHCb.xml` file.
"""
from Gauss.Detectors import xml_writer
geo_loc = self.getProp("GeometryLocation")
if not self.isPropertySet("GeometryLocation"):
geo_loc = os.path.join(os.environ["DETECTOR_PROJECT_ROOT"], "compact")
log.debug(f"-> Setting default DD4hep geometry location: '{geo_loc}'")
self.setProp("GeometryLocation", geo_loc)
if not os.path.isdir(geo_loc):
raise NotADirectoryError(
f"Invalid directory with compact XML files: '{geo_loc}'."
)
if self.isPropertySet("GeometryVersion"):
geo_version = self.getProp("GeometryVersion")
else:
geo_version = GEOMETRY_VERSIONS_DD4HEP.get(GaussGeometry.datatype)
if not geo_version:
raise ValueError(
"Failed to set default version of the DD4hep geometry version "
f"for datatype '{geo_version}'. Check the configuration again."
)
log.debug(f"-> Setting default DD4hep geometry version: '{geo_version}'")
self.setProp("GeometryVersion", geo_version)
geo_dir = os.path.join(geo_loc, geo_version)
if not os.path.isdir(geo_dir):
raise NotADirectoryError(
f"Invalid geometry version of compact XML files: '{geo_version}'."
)
def get_version_from_detector(geo_dir):
import xml.etree.ElementTree as ET
tree = ET.parse(os.path.join(geo_dir, "path.xml"))
root = tree.getroot()
constants = root.find("define").findall("constant")
detector_versions = {}
for constant in constants:
path_components = constant.get("value").split("/")
detector_name = constant.get("name").split(":")[1]
detector_versions[detector_name] = path_components[-1]
return detector_versions
# due to the different names of detectors in DD4HEP, a temporary hack is being used to call the Python files under Detectors. Hopefully, it can be delegated a.s.a.p
def update_detector_names(detector_versions):
detector_versions.update(
{
"Rich1Pmt": detector_versions.get("Rich", ""),
"Rich2Pmt": detector_versions.get("Rich", ""),
"Shield": detector_versions.get("NeutronShielding", ""),
"Ecal": detector_versions.get("ECAL", ""),
"Hcal": detector_versions.get("HCAL", ""),
"BeamPipe": detector_versions.get("Pipe", ""),
}
)
return detector_versions
def get_version(det):
if "Version" in self.getProp("DetectorGeo"):
comp_version = self.getProp("DetectorGeo")["Version"][det]
else:
comp_version = detector_versions[det]
if comp_version == None:
msg = f"Incompatible detector {det} for the the datatype {GaussGeometry.datatype}"
raise ValueError(msg)
return comp_version
detector_versions = update_detector_names(get_version_from_detector(geo_dir))
from Gauss.Detectors.Helpers import getsubdetector
detectors_geo = self.getProp("DetectorGeo")["Detectors"]
beampipe_version = DEFAULT_DD4HEP_BEAMPIPE_VERSION
# deal with the fact that SMOG2 is not in geometry version run3/2024.Q1.2-v00.00
if geo_version == "run3/2024.Q1.2-v00.00":
try:
detectors_geo.remove("SMOG2")
except:
pass
for det in detectors_geo:
getsubdetector(det).ApplyDetectorDD4hep(
get_version(det), self._base_pieces, self._det_pieces
)
beampipeoff = "BeamPipeOff" == getsubdetector("BeamPipe").getProp("State")
if not beampipeoff:
getsubdetector("BeamPipe").ApplyDetectorDD4hep(
get_version("BeamPipe"), self._base_pieces, self._det_pieces
)
return xml_writer.create_xml(
self._listOfXMLIncludes_,
directory=geo_loc,
beampipeversion=get_version("BeamPipe"),
materialsversion=get_version("Materials"),
regionsversion=get_version("Regions"),
beampipeoff=beampipeoff,
)