19150063725

Committed 06 Nov 2025 09:14PM UTC coverage: 37.719% (+0.7%) from 37.002%

Build # 19150063725

Build Type

Pull #2521

github

Committed by

web-flow

Commit Message

Merge b3bf9e090 into 6a90a0547

Pull Request Pull Request #2521: Add redshift QA scripts

Run Details

12990 of 34439 relevant lines covered (37.72%)

0.38 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/py/desispec/scripts/proc_tilenight.py

"""
desispec.scripts.proc_tilenight
===============================

Script for science processing of a given DESI tile and night
"""
import sys
import subprocess
import time, datetime
from desispec.workflow.batch_writer import create_desi_proc_tilenight_batch_script
import numpy as np

from os.path import dirname, abspath
from desiutil.log import get_logger, DEBUG, INFO

from desispec.io import specprod_root
from desispec.workflow.exptable import get_exposure_table_pathname
from desispec.workflow.tableio import load_table
from desispec.io.util import decode_camword, create_camword, camword_union, camword_intersection, difference_camwords

import desispec.scripts.proc as proc
import desispec.scripts.proc_joint_fit as proc_joint_fit

from desispec.workflow.desi_proc_funcs import assign_mpi, get_desi_proc_tilenight_parser
from desispec.workflow.desi_proc_funcs import update_args_with_headers

import desispec.gpu

#########################################
######## Begin Body of the Code #########
#########################################

def parse(options=None):
    parser = get_desi_proc_tilenight_parser()
    args = parser.parse_args(options)

    #- convert comma separated steps to list of str
    if isinstance(args.laststeps, str):
        args.laststeps = [laststep.strip().lower() for laststep in args.laststeps.split(',')]

    return args

