docs: def get_composite_source_model(oqparam, monitor=None, in_memory=True,
                                 srcfilter=SourceFilter(None, {})):
      """
      Parse the XML and build a complete composite source model in memory.
  
      :param oqparam:
          an :class:`openquake.commonlib.oqvalidation.OqParam` instance
      :param monitor:
           a `openquake.baselib.performance.Monitor` instance
      :param in_memory:
          if False, just parse the XML without instantiating the sources
      :param srcfilter:
          if not None, use it to prefilter the sources
      """
      ucerf = oqparam.calculation_mode.startswith('ucerf')
      source_model_lt = get_source_model_lt(oqparam, validate=not ucerf)
      trts = source_model_lt.tectonic_region_types
      trts_lower = {trt.lower() for trt in trts}
      reqv = oqparam.inputs.get('reqv', {})
      for trt in reqv:  # these are lowercase because they come from the job.ini
          if trt not in trts_lower:
              raise ValueError('Unknown TRT=%s in %s [reqv]' %
                               (trt, oqparam.inputs['job_ini']))
      gsim_lt = get_gsim_lt(oqparam, trts or ['*'])
      p = source_model_lt.num_paths * gsim_lt.get_num_paths()
      if oqparam.number_of_logic_tree_samples:
          logging.info('Considering {:,d} logic tree paths out of {:,d}'.format(
              oqparam.number_of_logic_tree_samples, p))
      else:  # full enumeration
          if oqparam.is_event_based() and p > oqparam.max_potential_paths:
              raise ValueError(
                  'There are too many potential logic tree paths (%d) '
                  'use sampling instead of full enumeration' % p)
          logging.info('Potential number of logic tree paths = {:,d}'.format(p))
  
      if source_model_lt.on_each_source:
          logging.info('There is a logic tree on each source')
      if monitor is None:
          monitor = performance.Monitor()
      smodels = []
      for source_model in get_source_models(
              oqparam, gsim_lt, source_model_lt, monitor, in_memory, srcfilter):
          for src_group in source_model.src_groups:
              src_group.sources = sorted(src_group, key=getid)
              for src in src_group:
                  # there are two cases depending on the flag in_memory:
                  # 1) src is a hazardlib source and has a src_group_id
                  #    attribute; in that case the source has to be numbered
                  # 2) src is a Node object, then nothing must be done
                  if isinstance(src, Node):
                      continue
          smodels.append(source_model)
      csm = source.CompositeSourceModel(gsim_lt, source_model_lt, smodels,
                                        oqparam.optimize_same_id_sources)
      for sm in csm.source_models:
          counter = collections.Counter()
          for sg in sm.src_groups:
              for srcid in map(getid, sg):
                  counter[srcid] += 1
          dupl = [srcid for srcid in counter if counter[srcid] > 1]
          if dupl:
              raise nrml.DuplicatedID('Found duplicated source IDs in %s: %s'
                                      % (sm, dupl))
      if not in_memory:
          return csm
  
      if oqparam.is_event_based():
          # initialize the rupture serial numbers before splitting/filtering; in
          # this way the serials are independent from the site collection
          csm.init_serials(oqparam.ses_seed)
  
      if oqparam.disagg_by_src:
          csm = csm.grp_by_src()  # one group per source
  
      csm.info.gsim_lt.check_imts(oqparam.imtls)
      parallel.Starmap.shutdown()  # save memory
      return csm