Inheritance diagram for python.db.DefectsDB:

Collaboration diagram for python.db.DefectsDB:

Public Member Functions
None	__init__ (self, str connection_string=DEFAULT_CONNECTION_STRING, bool read_only=True, bool create=False, Union[str, Tuple] tag="HEAD")

None	__del__ (self)

None	create_defect (self, str name, str description)

IOVSet	retrieve (self, Optional[Union[int, Tuple[int, int], RunLumi]] since=None, Optional[Union[int, Tuple[int, int], RunLumi]] until=None, Optional[Iterable[Union[str, int]]] channels=None, bool nonpresent=False, bool primary_only=False, Optional[Collection[str]] ignore=None, bool with_primary_dependencies=False, bool intersect=False, bool with_time=False, bool evaluate_full=True)

def	storage_buffer (self)

None	insert (self, Union[str, int] defect_id, int since, int until, str comment, str added_by, bool present=True, bool recoverable=False)

def	insert_multiple (self, Iterable[IOVType] defect_list, str tag='HEAD', bool use_flask=False, str flask_cool_target='oracle://ATONR_COOLOFL_GPN/ATLAS_COOLOFL_GLOBAL', Mapping[str, str] flask_auth={}, str flask_db='CONDBR2', str flask_uri='https://cool-proxy-app.cern.ch/cool/multi_iovs')

None	validate_clauses (self, str clauses)

None	update_virtual_defect (self, str defect_name, str clauses, Optional[str] comment=None)

None	rename_defect (self, str defect_name, str new_defect_name)

int	new_virtual_defect (self, str defect_name, Optional[str] comment, str clauses)

Set[str]	resolve_primary_defects (self, Iterable[DefectLogic] defect_logics)

Mapping[str, DefectLogic]	virtual_defect_logics (self)

def	get_intolerable_defects (self, bool old_primary_only=True, Sequence[str] exclude=['TIGHT', 'IDTIGHT', 'PHYS_.*'])

str	logics_tag (self)

List[str]	logics_tags (self)

List[str]	defects_tags (self)

str	next_logics_tag (self)

str	new_hierarchical_tag (self, str defects_tag, str logics_tag)

bool	defects_tag_valid (self)

bool	logics_tag_valid (self)

None	check_tag_validity (self)

str	defects_tag (self)

tagtype	tag (self)

List[str]	tags (self)

str	new_logics_tag (self, str description="")

str	new_defects_tag (self, str name, str description, Optional[Iterable] iovranges=None)

cool.IFolderSet	parent_folderset (self)

cool.IFolder	defects_folder (self)

cool.IFolder	defect_logic_folder (self)

Set[int]	defect_ids (self)

Set[str]	defect_names (self)

MutableMapping[Union[str, int], Union[str, int]]	defect_id_map (self)

Set[int]	virtual_defect_ids (self)

Set[str]	virtual_defect_names (self)

MutableMapping[Union[str, int], Union[str, int]]	virtual_defect_id_map (self)

int	defect_chan_as_id (self, Union[str, int] channel, bool primary_only=False)

List[int]	defect_names_as_ids (self, Iterable[Union[str, int]] channels)

Tuple[Set[int], Set[str], Mapping[Union[str, int], Union[str, int]]]	get_channels (self)

Tuple[Set[int], Set[str], Mapping[Union[str, int], Union[str, int]]]	get_virtual_channels (self)

MutableMapping[Union[str, int], str]	get_channel_descriptions (self, Iterable[Union[str, int]] channels)

MutableMapping[Union[str, int], str]	get_virtual_channel_descriptions (self, Iterable[Union[str, int]] channels)

Mapping[Union[str, int], str]	all_defect_descriptions (self)

None	set_channel_description (self, Union[str, int] channel, str description)

bool	defect_is_virtual (self, Union[str, int] defect_id)

Union[str, List[str]]	normalize_defect_names (self, Union[str, Iterable[str]] defect_id)

Public Attributes
	connection_string

	logics_tag

	logics_tag

Private Member Functions
None	_create_defect_with_id (self, int did, str name, str description)

None	_insert_iov (self, IOVType iov, str tag)

None	_insert (self, Union[str, int] defect_id, Union[int, Tuple[int, int], RunLumi] since, Union[int, Tuple[int, int], RunLumi] until, str comment, str added_by, bool present=True, bool recoverable=False, str tag='HEAD')

def	_insert_multiple_flask (self, Iterable[IOVType] defect_list, str tag, str flask_cool_target, Mapping[str, str] flask_auth, str flask_db, str flask_uri)

None	_update_virtual_defect (self, str defect_name, str clauses, Optional[str] comment=None, Optional[str] tag=None)

List[DefectLogic]	_resolve_evaluation_order (self, Optional[Iterable[str]] defects=None)

Mapping[str, DefectLogic]	_get_virtual_defect_logics (self, str tag)

bool	_virtual_defect_consistency_check (self)

None	_tag_head_and_lock (self, cool.IFolder folder, str name, str description)

None	_clear_connections (self)

Tuple[cool.IFolder, cool.Record]	_load_folder (self, folder, create_function)

None	_load_defects_folder (self)

cool.IFolder	_create_defects_folder (self, cool.IDatabase db)

None	_load_defect_logic_folder (self)

cool.IFolder	_create_defect_logic_folder (self, cool.IDatabase db)

None	_populate_defect_ids (self)

None	_populate_virtual_defect_ids (self)

None	_new_defect (self, int did, str dname)

None	_new_virtual_defect (self, int did, str dname)

Detailed Description

Implementation is split into mixins.

    DefectsDBIDsNamesMixin:
        Defines functions for dealing with defect IDs
    
    DefectsDBFolderMixin:
        Logic for loading or creating the COOL folder/database
    
    DefectsDBFoldersMixin:
        Functions for managing tags
    
    DefectsDBVirtualDefectsMixin:
        Internal code for managing and computing virutal defects
        
Public interface is nominally defined in this class (DefectsDB).

Definition at line 38 of file DQDefects/python/db.py.

Constructor & Destructor Documentation

◆ init()

None python.db.DefectsDB.__init__	(		self,
		str	connection_string = `DEFAULT_CONNECTION_STRING`,
		bool	read_only = `True`,
		bool	create = `False`,
		Union[str, Tuple]	tag = `"HEAD"`
	)

Create a new DefectsDB instance.

The COOL folder resides at `DEFECTS_FOLDER`, which is a module-level
varaiable in this module.

Parameters:
    `connection_string` : A COOL connection string, or a string ending 
                  in .db. In the latter case, the database is
                  assumed to be a sqlite database at that path.
    `read_only` : Indicates whether insertions will be disallowed on
          this connection. (Default: True)
    `create`    : If True, attempt to create the database if it doesn't
          exist. (Default: False)
    `tag`       : The COOL tag(s) to use.  Either a single string used
          for both defects and virtual defect logic, or a
          2-tuple (defecttag, logictag).  The default is to
          use HEAD for both defects and logic.  Either
          folder-level or hierarchical tags can be given.

Definition at line 61 of file DQDefects/python/db.py.

     def __init__(self, connection_string: str = DEFAULT_CONNECTION_STRING,
                  read_only: bool = True, create: bool = False, tag: Union[str, Tuple] = "HEAD") -> None:
         """
         Create a new DefectsDB instance.
         
         The COOL folder resides at `DEFECTS_FOLDER`, which is a module-level
         varaiable in this module.
         
         Parameters:
             `connection_string` : A COOL connection string, or a string ending 
                                   in .db. In the latter case, the database is
                                   assumed to be a sqlite database at that path.
             `read_only` : Indicates whether insertions will be disallowed on
                           this connection. (Default: True)
             `create`    : If True, attempt to create the database if it doesn't
                           exist. (Default: False)
             `tag`       : The COOL tag(s) to use.  Either a single string used
                           for both defects and virtual defect logic, or a
                           2-tuple (defecttag, logictag).  The default is to
                           use HEAD for both defects and logic.  Either
                           folder-level or hierarchical tags can be given.
         """
         self.connection_string = connection_string
         self._read_only = read_only
         self._create = create
         if isinstance(tag, six.string_types):
             self._tag = tagtype(tag, tag) if tag else tagtype("HEAD", "HEAD")
         else:
             try:
                 tag = tagtype._make(tag)
             except Exception:
                 raise TypeError('tag argument must be a 2-element sequence')
             if len(tag) != 2:
                 raise TypeError('tag argument must be a 2-element sequence')
             self._tag = tag
         self._tag = tagtype(self._tag[0],
                             self._tag[1])
  
         # COOL has no way of emptying a storage buffer. Creating a new storage
         # buffer flushes the old one. Therefore, if an exception happens 
         # halfway through setting up a storage buffer, subsequent operations
         # risk flushing a half-completed operation. In that case, we refuse to 
         # do any more write operations with that database connection.
         self._bad_state = False
         
         super(DefectsDB, self).__init__()
         
         if create:
             # Trigger creation of folders through properties
             self.defects_folder
             self.defect_logic_folder
     

◆ del()

None python.db.DefectsDB.__del__ ( self )

Ideally we would use inheritance to call destructors, but this isn't 
possible in the general case with the way we (ab)use mixins, so we just 
do it here.

Definition at line 113 of file DQDefects/python/db.py.

     def __del__(self) -> None:
         """
         Ideally we would use inheritance to call destructors, but this isn't 
         possible in the general case with the way we (ab)use mixins, so we just 
         do it here.
         """
         self._clear_connections()
     

