LSST Applications  21.0.0-172-gfb10e10a+18fedfabac,22.0.0+297cba6710,22.0.0+80564b0ff1,22.0.0+8d77f4f51a,22.0.0+a28f4c53b1,22.0.0+dcf3732eb2,22.0.1-1-g7d6de66+2a20fdde0d,22.0.1-1-g8e32f31+297cba6710,22.0.1-1-geca5380+7fa3b7d9b6,22.0.1-12-g44dc1dc+2a20fdde0d,22.0.1-15-g6a90155+515f58c32b,22.0.1-16-g9282f48+790f5f2caa,22.0.1-2-g92698f7+dcf3732eb2,22.0.1-2-ga9b0f51+7fa3b7d9b6,22.0.1-2-gd1925c9+bf4f0e694f,22.0.1-24-g1ad7a390+a9625a72a8,22.0.1-25-g5bf6245+3ad8ecd50b,22.0.1-25-gb120d7b+8b5510f75f,22.0.1-27-g97737f7+2a20fdde0d,22.0.1-32-gf62ce7b1+aa4237961e,22.0.1-4-g0b3f228+2a20fdde0d,22.0.1-4-g243d05b+871c1b8305,22.0.1-4-g3a563be+32dcf1063f,22.0.1-4-g44f2e3d+9e4ab0f4fa,22.0.1-42-gca6935d93+ba5e5ca3eb,22.0.1-5-g15c806e+85460ae5f3,22.0.1-5-g58711c4+611d128589,22.0.1-5-g75bb458+99c117b92f,22.0.1-6-g1c63a23+7fa3b7d9b6,22.0.1-6-g50866e6+84ff5a128b,22.0.1-6-g8d3140d+720564cf76,22.0.1-6-gd805d02+cc5644f571,22.0.1-8-ge5750ce+85460ae5f3,master-g6e05de7fdc+babf819c66,master-g99da0e417a+8d77f4f51a,w.2021.48
LSST Data Management Base Package
ingestDriver.py
Go to the documentation of this file.
1 from lsst.ctrl.pool.pool import Pool, startPool, abortOnError
2 from lsst.ctrl.pool.parallel import BatchCmdLineTask
3 from lsst.pipe.base import Struct
4 from lsst.pipe.tasks.ingest import IngestTask, IngestError
5 
6 
8  """Parallel version of IngestTask"""
9  @classmethod
10  def batchWallTime(cls, time, parsedCmd, numCores):
11  return float(time)*len(parsedCmd.files)/numCores
12 
13  @classmethod
14  def _makeArgumentParser(cls, *args, **kwargs):
15  """Build an ArgumentParser
16 
17  Removes the batch-specific parts.
18  """
19  kwargs.pop("doBatch", False)
20  kwargs.pop("add_help", False)
21  return cls.ArgumentParserArgumentParser(*args, name="ingest", **kwargs)
22 
23  @classmethod
24  def parseAndRun(cls, *args, **kwargs):
25  """Run with a MPI process pool"""
26  pool = startPool()
27  config = cls.ConfigClassConfigClass()
28  parser = cls.ArgumentParserArgumentParser(name=cls._DefaultName_DefaultName)
29  args = parser.parse_args(config)
30  task = cls(config=args.config)
31  task.run(args)
32  pool.exit()
33 
34  def runFileWrapper(self, struct, args):
35  """Run ingest on one file
36 
37  This is a wrapper method for calling ``runFile``.
38 
39  Parameters
40  ----------
41  struct : `lsst.pipe.base.Struct`
42  Structure containing ``filename`` (`str`) and ``position`` (`int`).
43  args : `argparse.Namespace`
44  Parsed command-line arguments.
45 
46  Returns
47  -------
48  hduInfoList : `list` of `dict`
49  Parsed information from FITS HDUs, or ``None``.
50  """
51  filename = struct.filename
52  position = struct.position
53  try:
54  return self.runFilerunFile(filename, None, args, position)
55  except IngestError as exc:
56  self.log.warn(f"Unable to ingest {filename}: {exc}")
57  return None
58 
59  @abortOnError
60  def run(self, args):
61  """Run ingest
62 
63  We read and ingest the files in parallel, and then
64  stuff the registry database in serial.
65  """
66  # Parallel
67  pool = Pool(None)
68  filenameList = self.expandFilesexpandFiles(args.files)
69  dataList = [Struct(filename=filename, position=ii) for ii, filename in enumerate(filenameList)]
70  infoList = pool.map(self.runFileWrapperrunFileWrapper, dataList, args)
71 
72  # Serial
73  root = args.input
74  context = self.register.openRegistry(root, create=args.create, dryrun=args.dryrun)
75  with context as registry:
76  for hduInfoList in infoList:
77  if hduInfoList is None:
78  continue
79  for info in hduInfoList:
80  self.register.addRow(registry, info, dryrun=args.dryrun, create=args.create)
81 
82  def writeConfig(self, *args, **kwargs):
83  pass
84 
85  def writeMetadata(self, *args, **kwargs):
86  pass
def writeConfig(self, *args, **kwargs)
Definition: ingestDriver.py:82
def batchWallTime(cls, time, parsedCmd, numCores)
Return walltime request for batch job.
Definition: ingestDriver.py:10
def writeMetadata(self, *args, **kwargs)
Definition: ingestDriver.py:85
def expandFiles(self, fileNameList)
Expand a set of filenames and globs, returning a list of filenames.
Definition: ingest.py:557
def runFile(self, infile, registry, args, pos)
Examine and ingest a single file.
Definition: ingest.py:576
def startPool(comm=None, root=0, killSlaves=True)
Start a process pool.
Definition: pool.py:1216