def main(args=None, comm=None):
    if args is None:
        args = parse()

    log = get_logger()
    start_time = time.time()
    error_count = 0

    if comm is not None:
        #- Use the provided comm to determine rank and size
        rank = comm.rank
        size = comm.size
    else:
        #- Check MPI flags and determine the comm, rank, and size given the arguments
        comm, rank, size = assign_mpi(do_mpi=args.mpi, do_batch=args.batch, log=log)

    if rank == 0:
        thisfile=dirname(abspath(__file__))
        thistime=datetime.datetime.fromtimestamp(start_time).isoformat()
        log.info(f'Tilenight started main in {thisfile} at {thistime}')

    #- Read + broadcast exposure table for this night
    exptable = None
    if rank == 0:
        exptable_file = get_exposure_table_pathname(args.night)
        log.info(f'Reading exptable in {exptable_file}')
        exptable = load_table(exptable_file)

    if comm is not None:
        exptable = comm.bcast(exptable, root=0)

    #- Set the eligible laststep values for different processing steps
    if args.laststeps is None:
        prestd_laststeps = jntpst_laststeps = ['all', 'fluxcalib']
    else:
        prestd_laststeps = args.laststeps
        jntpst_laststeps = list(set(['all', 'fluxcalib']) & set(args.laststeps))

    #- Determine expids and cameras for a tile night
    keep  = exptable['OBSTYPE'] == 'science'
    keep &= exptable['TILEID']  == int(args.tileid)
    if args.laststeps is not None:
        keep &= np.isin(exptable['LASTSTEP'].astype(str), args.laststeps)

    #- Camera superset from args.cameras; if None default to all
    camera_superset = 'a0123456789'
    if args.cameras is not None:
        camera_superset = args.cameras

    exptable = exptable[keep]

    if len(exptable) == 0:
        if rank == 0:
            log.critical(f'No good exposures found for tile {args.tileid} night {args.night}')
        sys.exit(1)

    expids = list(exptable['EXPID'])
    prestdstar_expids = []
    stdstar_expids = []
    poststdstar_expids = []
    prestd_camwords = dict()
    badamps = dict()
    for i in range(len(expids)):
        expid=expids[i]
        camword=exptable['CAMWORD'][i]
        badcamword=exptable['BADCAMWORD'][i]
        badamps[expid] = exptable['BADAMPS'][i]
        if isinstance(badcamword, str):
            prestd_camwords[expids[i]] = difference_camwords(camword,badcamword,suppress_logging=True)
        else:
            prestd_camwords[expids[i]] = camword
        prestd_camwords[expids[i]] = camword_intersection([prestd_camwords[expids[i]],camera_superset])

        laststep = str(exptable['LASTSTEP'][i]).lower()
        if laststep in prestd_laststeps:
            prestdstar_expids.append(expid)
        if laststep in jntpst_laststeps:
            stdstar_expids.append(expid)
            poststdstar_expids.append(expid)

    joint_camwords = camword_union(list(prestd_camwords.values()), full_spectros_only=True)
    joint_camwords = camword_intersection([joint_camwords, camera_superset])

    poststd_camwords = dict()
    for expid, camword in prestd_camwords.items():
        poststd_camwords[expid] = camword_intersection([joint_camwords, camword, camera_superset])

    #-------------------------------------------------------------------------
    #- Create and submit a batch job if requested

    if args.batch:
        ncameras = len(decode_camword(joint_camwords))
        scriptfile = create_desi_proc_tilenight_batch_script(night=args.night,
                                                   exp=expids,
                                                   ncameras=ncameras,
                                                   tileid=args.tileid,
                                                   queue=args.queue,
                                                   system_name=args.system_name,
                                                   mpistdstars=args.mpistdstars,
                                                   use_specter=args.use_specter,
                                                   no_gpu=args.no_gpu,
                                                   cameras=camera_superset
                                                   )
        err = 0
        if not args.nosubmit:
            err = subprocess.call(['sbatch', scriptfile])
        sys.exit(err)

    #-------------------------------------------------------------------------
    #- Proceeding with running

    #- What are we going to do?
    if rank == 0:
        log.info('----------')
        log.info('Tile {} night {} prestdstar expids {} stdstar_expids {} poststdstar_expids {}'.format(
            args.tileid, args.night, prestdstar_expids, stdstar_expids, poststdstar_expids))
        log.info('Output root {}'.format(specprod_root()))
        log.info('----------')

    #- common arguments
    common_args = f'--night {args.night}'
    if args.no_gpu:
        common_args += ' --no-gpu'

    #- extract options
    extract_args=''
    if args.use_specter:
        extract_args += ' --use-specter'

    #- mpi options
    mpi_args=''
    if args.mpi:
        mpi_args += ' --mpistdstars'

    #- run desiproc prestdstar over exps
    for expid in prestdstar_expids:
        prestdstar_args = common_args + extract_args
        prestdstar_args += f' --nostdstarfit --nofluxcalib --expid {expid} --cameras {prestd_camwords[expid]}'
        if len(badamps[expid]) > 0:
            prestdstar_args += f' --badamps {badamps[expid]}'
        if rank==0:
            log.info(f'running desi_proc {prestdstar_args}')
        prestdstar_args_parsed = proc.parse(prestdstar_args.split())
        if not args.dryrun:
            try:
                errcode = proc.main(prestdstar_args_parsed,comm)
                if errcode != 0:
                    error_count += 1
            except (BaseException, Exception) as e:
                import traceback
                lines = traceback.format_exception(*sys.exc_info())
                log.error(f"prestdstar step using desi_proc with args {prestdstar_args} raised an exception:")
                print("".join(lines))
                log.warning(f'continuing to next prestdstar exposure, if any')
                error_count += 1

    #- collect post prestdstar error count
    if comm is not None:
        all_error_counts = comm.gather(error_count, root=0)
        error_count_pre = int(comm.bcast(np.sum(all_error_counts), root=0))
    else:
        error_count_pre = error_count

    continue_tilenight=True
    if error_count_pre > 0:
        if rank == 0:
            log.info(f'prestdstar failed with {error_count_pre} errors, skipping rest of tilenight steps')
        continue_tilenight=False

    #- Cleanup GPU memory and rank assignments before continuing
    desispec.gpu.redistribute_gpu_ranks(comm)

    if continue_tilenight and len(stdstar_expids) > 0:
        #- run joint stdstar fit using all exp for this tile night
        stdstar_args  = common_args + mpi_args
        stdstar_args += f' --obstype science --expids {",".join(map(str, stdstar_expids))} --cameras {joint_camwords}'
        if rank==0:
            log.info(f'running desi_proc_joint_fit {stdstar_args}')
        stdstar_args_parsed = proc_joint_fit.parse(stdstar_args.split())
        if not args.dryrun:
            try:
                errcode = proc_joint_fit.main(stdstar_args_parsed, comm)
                if errcode != 0:
                    error_count += 1
            except (BaseException, Exception) as e:
                import traceback
                lines = traceback.format_exception(*sys.exc_info())
                log.error(f"joint fit step using desi_proc_joint_fit with args {stdstar_args} raised an exception:")
                print("".join(lines))
                error_count += 1

        #- collect post joint fit error count
        if comm is not None:
            all_error_counts = comm.gather(error_count, root=0)
            error_count_jnt = int(comm.bcast(np.sum(all_error_counts), root=0))
        else:
            error_count_jnt = error_count

        if error_count_jnt > 0:
            if rank == 0:
                log.info(f'joint fitting failed with {error_count_jnt} errors, skipping rest of tilenight steps')
            continue_tilenight=False

    #- Cleanup GPU memory and rank assignments before continuing
    desispec.gpu.redistribute_gpu_ranks(comm)

    if continue_tilenight:
        #- run desiproc poststdstar over exps
        for expid in poststdstar_expids:
            poststdstar_args  = common_args
            poststdstar_args += f' --nostdstarfit --noprestdstarfit --expid {expid} --cameras {poststd_camwords[expid]}'
            if len(badamps[expid]) > 0:
                poststdstar_args += f' --badamps {badamps[expid]}'
            if rank==0:
                log.info(f'running desi_proc {poststdstar_args}')
            poststdstar_args_parsed = proc.parse(poststdstar_args.split())
            if not args.dryrun:
                try:
                    errcode = proc.main(poststdstar_args_parsed, comm)
                    if errcode != 0:
                        error_count += 1
                except (BaseException, Exception) as e:
                    import traceback
                    lines = traceback.format_exception(*sys.exc_info())
                    log.error(f"poststdstar step using proc.main with args {poststdstar_args} raised an exception:")
                    print("".join(lines))
                    error_count += 1

    #-------------------------------------------------------------------------
    #- Collect error count
    if comm is not None:
        all_error_counts = comm.gather(error_count, root=0)
        error_count = int(comm.bcast(np.sum(all_error_counts), root=0))

    if rank == 0 and error_count > 0:
        log.error(f'{error_count} processing errors in tilenight; see logs above')

    #-------------------------------------------------------------------------
    #- Done

    if rank == 0:
        duration_seconds = time.time() - start_time
        mm = int(duration_seconds) // 60
        ss = int(duration_seconds - mm*60)

        log.info(f'Tilenight main in {thisfile} returned at {time.asctime()}; duration {mm}m{ss}s')

    return error_count