Member Function Documentation

◆ _clear_connections()

None python.folders.DefectsDBFoldersMixin._clear_connections ( self )

privateinherited

Definition at line 29 of file folders.py.

     def _clear_connections(self) -> None:
         log.debug("DefectsDB connections cleared")
         for connection in self._connections:
             connection.closeDatabase()
         self._connections = []
     

◆ _create_defect_logic_folder()

cool.IFolder python.folders.DefectsDBFoldersMixin._create_defect_logic_folder	(		self,
		cool.IDatabase	db
	)

privateinherited

Creates the COOL database/folder for virtual defect logic, if they don't exist.
Internal - use create=True in the constructor to create the COOL folder.

Definition at line 115 of file folders.py.

     def _create_defect_logic_folder(self, db: cool.IDatabase) -> cool.IFolder:
         """
         Creates the COOL database/folder for virtual defect logic, if they don't exist.
         Internal - use create=True in the constructor to create the COOL folder.
         """
         log.info("Creating defect logic folder")
         
         record_spec = cool.RecordSpecification()
         record_spec.extend("clauses", ST.String4k)
         
         folder_spec = cool.FolderSpecification(FV.MULTI_VERSION, record_spec)
         return db.createFolder(DEFECT_LOGIC_FOLDER, folder_spec, 
                                DEFECT_LOGIC_FOLDER_DESC, True)

◆ _create_defect_with_id()

None python.db.DefectsDB._create_defect_with_id	(		self,
		int	did,
		str	name,
		str	description
	)

private

Create a new type of defect, specifying the defect ID.
See also: `create_defect`, `new_system_defect`.

Parameters:
    `did` : defect_id
    `name` : name of the defect
    `description` : a human-readable description of the defect

Definition at line 136 of file DQDefects/python/db.py.

     def _create_defect_with_id(self, did: int, name: str, description: str) -> None:
         """
         Create a new type of defect, specifying the defect ID.
         See also: `create_defect`, `new_system_defect`.
         
         Parameters:
             `did` : defect_id
             `name` : name of the defect
             `description` : a human-readable description of the defect
         """
         log.info("Creating new defect %s (0x%08x)", name, did)
         if did in self.defect_ids: raise DefectExistsError(did)
         try:
             oldname = self.normalize_defect_names(name)
             raise DefectExistsError(f'Defect {oldname} already exists')
         except DefectUnknownError:
             pass
         
         self.defects_folder.createChannel(did,
                                           name.encode('ascii'),
                                           description.encode('utf-8'))
         self._new_defect(did, name)
     

◆ _create_defects_folder()

cool.IFolder python.folders.DefectsDBFoldersMixin._create_defects_folder	(		self,
		cool.IDatabase	db
	)

privateinherited

Creates the COOL database/folder for defects, if they don't exist.
Internal - use create=True in the constructor to create the COOL folder.

Definition at line 82 of file folders.py.

     def _create_defects_folder(self, db: cool.IDatabase) -> cool.IFolder:
         """
         Creates the COOL database/folder for defects, if they don't exist.
         Internal - use create=True in the constructor to create the COOL folder.
         """
         log.info("Creating defects folder")
         
         record_spec = cool.RecordSpecification()
         record_spec.extend("present",     ST.Bool)
         record_spec.extend("recoverable", ST.Bool)
         record_spec.extend("user",        ST.String255)
         record_spec.extend("comment",     ST.String255)
                 
         folder_spec = cool.FolderSpecification(FV.MULTI_VERSION, record_spec)
         return db.createFolder(DEFECTS_FOLDER, folder_spec, 
                                 DEFECT_FOLDER_DESC, True)
                                 

◆ _get_virtual_defect_logics()

Mapping[str, DefectLogic] python.virtual_mixin.DefectsDBVirtualDefectsMixin._get_virtual_defect_logics	(		self,
		str	tag
	)

privateinherited

Definition at line 233 of file virtual_mixin.py.

     def _get_virtual_defect_logics(self, tag: str) -> Mapping[str, DefectLogic]:
         if tag != "HEAD" and not self.defect_logic_folder.existsUserTag(tag.encode('ascii')):
             # The tag doesn't exist, so there is nothing to fetch.
             return {}
         
         logics = fetch_iovs(self.defect_logic_folder, tag=tag.encode('ascii'), 
                             named_channels=True)
  
         logics = {l.channel: DefectLogic(l) for l in logics}
         
         for _, defect_logic in six.iteritems (logics):
             defect_logic._populate(logics)
             
         return logics
  

◆ _insert()

None python.db.DefectsDB._insert	(		self,
		Union[str, int]	defect_id,
		Union[int, Tuple[int, int], RunLumi]	since,
		Union[int, Tuple[int, int], RunLumi]	until,
		str	comment,
		str	added_by,
		bool	present = `True`,
		bool	recoverable = `False`,
		str	tag = `'HEAD'`
	)

private

Implementation of insert, allows tag specification for internal
functions

Definition at line 328 of file DQDefects/python/db.py.

     def _insert(self, defect_id: Union[str, int], since: Union[int, Tuple[int, int], RunLumi], 
                 until: Union[int, Tuple[int, int], RunLumi], comment: str, added_by: str,
                 present: bool = True, recoverable: bool = False, tag: str = 'HEAD') -> None:
         """
         Implementation of insert, allows tag specification for internal
         functions
         """
         assert not self._read_only, "Insertion on read-only database"
         assert not self._bad_state, "Please see comment in DefectsDB constructor"
         
         # Force load of defects_folder to populate _defect_payload
         store = self.defects_folder.storeObject
         p = self._defect_payload
         
         p["present"] = present
         p["recoverable"] = recoverable
         p["user"] = added_by.encode('utf-8')
         p["comment"] = comment.encode('utf-8')
  
         defect_id = self.defect_chan_as_id(defect_id, True)
         
         store(since, until, p, defect_id, tag.encode('ascii'),
               (True if tag != 'HEAD' else False))
  

◆ _insert_iov()

None python.db.DefectsDB._insert_iov	(		self,
		IOVType	iov,
		str	tag
	)

private

Helper function for inserting IOV objects, since record order doesn't
match function argument order

Definition at line 320 of file DQDefects/python/db.py.

     def _insert_iov(self, iov: IOVType, tag: str) -> None:
         """
         Helper function for inserting IOV objects, since record order doesn't
         match function argument order
         """
         return self._insert(iov.channel, iov.since, iov.until, iov.comment,
                             iov.user, iov.present, iov.recoverable, tag)
  

◆ _insert_multiple_flask()

def python.db.DefectsDB._insert_multiple_flask	(		self,
		Iterable[IOVType]	defect_list,
		str	tag,
		str	flask_cool_target,
		Mapping[str, str]	flask_auth,
		str	flask_db,
		str	flask_uri
	)

private

Definition at line 371 of file DQDefects/python/db.py.

     def _insert_multiple_flask(self, defect_list: Iterable[IOVType],
                                tag: str, 
                                flask_cool_target: str,
                                flask_auth: Mapping[str, str],
                                flask_db: str,
                                flask_uri: str):
         import requests
         import json
         import urllib.parse
         import os
         from DQUtils.oracle import get_authentication
         data = {'grant_type':'client_credentials',
                 'audience': 'cool-flask-server'}
         data.update(flask_auth)
         auth = requests.post('https://auth.cern.ch/auth/realms/cern/api-access/token',
                              data=data
                             )
         if not auth:
             raise RuntimeError('Cannot authenticate to Flask server')
         else:
             token = auth.json()['access_token']
             log.debug(f'auth succeeded {token}')
             username, password = get_authentication(flask_cool_target)
             p = urllib.parse.urlparse(flask_cool_target)
             if not p.hostname:
                 raise ValueError(f'Cannot interpret {flask_cool_target} as a path')
             server, schema = p.hostname.upper(), p.path[1:]
  
             submit_map = {'cmd': 'addIov',
                             'COOLDEST': flask_cool_target,
                             'ORA_SCHEMA': schema,
                             'ORA_INST': flask_db,
                             'COOLOPTS': 'None',
                             'COOLSINST': flask_db,
                             'ORA_SRV': server,
                             'COOLXP': password,
                             'ORA_USER': username,
                             'cool_data': {
                                 'folder': DEFECTS_FOLDER,
                                 'description': '',
                                 'tag': tag,
                                 'record': {'present': 'Bool', 'recoverable': 'Bool',
                                         'user': 'String255', 'comment': 'String255'},
                                 'size': 0,
                                 'iovs': []
                             }}
  
             for defect in defect_list:
                 submit_map['cool_data']['iovs'].append(
                     {'channel': self.defect_chan_as_id(defect.channel, True),
                      'since': defect.since,
                      'until': defect.until,
                      'payload': {'present': defect.present,
                                  'recoverable': defect.recoverable,
                                  'user': defect.user,
                                  'comment': defect.comment
                                  }
                     }
                 )
                 submit_map['cool_data']['size'] += 1
             r = requests.post(flask_uri,
                             headers={'Authorization': f'Bearer {token}'},
                             files={'file': ('iov.json', 
                                             json.dumps({'cool_multi_iov_request': submit_map}))},
                                             verify=('DQM_COOL_FLASK_NOVERIFY' not in os.environ)
                             )
             log.debug(r.content)
             if not r or r.json()['code'] != 0:
                 raise RuntimeError(f'Unable to upload defects. Flask server returned error:\n{r.json()["message"]}')

◆ _load_defect_logic_folder()

None python.folders.DefectsDBFoldersMixin._load_defect_logic_folder ( self )

privateinherited

Internal function for populating the self._defect_logic_folder variable

Definition at line 107 of file folders.py.

     def _load_defect_logic_folder(self) -> None:
         """
         Internal function for populating the self._defect_logic_folder variable
         """        
         res = self._load_folder(DEFECT_LOGIC_FOLDER, 
                                 self._create_defect_logic_folder)
         self._defect_logic_folder, self._defect_logic_payload = res
         

◆ _load_defects_folder()

None python.folders.DefectsDBFoldersMixin._load_defects_folder ( self )

privateinherited

Internal function for populating the self._defects_folder variable

Definition at line 74 of file folders.py.

     def _load_defects_folder(self) -> None:
         """
         Internal function for populating the self._defects_folder variable
         """
         log.debug("Loading defects folder")
         res = self._load_folder(DEFECTS_FOLDER, self._create_defects_folder)
         self._defects_folder, self._defect_payload = res
     

◆ _load_folder()

Tuple[cool.IFolder, cool.Record] python.folders.DefectsDBFoldersMixin._load_folder	(	self,
		folder,
		create_function
	)

privateinherited

Internal function used to load a COOL folder

Definition at line 46 of file folders.py.

     def _load_folder(self, folder, create_function) -> Tuple[cool.IFolder, cool.Record]:
         """
         Internal function used to load a COOL folder
         """
         db_folder_string = "%s::%s" % (self.connection_string, folder)
     
         if not self._create:
             create_function = None
     
         db, folder = Databases.get_folder(db_folder_string, 
                                           read_only=self._read_only,
                                           create_function=create_function,
                                           also_db=True)
         
         self._connections.append(db)
         
         payload = cool.Record(folder.payloadSpecification())
         
         return folder, payload
         

◆ _new_defect()

None python.ids.DefectsDBIDsNamesMixin._new_defect	(		self,
		int	did,
		str	dname
	)

privateinherited

Internal function used to keep defect IDs/names uptodate.

Definition at line 121 of file ids.py.

     def _new_defect(self, did: int, dname: str) -> None:
         """
         Internal function used to keep defect IDs/names uptodate.
         """
         self.defect_ids.add(did)
         self.defect_names.add(dname)
         self.defect_id_map[did] = dname
         self.defect_id_map[dname] = did
         

◆ _new_virtual_defect()

None python.ids.DefectsDBIDsNamesMixin._new_virtual_defect	(		self,
		int	did,
		str	dname
	)

privateinherited

Internal function used to keep defect IDs/names uptodate.

Definition at line 130 of file ids.py.

     def _new_virtual_defect(self, did: int, dname: str) -> None:
         """
         Internal function used to keep defect IDs/names uptodate.
         """
         self.virtual_defect_ids.add(did)
         self.virtual_defect_names.add(dname)
         self.virtual_defect_id_map[did] = dname
         self.virtual_defect_id_map[dname] = did
         # Reset this so we will reload logic later
         self._virtual_defect_logics = None
     

◆ _populate_defect_ids()

None python.ids.DefectsDBIDsNamesMixin._populate_defect_ids ( self )

privateinherited

Called the first time any of defect_{ids,names,id_map,etc} is called,
and populates internal variables to store the channel ids/names for the
life of the DefectsDB instance.

Definition at line 90 of file ids.py.

     def _populate_defect_ids(self) -> None:
         """
         Called the first time any of defect_{ids,names,id_map,etc} is called,
         and populates internal variables to store the channel ids/names for the
         life of the DefectsDB instance.
         """
         ids, names, mapping = get_channel_ids_names(self.defects_folder)
         self._defect_ids = set(ids)
         self._defect_names = set(names)
         self._defect_id_map = mapping
         

◆ _populate_virtual_defect_ids()

None python.ids.DefectsDBIDsNamesMixin._populate_virtual_defect_ids ( self )

privateinherited

Called the first time any of virtual_defect_{ids,names,id_map,etc} is called,
and populates internal variables to store the channel ids/names for the
life of the DefectsDB instance.

Definition at line 101 of file ids.py.

     def _populate_virtual_defect_ids(self) -> None:
         """
         Called the first time any of virtual_defect_{ids,names,id_map,etc} is called,
         and populates internal variables to store the channel ids/names for the
         life of the DefectsDB instance.
         """
         _, _, mapping = get_channel_ids_names(self.defect_logic_folder)
         try:
             self._virtual_defect_names = set(self.virtual_defect_logics.keys())
             self._virtual_defect_ids = set(mapping[name] for name in self._virtual_defect_names)
             all_defects = self._virtual_defect_names | self._virtual_defect_ids
             self._virtual_defect_id_map = dict(item for item in mapping.items() 
                                                if item[0] in all_defects)
         except InvalidLogicTagError:
             # Presumably this tag doesn't exist yet, which is ok if we're about
             # to create some new virtual defects. ?
             self._virtual_defect_names = set()
             self._virtual_defect_ids = set()
             self._virtual_defect_map = {}
     

◆ _resolve_evaluation_order()

List[DefectLogic] python.virtual_mixin.DefectsDBVirtualDefectsMixin._resolve_evaluation_order	(		self,
		Optional[Iterable[str]]	defects = `None`
	)

privateinherited

Returns a list of DefectLogic objects that need to be evaluated, 
in the correct order for them to be consistent.

`defects` should be a list of names

Definition at line 184 of file virtual_mixin.py.

     def _resolve_evaluation_order(self, defects: Optional[Iterable[str]] = None) -> List[DefectLogic]:
         """
         Returns a list of DefectLogic objects that need to be evaluated, 
         in the correct order for them to be consistent.
         
         `defects` should be a list of names
         """
         if defects is None:
             logics = self.virtual_defect_logics.items()
         else:
             # flags begins as a list of names.
             # Make it a list of logic objects instead
             # If it is not in the logics map, the logic is not defined for the
             # query interval and we shouldn't try to resolve it.
             
             
             logics = [self.virtual_defect_logics[d] for d in defects]
         
         class MasterNode(object):
             dependencies = logics
             
         resolved = _resolve_dependencies(MasterNode)
         resolved.remove(MasterNode)
         return resolved
     

◆ _tag_head_and_lock()

None python.tags.DefectsDBTagsMixin._tag_head_and_lock	(		self,
		cool.IFolder	folder,
		str	name,
		str	description
	)

privateinherited

Give the current HEAD of `folder` a new tag and lock it.

Definition at line 52 of file tags.py.

     def _tag_head_and_lock(self, folder: cool.IFolder, name: str, description: str) -> None:
         """
         Give the current HEAD of `folder` a new tag and lock it.
         """
         LOCKED = cool.HvsTagLock.LOCKED
         aname = name.encode('ascii')
         folder.cloneTagAsUserTag('HEAD', aname, description.encode('utf-8'))
         folder.setTagLockStatus(aname, LOCKED)
     

◆ _update_virtual_defect()

None python.virtual_mixin.DefectsDBVirtualDefectsMixin._update_virtual_defect	(		self,
		str	defect_name,
		str	clauses,
		Optional[str]	comment = `None`,
		Optional[str]	tag = `None`
	)

privateinherited

Definition at line 63 of file virtual_mixin.py.

     def _update_virtual_defect(self, defect_name: str, clauses: str, 
                                comment: Optional[str] = None, tag: Optional[str] = None) -> None:
         defect_id = self.defect_chan_as_id(defect_name)
         
         assert self.defect_is_virtual(defect_id), ("Tried to update nonvirtual"
             " defect with update_virtual_defect()")
  
         self.validate_clauses(clauses)
  
         tag = tag if tag is not None else 'HEAD'
         ucomment = comment.encode('utf-8') if comment is not None else None
  
         self._defect_logic_payload["clauses"] = clauses.encode('ascii')
         
         store = self.defect_logic_folder.storeObject
         store(0, 2**63-1, self._defect_logic_payload, defect_id, tag.encode('ascii'),
               (True if tag != 'HEAD' else False))
         
         if comment is not None:
             self.defect_logic_folder.setChannelDescription(defect_id, ucomment)
  

◆ _virtual_defect_consistency_check()

bool python.virtual_mixin.DefectsDBVirtualDefectsMixin._virtual_defect_consistency_check ( self )

privateinherited

When called, uses an assertion to check that there are no missing 
defects. This is a database consistency check which should never be
violated (but was when DQDefects was new).

Definition at line 266 of file virtual_mixin.py.

     def _virtual_defect_consistency_check(self) -> bool:
         """
         When called, uses an assertion to check that there are no missing 
         defects. This is a database consistency check which should never be
         violated (but was when DQDefects was new).
         """
         all_defects = self.defect_names | self.virtual_defect_names
         used_defects = set()
         
         used_by = {}
         for defect, logic in six.iteritems (self.virtual_defect_logics):
             used_defects.update(logic.realclauses)
             for clause in logic.realclauses:
                 used_by.setdefault(clause, []).append(defect)
                 
         missing_defects = used_defects - all_defects
         
         if missing_defects:
             log.error("-- The missing defects are used by the following virtual defects")
             log.error("<missing> : <used by>")
             for missing in missing_defects:
                 log.error(missing + ":" + ", ".join(used_by[missing]))
             
         assert not missing_defects, ("The following defects are used but not "
                                      "defined anywhere: %r" % missing_defects)
         return True

◆ all_defect_descriptions()

Mapping[Union[str, int], str] python.ids.DefectsDBIDsNamesMixin.all_defect_descriptions ( self )

inherited

A dictionary of all (virtual and primary) defect descriptions

Definition at line 265 of file ids.py.

     def all_defect_descriptions(self) -> Mapping[Union[str, int], str]:
         """
         A dictionary of all (virtual and primary) defect descriptions
         """
         result = self.get_channel_descriptions(self.defect_names)
         gvcd = self.get_virtual_channel_descriptions
         result.update(gvcd(self.virtual_defect_names))
         return result
     

◆ check_tag_validity()

None python.tags.DefectsDBTagsMixin.check_tag_validity ( self )

inherited

Ensure that the tags that this DefectsDB instance was constructed with
are functional

Definition at line 110 of file tags.py.

     def check_tag_validity(self) -> None:
         """
         Ensure that the tags that this DefectsDB instance was constructed with
         are functional
         """
         if not (self.defects_tag_valid or self.logics_tag_valid):
             raise InvalidTagError("Tag doesn't resolve: {0}".format(self._tag))
     

◆ create_defect()

None python.db.DefectsDB.create_defect	(		self,
		str	name,
		str	description
	)

Create a new type of defect; tries to figure out system ID from the
defect name. See also: `create_defect_with_id`, `new_system_defect`.

Parameters:
    `name` : name of the defect
    `description` : a human-readable description of the defect
Returns:
    the ID number of the new defect

Definition at line 121 of file DQDefects/python/db.py.

     def create_defect(self, name: str, description: str) -> None:
         """
         Create a new type of defect; tries to figure out system ID from the
         defect name. See also: `create_defect_with_id`, `new_system_defect`.
         
         Parameters:
             `name` : name of the defect
             `description` : a human-readable description of the defect
         Returns:
             the ID number of the new defect
         """        
         sysid = choose_new_defect_id(self.defect_id_map, name)
         log.info("Creating new defect %s: system ID %08x", name, sysid)
         self._create_defect_with_id(sysid, name, description)
  

◆ defect_chan_as_id()

int python.ids.DefectsDBIDsNamesMixin.defect_chan_as_id	(		self,
		Union[str, int]	channel,
		bool	primary_only = `False`
	)

inherited

Returns the defect ID for a virtual defect.
Accepts a `channel` as an integer/string and returns it as an integer.

Will raise DefectUnknownError if `channel` is an unknown ID or string.

This function first checks against non-virtual defects, then virutal 
defects. Thus virtual-defects are lazily loaded.

Definition at line 203 of file ids.py.

     def defect_chan_as_id(self, channel: Union[str, int], primary_only: bool = False) -> int:
         """
         Returns the defect ID for a virtual defect.
         Accepts a `channel` as an integer/string and returns it as an integer.
         
         Will raise DefectUnknownError if `channel` is an unknown ID or string.
         
         This function first checks against non-virtual defects, then virutal 
         defects. Thus virtual-defects are lazily loaded.        
         """
         from builtins import int
         if isinstance(channel, int):
             if (channel not in self.defect_ids and 
                 (not primary_only and channel not in self.virtual_defect_ids)):
                 raise DefectUnknownError(channel)
             return channel
         elif isinstance(channel, six.string_types):
             if channel in self.defect_names:
                 return self.defect_id_map[channel]
             if not primary_only and channel in self.virtual_defect_names:
                 return self.virtual_defect_id_map[channel]
             raise DefectUnknownError(channel)
         raise RuntimeError("Invalid `channel` type, got %s, expected integer"
                               " or string" % type(channel))
     

◆ defect_id_map()

MutableMapping[Union[str,int], Union[str,int]] python.ids.DefectsDBIDsNamesMixin.defect_id_map ( self )

inherited

Gives the dictionary relating COOL channel ids to defect names and vice
versa, retrieving them from the database if necessary.

Definition at line 163 of file ids.py.

     def defect_id_map(self) -> MutableMapping[Union[str,int], Union[str,int]]:
         """
         Gives the dictionary relating COOL channel ids to defect names and vice
         versa, retrieving them from the database if necessary.
         """
         if not self._initialized: 
             self._populate_defect_ids()
             self._initialized = True
         return self._defect_id_map
         

◆ defect_ids()

Set[int] python.ids.DefectsDBIDsNamesMixin.defect_ids ( self )

inherited

Gives the set of defect IDs that exist in COOL

Definition at line 142 of file ids.py.

     def defect_ids(self) -> Set[int]:
         """
         Gives the set of defect IDs that exist in COOL
         """
         if not self._initialized: 
             self._populate_defect_ids()
             self._initialized = True
         return self._defect_ids
     

◆ defect_is_virtual()

bool python.ids.DefectsDBIDsNamesMixin.defect_is_virtual	(		self,
		Union[str, int]	defect_id
	)

inherited

Returns True if the `defect_id` represents a virtual defect, False if it 
is not and raises if it doesn't exist

Parameters:
    `defect_id` : defect channel id or name

Definition at line 285 of file ids.py.

     def defect_is_virtual(self, defect_id: Union[str, int]) -> bool:
         """
         Returns True if the `defect_id` represents a virtual defect, False if it 
         is not and raises if it doesn't exist
         
         Parameters:
             `defect_id` : defect channel id or name
         """
         from builtins import int
         if isinstance(defect_id, int):
             return DefectID(defect_id).is_virtual
         
         if not isinstance(defect_id, six.string_types):
             raise RuntimeError("Invalid defect_id, expected int or string")
         
         if defect_id in self.defect_names:
             return False
             
         if defect_id in self.virtual_defect_names:
             return True
         
         raise DefectUnknownError(defect_id)
  

◆ defect_logic_folder()

cool.IFolder python.folders.DefectsDBFoldersMixin.defect_logic_folder ( self )

inherited

Returns the folder containing the virtual defect logic, loading it if necessary

Definition at line 100 of file folders.py.

     def defect_logic_folder(self) -> cool.IFolder:
         """
         Returns the folder containing the virtual defect logic, loading it if necessary
         """
         if self._defect_logic_folder is None: self._load_defect_logic_folder()
         return self._defect_logic_folder
     

◆ defect_names()

Set[str] python.ids.DefectsDBIDsNamesMixin.defect_names ( self )

inherited

Gives the set of defect names that exist in COOL

Definition at line 152 of file ids.py.

     def defect_names(self) -> Set[str]:
         """
         Gives the set of defect names that exist in COOL
         """
         if not self._initialized: 
             self._populate_defect_ids()
             self._initialized = True
         assert self._defect_names is not None, self._initialized
         return self._defect_names
     

◆ defect_names_as_ids()

List[int] python.ids.DefectsDBIDsNamesMixin.defect_names_as_ids	(		self,
		Iterable[Union[str, int]]	channels
	)

inherited

Returns a list of channels as IDs

Definition at line 228 of file ids.py.

     def defect_names_as_ids(self, channels: Iterable[Union[str, int]]) -> List[int]:
         """
         Returns a list of channels as IDs
         """
         return [self.defect_chan_as_id(chan) for chan in channels]
     

◆ defects_folder()

cool.IFolder python.folders.DefectsDBFoldersMixin.defects_folder ( self )

inherited

Returns the folder containing the defects, loading it if necessary

Definition at line 67 of file folders.py.

     def defects_folder(self) -> cool.IFolder:
         """
         Returns the folder containing the defects, loading it if necessary
         """
         if self._defects_folder is None: self._load_defects_folder()
         return self._defects_folder
     

◆ defects_tag()

str python.tags.DefectsDBTagsMixin.defects_tag ( self )

inherited

Return the tag used for retrieving defects on this DefectsDB instance

Definition at line 119 of file tags.py.

     def defects_tag(self) -> str:
         """
         Return the tag used for retrieving defects on this DefectsDB instance
         """
         if self._tag.defects == "HEAD": return "HEAD"
         try:
             return self.defects_folder.resolveTag(self._tag.defects)
         except Exception: 
             if self._read_only:
                 raise InvalidDefectTagError(self._tag.defects)
             # assume user knows what they're doing (e.g. writing to fresh tag)
             return self._tag.defects
         

◆ defects_tag_valid()

bool python.tags.DefectsDBTagsMixin.defects_tag_valid ( self )

inherited

Definition at line 93 of file tags.py.

     def defects_tag_valid(self) -> bool:
         try:
             self.defects_tag
         except InvalidTagError:
             return False
         else:
             return True
             

◆ defects_tags()

List[str] python.tags.DefectsDBTagsMixin.defects_tags ( self )

inherited

Returns a list of existing defect tags

Definition at line 34 of file tags.py.

     def defects_tags(self) -> List[str]:
         """
         Returns a list of existing defect tags
         """
         return [str(t) for t in self.defects_folder.listTags()]
     

◆ get_channel_descriptions()

MutableMapping[Union[str, int], str] python.ids.DefectsDBIDsNamesMixin.get_channel_descriptions	(		self,
		Iterable[Union[str, int]]	channels
	)

inherited

For the list of channel IDs "channels", return dict mapping ID to
description

Definition at line 246 of file ids.py.

     def get_channel_descriptions(self, channels: Iterable[Union[str, int]]) -> MutableMapping[Union[str, int], str]:
         """
         For the list of channel IDs "channels", return dict mapping ID to
         description
         """
         get_desc = self.defects_folder.channelDescription
         return {channel: get_desc(self.defect_chan_as_id(channel))
                      for channel in channels}
                      

◆ get_channels()

Tuple[Set[int], Set[str], Mapping[Union[str, int], Union[str, int]]] python.ids.DefectsDBIDsNamesMixin.get_channels ( self )

inherited

Return channel IDs, names, and dict relating the two

Definition at line 234 of file ids.py.

     def get_channels(self) -> Tuple[Set[int], Set[str], Mapping[Union[str, int], Union[str, int]]]:
         """
         Return channel IDs, names, and dict relating the two
         """
         return self.defect_ids, self.defect_names, self.defect_id_map
     

◆ get_intolerable_defects()

def python.virtual_mixin.DefectsDBVirtualDefectsMixin.get_intolerable_defects	(		self,
		bool	old_primary_only = `True`,
		Sequence[str]	exclude = `['TIGHT', 'IDTIGHT', 'PHYS_.*']`
	)

inherited

Returns primary defects that are depended on by a virtual defect
if old_primary_only == True, only return those depended on by a
virtual defect with no _ in the name
ignores virtual defects in exclude list
accepts regexes in exclude list

Definition at line 248 of file virtual_mixin.py.

     def get_intolerable_defects(self, old_primary_only: bool = True,
                                 exclude: Sequence[str] = ['TIGHT', 'IDTIGHT', 'PHYS_.*']):
         """
         Returns primary defects that are depended on by a virtual defect
         if old_primary_only == True, only return those depended on by a
         virtual defect with no _ in the name
         ignores virtual defects in exclude list
         accepts regexes in exclude list
         """
         vdl = self.virtual_defect_logics
         if old_primary_only:
             vdl = dict(l for l in vdl.items() if '_' not in l[0])
         if len(exclude) > 0:
             import re
             rex = re.compile('(%s)' % '|'.join(exclude))
             vdl = dict(l for l in vdl.items() if not rex.match(l[0]))
         return self.resolve_primary_defects(vdl.values())
         

◆ get_virtual_channel_descriptions()

MutableMapping[Union[str, int], str] python.ids.DefectsDBIDsNamesMixin.get_virtual_channel_descriptions	(		self,
		Iterable[Union[str, int]]	channels
	)

inherited

For the list of channel IDs "channels", return dict mapping ID to
descriptiondefect_id

Definition at line 255 of file ids.py.

     def get_virtual_channel_descriptions(self, channels: Iterable[Union[str, int]]) -> MutableMapping[Union[str, int], str]:
         """
         For the list of channel IDs "channels", return dict mapping ID to
         descriptiondefect_id
         """
         get_desc = self.defect_logic_folder.channelDescription
         return {channel: get_desc(self.defect_chan_as_id(channel))
                     for channel in channels}
  

◆ get_virtual_channels()

Tuple[Set[int], Set[str], Mapping[Union[str, int], Union[str, int]]] python.ids.DefectsDBIDsNamesMixin.get_virtual_channels ( self )

inherited

Return channel IDs, names, and dict relating the two

Definition at line 240 of file ids.py.

     def get_virtual_channels(self) -> Tuple[Set[int], Set[str], Mapping[Union[str, int], Union[str, int]]]:
         """
         Return channel IDs, names, and dict relating the two
         """
         return self.virtual_defect_ids, self.virtual_defect_names, self.virtual_defect_id_map
     

◆ insert()

None python.db.DefectsDB.insert	(		self,
		Union[str, int]	defect_id,
		int	since,
		int	until,
		str	comment,
		str	added_by,
		bool	present = `True`,
		bool	recoverable = `False`
	)

Insert a new defect into the database.

Parameters:
    `defect_id` : The name or channel identifier for the deect
    `since`, `until` : The COOL IoV for the range
    `comment`   : String255 arbitrary text comment
    `added_by`  : The user name or "sys:"-prefixed string of the
          application that inserted the defect
    `present`   : The state of the flag (Default: True)
    `recoverable` : Indicates whether there is any possibility to set
            present=False in the future (Default: False)

Definition at line 302 of file DQDefects/python/db.py.

     def insert(self, defect_id: Union[str, int], since: int, until: int, comment: str, added_by: str, 
                present: bool = True, recoverable: bool = False) -> None:
         """
         Insert a new defect into the database.
         
         Parameters:
             `defect_id` : The name or channel identifier for the deect
             `since`, `until` : The COOL IoV for the range
             `comment`   : String255 arbitrary text comment
             `added_by`  : The user name or "sys:"-prefixed string of the
                           application that inserted the defect
             `present`   : The state of the flag (Default: True)
             `recoverable` : Indicates whether there is any possibility to set
                             present=False in the future (Default: False)
         """
         return self._insert(defect_id, since, until, comment, added_by,
                             present, recoverable, self.defects_tag)
  

◆ insert_multiple()

def python.db.DefectsDB.insert_multiple	(		self,
		Iterable[IOVType]	defect_list,
		str	tag = `'HEAD'`,
		bool	use_flask = `False`,
		str	flask_cool_target = `'oracle://ATONR_COOLOFL_GPN/ATLAS_COOLOFL_GLOBAL'`,
		Mapping[str, str]	flask_auth = `{}`,
		str	flask_db = `'CONDBR2'`,
		str	flask_uri = `'https://cool-proxy-app.cern.ch/cool/multi_iovs'`
	)

Definition at line 352 of file DQDefects/python/db.py.

     def insert_multiple(self, defect_list: Iterable[IOVType],
                         tag: str = 'HEAD', use_flask: bool = False,
                         flask_cool_target: str = 'oracle://ATONR_COOLOFL_GPN/ATLAS_COOLOFL_GLOBAL',
                         flask_auth: Mapping[str, str] = {},
                         flask_db: str = 'CONDBR2',
                         flask_uri: str = 'https://cool-proxy-app.cern.ch/cool/multi_iovs'
                         ):
         if not use_flask:
             for defect in defect_list:
                 self._insert_iov(defect, tag)
         else:
             # Allow for override from environment
             import os
             flask_uri = os.environ.get('DQM_COOL_FLASK_URI', flask_uri)
  
             log.debug(f'Flask server URI: {flask_uri}')
             self._insert_multiple_flask(defect_list, tag, flask_cool_target, 
                                         flask_auth, flask_db, flask_uri)
  

◆ logics_tag()

str python.tags.DefectsDBTagsMixin.logics_tag ( self )

inherited

Return the tag used for retrieving virtual defect clauses

Definition at line 133 of file tags.py.

     def logics_tag(self) -> str:
         """
         Return the tag used for retrieving virtual defect clauses
         """
         if self._tag.logic == "HEAD": return "HEAD"
         try:
             return self.defect_logic_folder.resolveTag(self._tag.logic)
         except Exception:
             if self._read_only:
                 raise InvalidLogicTagError(self._tag.logic)
             # assume user knows what they're doing (e.g. writing to fresh tag)
             return self._tag.logic
     

◆ logics_tag_valid()

bool python.tags.DefectsDBTagsMixin.logics_tag_valid ( self )

inherited

Definition at line 102 of file tags.py.

     def logics_tag_valid(self) -> bool:
         try:
             self.logics_tag
         except InvalidTagError:
             return False
         else:
             return True
     

◆ logics_tags()

List[str] python.tags.DefectsDBTagsMixin.logics_tags ( self )

inherited

Returns a list of existing logic tags

Definition at line 27 of file tags.py.

     def logics_tags(self) -> List[str]:
         """
         Returns a list of existing logic tags
         """
         return [str(t) for t in self.defect_logic_folder.listTags()]
  

◆ new_defects_tag()

str python.tags.DefectsDBTagsMixin.new_defects_tag	(		self,
		str	name,
		str	description,
		Optional[Iterable]	iovranges = `None`
	)

inherited

Clones the current DEFECTS tag (specified in the constructor) to a new one
If iovranges != None, does the slower thing of copying IOVs one by one

Parameters:
    `name` : Name of the new tag
    `description` : Description of the contents of this tag

Definition at line 175 of file tags.py.

     def new_defects_tag(self, name: str, description: str, iovranges: Optional[Iterable] = None) -> str:
         """
         Clones the current DEFECTS tag (specified in the constructor) to a new one
         If iovranges != None, does the slower thing of copying IOVs one by one
         
         Parameters:
             `name` : Name of the new tag
             `description` : Description of the contents of this tag
         """
         if name.startswith("DetStatus"):
             raise RuntimeError("Only specify the last part of the defect tag")
         adescription = description.encode('utf-8')
         
         log.info("Creating new tag %s", name)
         
         name = f"DetStatusDEFECTS-{name}"
         aname = name.encode('ascii')
         
         if iovranges is None:
             self.defects_folder.cloneTagAsUserTag(self.defects_tag.encode('ascii'), aname, adescription)
             return name
         
         # Fetch all primary defects relevant to `iovranges`
         kwargs = dict(primary_only=True, nonpresent=True, intersect=True)
         iovsets = (self.retrieve(*iovrange, **kwargs) for iovrange in iovranges)
         to_copy = IOVSet.from_iovsets(iovsets)
         
         log.info("Copying IoVs: %r", to_copy)
         
         with self.storage_buffer:
             for iov in to_copy:
                 self._insert_iov(iov, name)
         
         # If there are no IOVs to copy, there is no new tag to describe
         if to_copy:
             self.defects_folder.setTagDescription(aname, adescription)
             
         return name

◆ new_hierarchical_tag()

str python.tags.DefectsDBTagsMixin.new_hierarchical_tag	(		self,
		str	defects_tag,
		str	logics_tag
	)

inherited

Create a new heirarchical tag relating the defects and logics

New tag is of the form
"DetStatus-[Logic revision number]-[Defect tagname]"
and has description
"(v%i) blah"

Definition at line 61 of file tags.py.

     def new_hierarchical_tag(self, defects_tag: str, logics_tag: str) -> str:
         """
         Create a new heirarchical tag relating the defects and logics
         
         New tag is of the form
         "DetStatus-[Logic revision number]-[Defect tagname]"
         and has description
         "(v%i) blah"
         """
         logic_revision = int(logics_tag.split("-")[-1])
         defect_part = "-".join(defects_tag.split("-")[1:])
         hierarchical_tag = "DetStatus-v%02i-%s" % (logic_revision, defect_part)
         logicspart = "(%i) " % logic_revision
         
         adefects_tag = defects_tag.encode('ascii')
         alogics_tag = logics_tag.encode('ascii')
         ahierarchical_tag = hierarchical_tag.encode('ascii')
         logicspart = logicspart.encode('ascii')
         defect_descr = self.defects_folder.tagDescription(adefects_tag).encode('utf-8')
         
         # Protection here against making descriptions too long
         description = logicspart + defect_descr[:255 - len(logicspart)]
         
         parent_folder = self.parent_folderset
         self.defects_folder.createTagRelation(ahierarchical_tag, adefects_tag)
         self.defect_logic_folder.createTagRelation(ahierarchical_tag, alogics_tag)
         parent_folder.setTagDescription(ahierarchical_tag, description)
         
         log.info("New hierarchical tag %s", hierarchical_tag)
         return hierarchical_tag
     

◆ new_logics_tag()

str python.tags.DefectsDBTagsMixin.new_logics_tag	(		self,
		str	description = `""`
	)

inherited

Create a new tag for the logic folder and lock it.

Parameters:
    `description` : What changed in this tag? (optional, default "")

Definition at line 160 of file tags.py.

     def new_logics_tag(self, description: str = "") -> str:
         """
         Create a new tag for the logic folder and lock it.
         
         Parameters:
             `description` : What changed in this tag? (optional, default "")
         """
         assert self.logics_tag == "HEAD", NONHEAD_MODIFICATION_MSG
         
         new_tag_name = self.next_logics_tag
         folder = self.defect_logic_folder
         self._tag_head_and_lock(folder, new_tag_name, description)
         log.info("Tagged DEFECTLOGICS HEAD with %s", new_tag_name)
         return new_tag_name
     

◆ new_virtual_defect()

int python.virtual_mixin.DefectsDBVirtualDefectsMixin.new_virtual_defect	(		self,
		str	defect_name,
		Optional[str]	comment,
		str	clauses
	)

inherited

Create a new virtual defect

Definition at line 149 of file virtual_mixin.py.

     def new_virtual_defect(self, defect_name: str, comment: Optional[str], clauses: str) -> int:
         """
         Create a new virtual defect
         """
         assert self.logics_tag == "HEAD", NONHEAD_MODIFICATION_MSG
         assert not self._read_only, "Insertion on read-only database"
         from DQUtils.channel_mapping import get_channel_ids_names
         
         # Force load of defects_folder to populate _defect_payload
         store = self.defect_logic_folder.storeObject
         p = self._defect_logic_payload
         
         self.validate_clauses(clauses)
  
         p["clauses"] = clauses.encode('ascii')
  
         # need to get true mapping of folder, not just what is visible
         # from this tag
         ids, _, mapping = get_channel_ids_names(self.defect_logic_folder)
         defect_id = choose_new_defect_id(mapping, defect_name, True)
         log.debug("Creating new defect %s: 0x%08x", defect_name, defect_id)
         
         try:
             oldname = self.normalize_defect_names(defect_name)
             raise DefectExistsError('Defect %s already exists' % oldname)
         except DefectUnknownError:
             pass
  
         if (defect_id in ids):
             raise DefectExistsError(defect_name)
         self.defect_logic_folder.createChannel(defect_id, defect_name.encode('ascii'), comment.encode('utf-8'))
         store(0, 2**63-1, p, defect_id)
         self._new_virtual_defect(defect_id, defect_name)
         return defect_id
     

◆ next_logics_tag()

str python.tags.DefectsDBTagsMixin.next_logics_tag ( self )

inherited

Gives the next available DEFECTLOGICS tag

Definition at line 41 of file tags.py.

     def next_logics_tag(self) -> str:
         """
         Gives the next available DEFECTLOGICS tag
         """
         existing_tags = self.logics_tags
         if existing_tags:
             new_id = max(int(str(t).split("-")[-1]) for t in existing_tags) + 1
         else:
             new_id = 0
         return DEFECT_LOGIC_TAG_FORMAT % new_id
     

◆ normalize_defect_names()

Union[str, List[str]] python.ids.DefectsDBIDsNamesMixin.normalize_defect_names	(		self,
		Union[str, Iterable[str]]	defect_id
	)

inherited

Returns correct name(s) of defects, given name(s) that possibly differ
from the correct ones by case.  Raises if an input name doesn't map to
any existing defect.  You can pass either a string or an iterable
object as `defect_id`.

Definition at line 308 of file ids.py.

     def normalize_defect_names(self, defect_id: Union[str, Iterable[str]]) -> Union[str, List[str]]:
         """
         Returns correct name(s) of defects, given name(s) that possibly differ
         from the correct ones by case.  Raises if an input name doesn't map to
         any existing defect.  You can pass either a string or an iterable
         object as `defect_id`.
         """
         wasstring = False
         if isinstance(defect_id, six.string_types):
             defect_id = [defect_id]
             wasstring = True
         if any(not isinstance(i, six.string_types) for i in defect_id):
             raise ValueError('All input values must be strings')
  
         uppered_defects = dict((i.upper(), i) for i in self.defect_names)
         uppered_defects.update(dict((i.upper(), i) for i in self.virtual_defect_names))
  
         rv = []
         for i in defect_id:
             up = i.upper()
             if up in uppered_defects:
                 rv.append(uppered_defects[up])
             else:
                 raise DefectUnknownError(i)
  
         if wasstring:
             return rv[0]
         else:
             return rv

◆ parent_folderset()

cool.IFolderSet python.folders.DefectsDBFoldersMixin.parent_folderset ( self )

inherited

Lazy-loads parent folderset (useful for heirarchical tag manipulation)

Definition at line 36 of file folders.py.

     def parent_folderset(self) -> cool.IFolderSet:
         """
         Lazy-loads parent folderset (useful for heirarchical tag manipulation)
         """
         if self._parent_folderset is None:
             db = Databases.get_instance(self.connection_string, 
                                         read_only=self._read_only)
             self._parent_folderset = db.getFolderSet(PARENT_FOLDERSET)
         return self._parent_folderset
     

◆ rename_defect()

None python.virtual_mixin.DefectsDBVirtualDefectsMixin.rename_defect	(		self,
		str	defect_name,
		str	new_defect_name
	)

inherited

Rename a defect (primary or virtual).  Will keep data and channel ID.
Will fix up all virtual defect dependencies in all tags.

Definition at line 84 of file virtual_mixin.py.

     def rename_defect(self, defect_name: str, new_defect_name: str) -> None:
         """
         Rename a defect (primary or virtual).  Will keep data and channel ID.
         Will fix up all virtual defect dependencies in all tags.
         """
         assert not self._read_only, "Channel rename on read-only database"
         
         try:
             oldname = self.normalize_defect_names(new_defect_name)
             raise DefectExistsError('Defect %s already exists' % oldname)
         except DefectUnknownError:
             pass
         
         defect_id = self.defect_chan_as_id(defect_name)
         
         anew_defect_name = new_defect_name.encode('ascii')
         if self.defect_is_virtual(defect_id):
             self.defect_logic_folder.setChannelName(defect_id, anew_defect_name)
             self._virtual_defect_names = set()
             self._virtual_defect_id_map = {}
             self._virtual_defect_logics = None
             self._virtual_initialized = False
         else:
             self.defects_folder.setChannelName(defect_id, anew_defect_name)
             self._defect_names = set()
             self._defect_id_map = {}
             self._initialized = False
         
         import contextlib
         @contextlib.contextmanager
         def logics_unlocking(tag):
             atag = tag.encode('ascii')
             log.info('Unlocking tag %s', tag)
             folder = self.defect_logic_folder
             orig_status = folder.tagLockStatus(atag)
             if orig_status != 0:
                 folder.setTagLockStatus(atag, 0)
             yield
             if orig_status != 0:
                 folder.setTagLockStatus(atag, orig_status)
             log.info('Done with tag %s', tag)
  
         def logic_substitute(clause, oldname, newname):
             oclause = clause
             prefix = ''
             if clause[0] in ('!',):
                 prefix = clause[0]
                 clause = clause[1:]
             if clause != oldname:
                 return oclause
             else:
                 return prefix + newname
         
         for tag in ['HEAD'] + self.logics_tags:
             logics = self._get_virtual_defect_logics(tag)
             for vd, logic in logics.items():
                 if defect_name in logic.realclauses:
                     newclauses = ' '.join([logic_substitute(x, defect_name,
                                                             new_defect_name)
                                            for x in logic.clauses
                                            ])
                     with logics_unlocking(tag):
                         self._update_virtual_defect(vd, newclauses, tag=tag)
         self._virtual_defect_logics = None
     

◆ resolve_primary_defects()

Set[str] python.virtual_mixin.DefectsDBVirtualDefectsMixin.resolve_primary_defects	(		self,
		Iterable[DefectLogic]	defect_logics
	)

inherited

Determine which primary flags are used for 
a given list of input `virtual_defect_names`.

Definition at line 209 of file virtual_mixin.py.

     def resolve_primary_defects(self, defect_logics: Iterable[DefectLogic]) -> Set[str]:
         """
         Determine which primary flags are used for 
         a given list of input `virtual_defect_names`.
         """
         primary_defects = set()
         for defect_logic in defect_logics:
             primary_defects |= defect_logic.primary_defects
             
         return primary_defects
     

◆ retrieve()

IOVSet python.db.DefectsDB.retrieve	(		self,
		Optional[Union[int, Tuple[int,int], RunLumi]]	since = `None`,
		Optional[Union[int, Tuple[int,int], RunLumi]]	until = `None`,
		Optional[Iterable[Union[str,int]]]	channels = `None`,
		bool	nonpresent = `False`,
		bool	primary_only = `False`,
		Optional[Collection[str]]	ignore = `None`,
		bool	with_primary_dependencies = `False`,
		bool	intersect = `False`,
		bool	with_time = `False`,
		bool	evaluate_full = `True`
	)

Retrieve defects from the database.

Parameters:
    `since`, `until` : IoV range to query (Default: All)
    `channels`  : A list of channels to query. Can contain a mixture of
         defect names and ids. (Default: None means all 
        channels, including all virtual)
    `nonpresent` : Only return IoVs which are currently "present"
    `primary_only` : Only return primary IoVs, no virtual ones.
    `ignore` : Set of defects which won't be treated as bad.
    `with_primary_dependencies` : When querying virtual flags, also get
primary flags which went into the calculation.
    `intersect` : Intersect the result with the query range so that no
          iov spans outside the query range
    `with_time` : Also retrieves the time the defect was inserted
          ~2x slower queries, doesn't work for virtual defects
    `evaluate_full` : If specified, also compute the `comment` and
              `recoverable` fields of virtual defects.
              Causes a ~0.6x slowdown

Definition at line 159 of file DQDefects/python/db.py.

     def retrieve(self, since: Optional[Union[int, Tuple[int,int], RunLumi]] = None,
                  until: Optional[Union[int, Tuple[int,int], RunLumi]] = None, 
                  channels: Optional[Iterable[Union[str,int]]] = None, nonpresent: bool = False,
                  primary_only: bool = False, ignore: Optional[Collection[str]] = None,
                  with_primary_dependencies: bool = False, intersect: bool = False,
                  with_time: bool = False, evaluate_full: bool = True) -> IOVSet:
         """
         Retrieve defects from the database.
         
         Parameters:
             `since`, `until` : IoV range to query (Default: All)
             `channels`  : A list of channels to query. Can contain a mixture of
                          defect names and ids. (Default: None means all 
                         channels, including all virtual)
             `nonpresent` : Only return IoVs which are currently "present"
             `primary_only` : Only return primary IoVs, no virtual ones.
             `ignore` : Set of defects which won't be treated as bad.
             `with_primary_dependencies` : When querying virtual flags, also get
                 primary flags which went into the calculation.
             `intersect` : Intersect the result with the query range so that no
                           iov spans outside the query range
             `with_time` : Also retrieves the time the defect was inserted
                           ~2x slower queries, doesn't work for virtual defects
             `evaluate_full` : If specified, also compute the `comment` and
                               `recoverable` fields of virtual defects.
                               Causes a ~0.6x slowdown
         """
         if ignore is not None and not isinstance(ignore, set):
             raise RuntimeError("ignore parameter should be set type")
         
         desired_channels = None
         # Figure out the IDs of channels to query and their virtuality
         if channels is not None:
             query_channels = set(self.defect_names_as_ids(channels))
             virtual_channels = query_channels - self.defect_ids
             primary_channels = query_channels & self.defect_ids
         else:
             # Empty channels list means query all channels
             # (including all virtual)
             if primary_only:
                 virtual_channels = None
             else:
                 virtual_channels = self.virtual_defect_ids
             primary_channels = set(self.defect_ids)
             query_channels = None # (all)
         
         primary_output_names = [self.defect_id_map[pid] for pid in primary_channels]
         virtual_output_names = [] # (optionally populated below)
         
         # Resolve virtual channels to query here, and the primary dependents.
         if virtual_channels:
             assert not primary_only, "Requested virtual channels with primary_only=True"
             assert not with_time, "with_time flag only works for primary defects"
             virtual_output_names = [self.virtual_defect_id_map[vid] 
                                     for vid in virtual_channels]
             
             ordered_logics = self._resolve_evaluation_order(virtual_output_names)
             
             if channels is not None:
                 # Since not all channels are being queried, it is necessary to
                 # add the desired primary channels to the query
                 primary_needed = self.resolve_primary_defects(ordered_logics)
                 primary_channels |= set(self.defect_names_as_ids(primary_needed))
                 query_channels = primary_channels
                 
                 if with_primary_dependencies:
                     primary_output_names.extend(sorted(primary_needed))
         
             for logic in ordered_logics:
                 logic.set_evaluation(evaluate_full)
         else:
             ordered_logics = []
         
         # Figure out if the set of channels will produce too many ranges for COOL
         if query_channels is not None:
             query_channels = sorted(query_channels)
             query_ranges = list_to_channelselection(query_channels, None, True)
             
             if len(query_ranges) >= 50:
                 # We're querying too many ranges. Query everything, filter later.
                 desired_channels = set(primary_output_names + virtual_output_names)
                 query_channels = None # (everything)
             
         # Retrieve primary IoVs
         primary_iovs = fetch_iovs(self.defects_folder, since, until, 
                                   query_channels, self.defects_tag, 
                                   named_channels=True, unicode_strings=True,
                                   with_time=with_time)
         
         # Calculate virtual defects (if necessary)
         if primary_only or not virtual_channels:
             result = primary_iovs
         else:
             if not primary_iovs:
                 return IOVSet()
             args = (primary_iovs, ordered_logics, 
                     virtual_output_names, primary_output_names, 
                     since, until, ignore)
             result = calculate_virtual_defects(*args)
             
         # Filter out results which have their present bit removed 
         # (unless otherwise specified)
         if not nonpresent:
             result = IOVSet(iov for iov in result if iov.present)
         
         # Filter out channels which weren't queried by the user
         # (to get around 50 channelselection COOL limit)
         if desired_channels:
             result = IOVSet(iov for iov in result 
                             if iov.channel in desired_channels)
         
         if intersect:
             result = result.intersect_range((since, until))
         
         return result
     

◆ set_channel_description()

None python.ids.DefectsDBIDsNamesMixin.set_channel_description	(		self,
		Union[str, int]	channel,
		str	description
	)

inherited

Set a defect description

Definition at line 274 of file ids.py.

     def set_channel_description(self, channel: Union[str, int], description: str) -> None:
         """
         Set a defect description
         """
         chan_id = self.defect_chan_as_id(channel)
         if self.defect_is_virtual(chan_id):
             folder = self.defect_logic_folder
         else:
             folder = self.defects_folder
         folder.setChannelDescription(chan_id, description.encode('utf-8'))
     

◆ storage_buffer()

def python.db.DefectsDB.storage_buffer ( self )

Gives a new context manager for use with the with statement, e.g.:
    with ddb.storage_buffer:
for d in many_defects:
    ddb.insert(...defect...)

Definition at line 276 of file DQDefects/python/db.py.

     def storage_buffer(self):
         """
         Gives a new context manager for use with the with statement, e.g.:
             with ddb.storage_buffer:
                 for d in many_defects:
                     ddb.insert(...defect...)
         """
         assert not self._bad_state, "Please see comment in DefectsDB constructor"
         
         @contextmanager
         def thunk():
             log.debug("setupStorageBuffer()")
             self.defects_folder.setupStorageBuffer()
             try:
                 yield
             except Exception:
                 log.warning("Exception raised during DefectsDB.storage_buffer. "
                             "Not flushing storage buffer - but COOL has no way "
                             "to empty it. ")
                 self._bad_state = True
                 raise
             else:
                 log.debug("flushStorageBuffer()")
                 self.defects_folder.flushStorageBuffer()
         return thunk()
     

◆ tag()

tagtype python.tags.DefectsDBTagsMixin.tag ( self )

inherited

Readonly property stating the tag that the database was instantiated on

Definition at line 147 of file tags.py.

     def tag(self) -> tagtype:
         """
         Readonly property stating the tag that the database was instantiated on
         """
         return self._tag
     

◆ tags()

List[str] python.tags.DefectsDBTagsMixin.tags ( self )

inherited

The list of tags which are on the database

Definition at line 154 of file tags.py.

     def tags(self) -> List[str]:
         """
         The list of tags which are on the database
         """
         return [str(t) for t in self.parent_folderset.listTags()]
     

◆ update_virtual_defect()

None python.virtual_mixin.DefectsDBVirtualDefectsMixin.update_virtual_defect	(		self,
		str	defect_name,
		str	clauses,
		Optional[str]	comment = `None`
	)

inherited

Definition at line 56 of file virtual_mixin.py.

     def update_virtual_defect(self, defect_name: str, clauses: str, comment: Optional[str] = None) -> None:
         # HEAD protection is here, to allow other functions to
         # change other tags in a known manner with _update_virtual_defect
         assert self.logics_tag == "HEAD", NONHEAD_MODIFICATION_MSG
         assert not self._read_only, "Insertion on read-only database"
         self._update_virtual_defect(defect_name, clauses, comment)
     

◆ validate_clauses()

None python.virtual_mixin.DefectsDBVirtualDefectsMixin.validate_clauses	(		self,
		str	clauses
	)

inherited

Definition at line 49 of file virtual_mixin.py.

     def validate_clauses(self, clauses: str) -> None:
         all_defects = self.defect_names | self.virtual_defect_names
         for clause in clauses.split():
             if clause[0] in ('!', '-'):
                 clause = clause[1:]
             assert clause in all_defects, (clause + " is not a known defect")
     

◆ virtual_defect_id_map()

MutableMapping[Union[str,int], Union[str,int]] python.ids.DefectsDBIDsNamesMixin.virtual_defect_id_map ( self )

inherited

Returns a dict() mapping virtual defect names to IDs and vice-versa.

Definition at line 194 of file ids.py.

     def virtual_defect_id_map(self) -> MutableMapping[Union[str,int], Union[str,int]]:
         """
         Returns a dict() mapping virtual defect names to IDs and vice-versa.
         """
         if not self._virtual_initialized: 
             self._populate_virtual_defect_ids()
             self._virtual_initialized = True
         return self._virtual_defect_id_map
         

◆ virtual_defect_ids()

Set[int] python.ids.DefectsDBIDsNamesMixin.virtual_defect_ids ( self )

inherited

Returns the set of existing virtual defect IDs

Definition at line 174 of file ids.py.

     def virtual_defect_ids(self) -> Set[int]:
         """
         Returns the set of existing virtual defect IDs
         """
         if not self._virtual_initialized: 
             self._populate_virtual_defect_ids()
             self._virtual_initialized = True
         return self._virtual_defect_ids
     

◆ virtual_defect_logics()

Mapping[str, DefectLogic] python.virtual_mixin.DefectsDBVirtualDefectsMixin.virtual_defect_logics ( self )

inherited

Returns all virtual defects in the form {"defect_name" : DefectLogic()}
for the tag DefectDB was constructed on.

Definition at line 221 of file virtual_mixin.py.

     def virtual_defect_logics(self) -> Mapping[str, DefectLogic]:
         """
         Returns all virtual defects in the form {"defect_name" : DefectLogic()}
         for the tag DefectDB was constructed on.
         """
         if self._virtual_defect_logics is not None:
             return self._virtual_defect_logics
                 
         self._virtual_defect_logics = self._get_virtual_defect_logics(self.logics_tag)
         
         return self._virtual_defect_logics
  

◆ virtual_defect_names()

Set[str] python.ids.DefectsDBIDsNamesMixin.virtual_defect_names ( self )

inherited

Returns the set of existing virtual defect names

Definition at line 184 of file ids.py.

     def virtual_defect_names(self) -> Set[str]:
         """
         Returns the set of existing virtual defect names
         """
         if not self._virtual_initialized: 
             self._populate_virtual_defect_ids()
             self._virtual_initialized = True
         return self._virtual_defect_names
     

Member Data Documentation

◆ _bad_state

python.db.DefectsDB._bad_state

private

Definition at line 103 of file DQDefects/python/db.py.

◆ _connections

python.folders.DefectsDBFoldersMixin._connections

privateinherited

Definition at line 26 of file folders.py.

◆ _create

python.db.DefectsDB._create

private

Definition at line 84 of file DQDefects/python/db.py.

◆ _defect_id_map

python.virtual_mixin.DefectsDBVirtualDefectsMixin._defect_id_map

privateinherited

Definition at line 109 of file virtual_mixin.py.

◆ _defect_ids

python.ids.DefectsDBIDsNamesMixin._defect_ids

privateinherited

Definition at line 97 of file ids.py.

◆ _defect_logic_folder

python.folders.DefectsDBFoldersMixin._defect_logic_folder

privateinherited

Definition at line 24 of file folders.py.

◆ _defect_logic_payload

python.folders.DefectsDBFoldersMixin._defect_logic_payload

privateinherited

Definition at line 113 of file folders.py.

◆ _defect_names

python.virtual_mixin.DefectsDBVirtualDefectsMixin._defect_names

privateinherited

Definition at line 108 of file virtual_mixin.py.

◆ _defect_payload

python.folders.DefectsDBFoldersMixin._defect_payload

privateinherited

Definition at line 80 of file folders.py.

◆ _defects_folder

python.folders.DefectsDBFoldersMixin._defects_folder

privateinherited

Definition at line 23 of file folders.py.

◆ _initialized

python.virtual_mixin.DefectsDBVirtualDefectsMixin._initialized

privateinherited

Definition at line 110 of file virtual_mixin.py.

◆ _parent_folderset

python.folders.DefectsDBFoldersMixin._parent_folderset

privateinherited

Definition at line 25 of file folders.py.

◆ _read_only

python.db.DefectsDB._read_only

private

Definition at line 83 of file DQDefects/python/db.py.

◆ _tag

python.db.DefectsDB._tag

private

Definition at line 86 of file DQDefects/python/db.py.

◆ _virtual_defect_id_map

python.virtual_mixin.DefectsDBVirtualDefectsMixin._virtual_defect_id_map

privateinherited

Definition at line 103 of file virtual_mixin.py.

◆ _virtual_defect_ids

python.ids.DefectsDBIDsNamesMixin._virtual_defect_ids

privateinherited

Definition at line 110 of file ids.py.

◆ _virtual_defect_logics

python.virtual_mixin.DefectsDBVirtualDefectsMixin._virtual_defect_logics

privateinherited

Definition at line 46 of file virtual_mixin.py.

◆ _virtual_defect_map

python.ids.DefectsDBIDsNamesMixin._virtual_defect_map

privateinherited

Definition at line 119 of file ids.py.

◆ _virtual_defect_names

python.virtual_mixin.DefectsDBVirtualDefectsMixin._virtual_defect_names

privateinherited

Definition at line 102 of file virtual_mixin.py.

◆ _virtual_initialized

python.virtual_mixin.DefectsDBVirtualDefectsMixin._virtual_initialized

privateinherited

Definition at line 105 of file virtual_mixin.py.

◆ connection_string

python.db.DefectsDB.connection_string

Definition at line 82 of file DQDefects/python/db.py.

◆ logics_tag [1/2]

python.virtual_mixin.DefectsDBVirtualDefectsMixin.logics_tag

inherited

Definition at line 59 of file virtual_mixin.py.

◆ logics_tag [2/2]

python.tags.DefectsDBTagsMixin.logics_tag

inherited

Definition at line 167 of file tags.py.

The documentation for this class was generated from the following file:

DQDefects/python/db.py

Private Attributes
	_read_only

	_create

	_tag

	_bad_state

	_virtual_defect_logics

	_virtual_defect_names

	_virtual_defect_id_map

	_virtual_initialized

	_defect_names

	_defect_id_map

	_initialized

	_defects_folder

	_defect_logic_folder

	_parent_folderset

	_connections

	_defect_payload

	_defect_logic_payload

	_defect_ids

	_virtual_defect_ids

	_virtual_defect_map

Public Member Functions

Public Attributes

Private Member Functions

Private Attributes

Detailed Description

Constructor & Destructor Documentation

◆ __init__()

◆ __del__()

Member Function Documentation

◆ _clear_connections()

◆ _create_defect_logic_folder()

◆ _create_defect_with_id()

◆ _create_defects_folder()

◆ _get_virtual_defect_logics()

◆ _insert()

◆ _insert_iov()

◆ _insert_multiple_flask()

◆ _load_defect_logic_folder()

◆ _load_defects_folder()

◆ _load_folder()

◆ _new_defect()

◆ _new_virtual_defect()

◆ _populate_defect_ids()

◆ _populate_virtual_defect_ids()

◆ _resolve_evaluation_order()

◆ _tag_head_and_lock()

◆ _update_virtual_defect()

◆ _virtual_defect_consistency_check()

◆ all_defect_descriptions()

◆ check_tag_validity()

◆ create_defect()

◆ defect_chan_as_id()

◆ defect_id_map()

◆ defect_ids()

◆ defect_is_virtual()

◆ defect_logic_folder()

◆ defect_names()

◆ defect_names_as_ids()

◆ defects_folder()

◆ defects_tag()

◆ defects_tag_valid()

◆ defects_tags()

◆ get_channel_descriptions()

◆ get_channels()

◆ get_intolerable_defects()

◆ get_virtual_channel_descriptions()

◆ get_virtual_channels()

◆ insert()

◆ insert_multiple()

◆ logics_tag()

◆ logics_tag_valid()

◆ logics_tags()

◆ new_defects_tag()

◆ new_hierarchical_tag()

◆ new_logics_tag()

◆ new_virtual_defect()

◆ next_logics_tag()

◆ normalize_defect_names()

◆ parent_folderset()

◆ rename_defect()

◆ resolve_primary_defects()

◆ retrieve()

◆ set_channel_description()

◆ storage_buffer()

◆ tag()

◆ tags()

◆ update_virtual_defect()

◆ validate_clauses()

◆ virtual_defect_id_map()

◆ virtual_defect_ids()

◆ virtual_defect_logics()

◆ virtual_defect_names()

Member Data Documentation

◆ _bad_state

◆ _connections

◆ _create

◆ _defect_id_map

◆ _defect_ids

◆ _defect_logic_folder

◆ _defect_logic_payload

◆ init()

◆ del()