1	"""
2	desispec.scripts.proc_tilenight
3	===============================
4
5	Script for science processing of a given DESI tile and night
6	"""
7	import sys	×
8	import subprocess	×
9	import time, datetime	×
10	from desispec.workflow.batch_writer import create_desi_proc_tilenight_batch_script	×
11	import numpy as np	×
12
13	from os.path import dirname, abspath	×
14	from desiutil.log import get_logger, DEBUG, INFO	×
15
16	from desispec.io import specprod_root	×
17	from desispec.workflow.exptable import get_exposure_table_pathname	×
18	from desispec.workflow.tableio import load_table	×
19	from desispec.io.util import decode_camword, create_camword, camword_union, camword_intersection, difference_camwords	×
20
21	import desispec.scripts.proc as proc	×
22	import desispec.scripts.proc_joint_fit as proc_joint_fit	×
23
24	from desispec.workflow.desi_proc_funcs import assign_mpi, get_desi_proc_tilenight_parser	×
25	from desispec.workflow.desi_proc_funcs import update_args_with_headers	×
26
27	import desispec.gpu	×
28
29	#########################################
30	######## Begin Body of the Code #########
31	#########################################
32
33	def parse(options=None):	×
34	parser = get_desi_proc_tilenight_parser()	×
35	args = parser.parse_args(options)	×
36
37	#- convert comma separated steps to list of str
38	if isinstance(args.laststeps, str):	×
39	args.laststeps = [laststep.strip().lower() for laststep in args.laststeps.split(',')]	×
40
41	return args	×
42
43	def main(args=None, comm=None):
44	if args is None:
45	args = parse()
46
47	log = get_logger()
48	start_time = time.time()
49	error_count = 0
50
51	if comm is not None:
52	#- Use the provided comm to determine rank and size
53	rank = comm.rank
54	size = comm.size
55	else:
56	#- Check MPI flags and determine the comm, rank, and size given the arguments
57	comm, rank, size = assign_mpi(do_mpi=args.mpi, do_batch=args.batch, log=log)
58
59	if rank == 0:
60	thisfile=dirname(abspath(__file__))
61	thistime=datetime.datetime.fromtimestamp(start_time).isoformat()
62	log.info(f'Tilenight started main in {thisfile} at {thistime}')
63
64	#- Read + broadcast exposure table for this night
65	exptable = None
66	if rank == 0:
67	exptable_file = get_exposure_table_pathname(args.night)
68	log.info(f'Reading exptable in {exptable_file}')
69	exptable = load_table(exptable_file)
70
71	if comm is not None:
72	exptable = comm.bcast(exptable, root=0)
73
74	#- Set the eligible laststep values for different processing steps
75	if args.laststeps is None:
76	prestd_laststeps = jntpst_laststeps = ['all', 'fluxcalib']
77	else:
78	prestd_laststeps = args.laststeps
79	jntpst_laststeps = list(set(['all', 'fluxcalib']) & set(args.laststeps))
80
81	#- Determine expids and cameras for a tile night
82	keep = exptable['OBSTYPE'] == 'science'
83	keep &= exptable['TILEID'] == int(args.tileid)
84	if args.laststeps is not None:
85	keep &= np.isin(exptable['LASTSTEP'].astype(str), args.laststeps)
86
87	#- Camera superset from args.cameras; if None default to all
88	camera_superset = 'a0123456789'
89	if args.cameras is not None:
90	camera_superset = args.cameras
91
92	exptable = exptable[keep]
93
94	if len(exptable) == 0:
95	if rank == 0:
96	log.critical(f'No good exposures found for tile {args.tileid} night {args.night}')
97	sys.exit(1)
98
99	expids = list(exptable['EXPID'])
100	prestdstar_expids = []
101	stdstar_expids = []
102	poststdstar_expids = []
103	prestd_camwords = dict()
104	badamps = dict()
105	for i in range(len(expids)):
106	expid=expids[i]
107	camword=exptable['CAMWORD'][i]
108	badcamword=exptable['BADCAMWORD'][i]
109	badamps[expid] = exptable['BADAMPS'][i]
110	if isinstance(badcamword, str):
111	prestd_camwords[expids[i]] = difference_camwords(camword,badcamword,suppress_logging=True)
112	else:
113	prestd_camwords[expids[i]] = camword
114	prestd_camwords[expids[i]] = camword_intersection([prestd_camwords[expids[i]],camera_superset])
115
116	laststep = str(exptable['LASTSTEP'][i]).lower()
117	if laststep in prestd_laststeps:
118	prestdstar_expids.append(expid)
119	if laststep in jntpst_laststeps:
120	stdstar_expids.append(expid)
121	poststdstar_expids.append(expid)
122
123	joint_camwords = camword_union(list(prestd_camwords.values()), full_spectros_only=True)
124	joint_camwords = camword_intersection([joint_camwords, camera_superset])
125
126	poststd_camwords = dict()
127	for expid, camword in prestd_camwords.items():
128	poststd_camwords[expid] = camword_intersection([joint_camwords, camword, camera_superset])
129
130	#-------------------------------------------------------------------------
131	#- Create and submit a batch job if requested
132
133	if args.batch:
134	ncameras = len(decode_camword(joint_camwords))
135	scriptfile = create_desi_proc_tilenight_batch_script(night=args.night,
136	exp=expids,
137	ncameras=ncameras,
138	tileid=args.tileid,
139	queue=args.queue,
140	system_name=args.system_name,
141	mpistdstars=args.mpistdstars,
142	use_specter=args.use_specter,
143	no_gpu=args.no_gpu,
144	cameras=camera_superset
145	)
146	err = 0
147	if not args.nosubmit:
148	err = subprocess.call(['sbatch', scriptfile])
149	sys.exit(err)
150
151	#-------------------------------------------------------------------------
152	#- Proceeding with running
153
154	#- What are we going to do?
155	if rank == 0:
156	log.info('----------')
157	log.info('Tile {} night {} prestdstar expids {} stdstar_expids {} poststdstar_expids {}'.format(
158	args.tileid, args.night, prestdstar_expids, stdstar_expids, poststdstar_expids))
159	log.info('Output root {}'.format(specprod_root()))
160	log.info('----------')
161
162	#- common arguments
163	common_args = f'--night {args.night}'
164	if args.no_gpu:
165	common_args += ' --no-gpu'
166
167	#- extract options
168	extract_args=''
169	if args.use_specter:
170	extract_args += ' --use-specter'
171
172	#- mpi options
173	mpi_args=''
174	if args.mpi:
175	mpi_args += ' --mpistdstars'
176
177	#- run desiproc prestdstar over exps
178	for expid in prestdstar_expids:
179	prestdstar_args = common_args + extract_args
180	prestdstar_args += f' --nostdstarfit --nofluxcalib --expid {expid} --cameras {prestd_camwords[expid]}'
181	if len(badamps[expid]) > 0:
182	prestdstar_args += f' --badamps {badamps[expid]}'
183	if rank==0:
184	log.info(f'running desi_proc {prestdstar_args}')
185	prestdstar_args_parsed = proc.parse(prestdstar_args.split())
186	if not args.dryrun:
187	try:
188	errcode = proc.main(prestdstar_args_parsed,comm)
189	if errcode != 0:
190	error_count += 1
191	except (BaseException, Exception) as e:
192	import traceback
193	lines = traceback.format_exception(*sys.exc_info())
194	log.error(f"prestdstar step using desi_proc with args {prestdstar_args} raised an exception:")
195	print("".join(lines))
196	log.warning(f'continuing to next prestdstar exposure, if any')
197	error_count += 1
198
199	#- collect post prestdstar error count
200	if comm is not None:
201	all_error_counts = comm.gather(error_count, root=0)
202	error_count_pre = int(comm.bcast(np.sum(all_error_counts), root=0))
203	else:
204	error_count_pre = error_count
205
206	continue_tilenight=True
207	if error_count_pre > 0:
208	if rank == 0:
209	log.info(f'prestdstar failed with {error_count_pre} errors, skipping rest of tilenight steps')
210	continue_tilenight=False
211
212	#- Cleanup GPU memory and rank assignments before continuing
213	desispec.gpu.redistribute_gpu_ranks(comm)
214
215	if continue_tilenight and len(stdstar_expids) > 0:
216	#- run joint stdstar fit using all exp for this tile night
217	stdstar_args = common_args + mpi_args
218	stdstar_args += f' --obstype science --expids {",".join(map(str, stdstar_expids))} --cameras {joint_camwords}'
219	if rank==0:
220	log.info(f'running desi_proc_joint_fit {stdstar_args}')
221	stdstar_args_parsed = proc_joint_fit.parse(stdstar_args.split())
222	if not args.dryrun:
223	try:
224	errcode = proc_joint_fit.main(stdstar_args_parsed, comm)
225	if errcode != 0:
226	error_count += 1
227	except (BaseException, Exception) as e:
228	import traceback
229	lines = traceback.format_exception(*sys.exc_info())
230	log.error(f"joint fit step using desi_proc_joint_fit with args {stdstar_args} raised an exception:")
231	print("".join(lines))
232	error_count += 1
233
234	#- collect post joint fit error count
235	if comm is not None:
236	all_error_counts = comm.gather(error_count, root=0)
237	error_count_jnt = int(comm.bcast(np.sum(all_error_counts), root=0))
238	else:
239	error_count_jnt = error_count
240
241	if error_count_jnt > 0:
242	if rank == 0:
243	log.info(f'joint fitting failed with {error_count_jnt} errors, skipping rest of tilenight steps')
244	continue_tilenight=False
245
246	#- Cleanup GPU memory and rank assignments before continuing
247	desispec.gpu.redistribute_gpu_ranks(comm)
248
249	if continue_tilenight:
250	#- run desiproc poststdstar over exps
251	for expid in poststdstar_expids:
252	poststdstar_args = common_args
253	poststdstar_args += f' --nostdstarfit --noprestdstarfit --expid {expid} --cameras {poststd_camwords[expid]}'
254	if len(badamps[expid]) > 0:
255	poststdstar_args += f' --badamps {badamps[expid]}'
256	if rank==0:
257	log.info(f'running desi_proc {poststdstar_args}')
258	poststdstar_args_parsed = proc.parse(poststdstar_args.split())
259	if not args.dryrun:
260	try:
261	errcode = proc.main(poststdstar_args_parsed, comm)
262	if errcode != 0:
263	error_count += 1
264	except (BaseException, Exception) as e:
265	import traceback
266	lines = traceback.format_exception(*sys.exc_info())
267	log.error(f"poststdstar step using proc.main with args {poststdstar_args} raised an exception:")
268	print("".join(lines))
269	error_count += 1
270
271	#-------------------------------------------------------------------------
272	#- Collect error count
273	if comm is not None:
274	all_error_counts = comm.gather(error_count, root=0)
275	error_count = int(comm.bcast(np.sum(all_error_counts), root=0))
276
277	if rank == 0 and error_count > 0:
278	log.error(f'{error_count} processing errors in tilenight; see logs above')
279
280	#-------------------------------------------------------------------------
281	#- Done
282
283	if rank == 0:
284	duration_seconds = time.time() - start_time
285	mm = int(duration_seconds) // 60
286	ss = int(duration_seconds - mm*60)
287
288	log.info(f'Tilenight main in {thisfile} returned at {time.asctime()}; duration {mm}m{ss}s')
289
290	return error_count

desihub / desispec / 19150063725

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous