Skip to content

admin.py

This module contains the admin classes that are registered with the Django Admin site.

Association

Bases: Model

model association between sources and measurements based on some parameters

Source code in vast_pipeline/models.py
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
class Association(models.Model):
    """
    model association between sources and measurements based on
    some parameters
    """
    source = models.ForeignKey(Source, on_delete=models.CASCADE)
    meas = models.ForeignKey(Measurement, on_delete=models.CASCADE)

    d2d = models.FloatField(
        default=0.,
        help_text='astronomical distance calculated by Astropy, arcsec.'
    )
    dr = models.FloatField(
        default=0.,
        help_text='De Ruiter radius calculated in advanced association.'
    )

    def __str__(self):
        return (
            f'distance: {self.d2d:.2f}' if self.dr == 0 else
            f'distance: {self.dr:.2f}'
        )

Band

Bases: Model

A band on the frequency spectrum used for imaging. Each image is associated with one band.

Source code in vast_pipeline/models.py
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
class Band(models.Model):
    """
    A band on the frequency spectrum used for imaging. Each image is
    associated with one band.
    """
    name = models.CharField(max_length=12, unique=True)
    frequency = models.FloatField(
        help_text='central frequency of band (integer MHz)'
    )
    bandwidth = models.FloatField(
        help_text='bandwidth (MHz)'
    )

    class Meta:
        ordering = ['frequency']

    def __str__(self):
        return self.name

Comment

Bases: Model

The model object for a comment.

Source code in vast_pipeline/models.py
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
class Comment(models.Model):
    """
    The model object for a comment.
    """
    author = models.ForeignKey(User, on_delete=models.CASCADE)
    datetime = models.DateTimeField(auto_now_add=True)
    comment = models.TextField()
    content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE)
    object_id = models.PositiveIntegerField()
    content_object = GenericForeignKey('content_type', 'object_id')

    def get_avatar_url(self) -> str:
        """Get the URL for the user's avatar from GitHub. If the user has
        no associated GitHub account (e.g. a Django superuser), return the URL
        to the default user avatar.

        Returns:
            The avatar URL.
        """
        social = UserSocialAuth.get_social_auth_for_user(self.author).first()
        if social and "avatar_url" in social.extra_data:
            return social.extra_data["avatar_url"]
        else:
            return static("img/user-32.png")

get_avatar_url()

Get the URL for the user's avatar from GitHub. If the user has no associated GitHub account (e.g. a Django superuser), return the URL to the default user avatar.

Returns:

Type Description
str

The avatar URL.

Source code in vast_pipeline/models.py
41
42
43
44
45
46
47
48
49
50
51
52
53
def get_avatar_url(self) -> str:
    """Get the URL for the user's avatar from GitHub. If the user has
    no associated GitHub account (e.g. a Django superuser), return the URL
    to the default user avatar.

    Returns:
        The avatar URL.
    """
    social = UserSocialAuth.get_social_auth_for_user(self.author).first()
    if social and "avatar_url" in social.extra_data:
        return social.extra_data["avatar_url"]
    else:
        return static("img/user-32.png")

CommentableModel

Bases: Model

A class to provide a commentable model.

Source code in vast_pipeline/models.py
56
57
58
59
60
61
62
63
64
65
66
67
68
class CommentableModel(models.Model):
    """
    A class to provide a commentable model.
    """
    comment = GenericRelation(
        Comment,
        content_type_field="content_type",
        object_id_field="object_id",
        related_query_name="%(class)s",
    )

    class Meta:
        abstract = True

Image

Bases: CommentableModel

An image is a 2D radio image from a cube

Source code in vast_pipeline/models.py
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
class Image(CommentableModel):
    """An image is a 2D radio image from a cube"""
    band = models.ForeignKey(Band, on_delete=models.CASCADE)
    run = models.ManyToManyField(Run)
    skyreg = models.ForeignKey(SkyRegion, on_delete=models.CASCADE)

    measurements_path = models.FilePathField(
        max_length=200,
        db_column='meas_path',
        help_text=(
            'the path to the measurements parquet that belongs to this image'
        )
    )
    POLARISATION_CHOICES = [
        ('I', 'I'),
        ('XX', 'XX'),
        ('YY', 'YY'),
        ('Q', 'Q'),
        ('U', 'U'),
        ('V', 'V'),
    ]
    polarisation = models.CharField(
        max_length=2,
        choices=POLARISATION_CHOICES,
        help_text='Polarisation of the image one of I,XX,YY,Q,U,V.'
    )
    name = models.CharField(
        unique=True,
        max_length=200,
        help_text='Name of the image.'
    )
    path = models.FilePathField(
        max_length=500,
        help_text='Path to the file containing the image.'
    )
    noise_path = models.FilePathField(
        max_length=300,
        blank=True,
        default='',
        help_text='Path to the file containing the RMS image.'
    )
    background_path = models.FilePathField(
        max_length=300,
        blank=True,
        default='',
        help_text='Path to the file containing the background image.'
    )

    datetime = models.DateTimeField(
        help_text='Date/time of observation or epoch.'
    )
    jd = models.FloatField(
        help_text='Julian date of the observation (days).'
    )
    duration = models.FloatField(
        default=0.,
        help_text='Duration of the observation.'
    )

    ra = models.FloatField(
        help_text='RA of the image centre (Deg).'
    )
    dec = models.FloatField(
        help_text='DEC of the image centre (Deg).'
    )
    fov_bmaj = models.FloatField(
        help_text='Field of view major axis (Deg).'
    )  # Major (Dec) radius of image (degrees)
    fov_bmin = models.FloatField(
        help_text='Field of view minor axis (Deg).'
    )  # Minor (RA) radius of image (degrees)
    physical_bmaj = models.FloatField(
        help_text='The actual size of the image major axis (Deg).'
    )  # Major (Dec) radius of image (degrees)
    physical_bmin = models.FloatField(
        help_text='The actual size of the image minor axis (Deg).'
    )  # Minor (RA) radius of image (degrees)
    radius_pixels = models.FloatField(
        help_text='Radius of the useable region of the image (pixels).'
    )

    beam_bmaj = models.FloatField(
        help_text='Major axis of image restoring beam (Deg).'
    )
    beam_bmin = models.FloatField(
        help_text='Minor axis of image restoring beam (Deg).'
    )
    beam_bpa = models.FloatField(
        help_text='Beam position angle (Deg).'
    )
    rms_median = models.FloatField(
        help_text='Background average RMS from the provided RMS map (mJy).'
    )
    rms_min = models.FloatField(
        help_text='Background minimum RMS from the provided RMS map (mJy).'
    )
    rms_max = models.FloatField(
        help_text='Background maximum RMS from the provided RMS map (mJy).'
    )

    class Meta:
        ordering = ['datetime']

    def __str__(self):
        return self.name

ImageAdmin

Bases: ModelAdmin

The ImageAdmin class.

Source code in vast_pipeline/admin.py
29
30
31
32
33
34
35
class ImageAdmin(admin.ModelAdmin):
    """
    The ImageAdmin class.
    """
    list_display = ('name', 'ra', 'dec', 'datetime')
    exclude = ('measurements_path', 'path', 'noise_path', 'background_path')
    search_fields = ('name',)

Measurement

Bases: CommentableModel

A Measurement is an object in the sky that has been detected at least once. Essentially a source single measurement in time.

Source code in vast_pipeline/models.py
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
class Measurement(CommentableModel):
    """
    A Measurement is an object in the sky that has been detected at least once.
    Essentially a source single measurement in time.
    """
    image = models.ForeignKey(
        Image,
        null=True,
        on_delete=models.CASCADE
    )  # first image seen in
    source = models.ManyToManyField(
        'Source',
        through='Association',
        through_fields=('meas', 'source')
    )

    name = models.CharField(max_length=64)

    ra = models.FloatField(help_text='RA of the source (Deg).')  # degrees
    ra_err = models.FloatField(
        help_text='RA error of the source (Deg).'
    )
    dec = models.FloatField(help_text='DEC of the source (Deg).')  # degrees
    dec_err = models.FloatField(
        help_text='DEC error of the source (Deg).'
    )

    bmaj = models.FloatField(
        help_text=(
            'The major axis of the Gaussian fit to the source (Deg).'
        )
    )
    err_bmaj = models.FloatField(help_text='Error major axis (Deg).')
    bmin = models.FloatField(
        help_text=(
            'The minor axis of the Gaussian fit to the source (Deg).'
        )
    )
    err_bmin = models.FloatField(help_text='Error minor axis (Deg).')
    pa = models.FloatField(
        help_text=(
            'Position angle of Gaussian fit east of north to bmaj '
            '(Deg).'
        )
    )
    err_pa = models.FloatField(help_text='Error position angle (Deg).')

    # supplied by user via config
    ew_sys_err = models.FloatField(
        help_text='Systematic error in east-west (RA) direction (Deg).'
    )
    # supplied by user via config
    ns_sys_err = models.FloatField(
        help_text='Systematic error in north-south (dec) direction (Deg).'
    )

    # estimate of maximum error radius (from ra_err and dec_err)
    # Used in advanced association.
    error_radius = models.FloatField(
        help_text=(
            'Estimate of maximum error radius using ra_err'
            ' and dec_err (Deg).'
        )
    )

    # quadratic sum of error_radius and ew_sys_err
    uncertainty_ew = models.FloatField(
        help_text=(
            'Total east-west (RA) uncertainty, quadratic sum of'
            ' error_radius and ew_sys_err (Deg).'
        )
    )
    # quadratic sum of error_radius and ns_sys_err
    uncertainty_ns = models.FloatField(
        help_text=(
            'Total north-south (Dec) uncertainty, quadratic sum of '
            'error_radius and ns_sys_err (Deg).'
        )
    )

    flux_int = models.FloatField()  # mJy/beam
    flux_int_err = models.FloatField()  # mJy/beam
    flux_int_isl_ratio = models.FloatField(
        help_text=(
            'Ratio of the component integrated flux to the total'
            ' island integrated flux.'
        )
    )
    flux_peak = models.FloatField()  # mJy/beam
    flux_peak_err = models.FloatField()  # mJy/beam
    flux_peak_isl_ratio = models.FloatField(
        help_text=(
            'Ratio of the component peak flux to the total'
            ' island peak flux.'
        )
    )
    chi_squared_fit = models.FloatField(
        db_column='chi2_fit',
        help_text='Chi-squared of the Guassian fit to the source.'
    )
    spectral_index = models.FloatField(
        db_column='spectr_idx',
        help_text='In-band Selavy spectral index.'
    )
    spectral_index_from_TT = models.BooleanField(
        default=False,
        db_column='spectr_idx_tt',
        help_text=(
            'True/False if the spectral index came from the taylor '
            'term.'
        )
    )

    local_rms = models.FloatField(
        help_text='Local rms in mJy from Selavy.'
    )  # mJy/beam

    snr = models.FloatField(
        help_text='Signal-to-noise ratio of the measurement.'
    )

    flag_c4 = models.BooleanField(
        default=False,
        help_text='Fit flag from Selavy.'
    )

    compactness = models.FloatField(
        help_text='Int flux over peak flux.'
    )

    has_siblings = models.BooleanField(
        default=False,
        help_text='True if the fit comes from an island that has more than 1 component.'
    )
    component_id = models.CharField(
        max_length=64,
        help_text=(
            'The ID of the component from which the source comes from.'
        )
    )
    island_id = models.CharField(
        max_length=64,
        help_text=(
            'The ID of the island from which the source comes from.'
        )
    )

    forced = models.BooleanField(
        default=False,
        help_text='True: the measurement is forced extracted.'
    )

    objects = MeasurementQuerySet.as_manager()

    class Meta:
        ordering = ['ra']

    def __str__(self):
        return self.name

MeasurementAdmin

Bases: ModelAdmin

The MeasurementAdmin class.

Source code in vast_pipeline/admin.py
63
64
65
66
67
68
69
class MeasurementAdmin(admin.ModelAdmin):
    """
    The MeasurementAdmin class.
    """
    list_display = ('name', 'ra', 'dec', 'forced')
    list_filter = ('forced',)
    search_fields = ('name',)

MeasurementQuerySet

Bases: QuerySet

Source code in vast_pipeline/models.py
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
class MeasurementQuerySet(models.QuerySet):

    def cone_search(
        self, ra: float, dec: float, radius_deg: float
    ) -> models.QuerySet:
        """
        Return all the Sources withing radius_deg of (ra,dec).
        Returns a QuerySet of Sources, ordered by distance from
        (ra,dec) ascending.

        Args:
            ra: The right ascension value of the cone search central
                coordinate.
            dec: The declination value of the cone search central coordinate.
            radius_deg: The radius over which to perform the cone search.

        Returns:
            Measurements found withing the cone search area.
        """
        return (
            self.extra(
                select={
                    "distance": "q3c_dist(ra, dec, %s, %s) * 3600"
                },
                select_params=[ra, dec],
                where=["q3c_radial_query(ra, dec, %s, %s, %s)"],
                params=[ra, dec, radius_deg],
            )
            .order_by("distance")
        )

Return all the Sources withing radius_deg of (ra,dec). Returns a QuerySet of Sources, ordered by distance from (ra,dec) ascending.

Parameters:

Name Type Description Default
ra float

The right ascension value of the cone search central coordinate.

required
dec float

The declination value of the cone search central coordinate.

required
radius_deg float

The radius over which to perform the cone search.

required

Returns:

Type Description
QuerySet

Measurements found withing the cone search area.

Source code in vast_pipeline/models.py
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
def cone_search(
    self, ra: float, dec: float, radius_deg: float
) -> models.QuerySet:
    """
    Return all the Sources withing radius_deg of (ra,dec).
    Returns a QuerySet of Sources, ordered by distance from
    (ra,dec) ascending.

    Args:
        ra: The right ascension value of the cone search central
            coordinate.
        dec: The declination value of the cone search central coordinate.
        radius_deg: The radius over which to perform the cone search.

    Returns:
        Measurements found withing the cone search area.
    """
    return (
        self.extra(
            select={
                "distance": "q3c_dist(ra, dec, %s, %s) * 3600"
            },
            select_params=[ra, dec],
            where=["q3c_radial_query(ra, dec, %s, %s, %s)"],
            params=[ra, dec, radius_deg],
        )
        .order_by("distance")
    )

PipelineConfig

Pipeline run configuration.

Attributes:

Name Type Description
SCHEMA

class attribute containing the YAML schema for the run config.

TEMPLATE_PATH str

class attribute containing the path to the default Jinja2 run config template file.

epoch_based bool

boolean indicating if the original run config inputs were provided with user-defined epochs.

Raises:

Type Description
PipelineConfigError

the input YAML config violates the schema.

Source code in vast_pipeline/pipeline/config.py
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
class PipelineConfig:
    """Pipeline run configuration.

    Attributes:
        SCHEMA: class attribute containing the YAML schema for the run config.
        TEMPLATE_PATH: class attribute containing the path to the default Jinja2 run
            config template file.
        epoch_based: boolean indicating if the original run config inputs were provided
            with user-defined epochs.

    Raises:
        PipelineConfigError: the input YAML config violates the schema.
    """

    # key: config input type, value: boolean indicating if it is required
    _REQUIRED_INPUT_TYPES: Dict[str, bool] = {
        "image": True,
        "selavy": True,
        "noise": True,
        "background": False,
    }
    # Inputs may be optional. All inputs will be either a unique list or a mapping (epoch
    # mode and/or glob expressions). These possibilities cannot be validated at once, so
    # it will accept Any and then revalidate later.
    _SCHEMA_INPUTS = {
        (k if v else yaml.Optional(k)): yaml.MapPattern(yaml.Str(), yaml.Any())
        | yaml.UniqueSeq(yaml.Str())
        for k, v in _REQUIRED_INPUT_TYPES.items()
    }
    _SCHEMA_GLOB_INPUTS = {"glob": yaml.Str() | yaml.Seq(yaml.Str())}
    _VALID_ASSOC_METHODS: List[str] = ["basic", "advanced", "deruiter"]
    SCHEMA = yaml.Map(
        {
            "run": yaml.Map(
                {
                    "path": yaml.Str(),
                    "suppress_astropy_warnings": yaml.Bool(),
                }
            ),
            "inputs": yaml.Map(_SCHEMA_INPUTS),
            "source_monitoring": yaml.Map(
                {
                    "monitor": yaml.Bool(),
                    "min_sigma": yaml.Float(),
                    "edge_buffer_scale": yaml.Float(),
                    "cluster_threshold": yaml.Float(),
                    "allow_nan": yaml.Bool(),
                }
            ),
            "source_association": yaml.Map(
                {
                    "method": yaml.Enum(_VALID_ASSOC_METHODS),
                    "radius": yaml.Float(),
                    "deruiter_radius": yaml.Float(),
                    "deruiter_beamwidth_limit": yaml.Float(),
                    "parallel": yaml.Bool(),
                    "epoch_duplicate_radius": yaml.Float(),
                }
            ),
            "new_sources": yaml.Map(
                {
                    "min_sigma": yaml.Float(),
                }
            ),
            "measurements": yaml.Map(
                {
                    "source_finder": yaml.Enum(["selavy"]),
                    "flux_fractional_error": yaml.Float(),
                    "condon_errors": yaml.Bool(),
                    "selavy_local_rms_fill_value": yaml.Float(),
                    "write_arrow_files": yaml.Bool(),
                    "ra_uncertainty": yaml.Float(),
                    "dec_uncertainty": yaml.Float(),
                }
            ),
            "variability": yaml.Map(
                {
                    "pair_metrics": yaml.Bool(),
                    "source_aggregate_pair_metrics_min_abs_vs": yaml.Float(),
                }
            ),
            yaml.Optional("processing"): yaml.Map(
                {
                    yaml.Optional(
                        "num_workers",
                        default=settings.PIPE_RUN_CONFIG_DEFAULTS['num_workers']):
                        yaml.NullNone() | yaml.Int() | yaml.Str(),
                    yaml.Optional(
                        "num_workers_io",
                        default=settings.PIPE_RUN_CONFIG_DEFAULTS['num_workers_io']):
                        yaml.NullNone() | yaml.Int() | yaml.Str(),
                    yaml.Optional(
                        "max_partition_mb",
                        default=settings.PIPE_RUN_CONFIG_DEFAULTS['max_partition_mb']):
                        yaml.Int()
                }
            )
        }
    )
    # path to default run config template
    TEMPLATE_PATH: str = os.path.join(
        settings.BASE_DIR, "vast_pipeline", "config_template.yaml.j2"
    )

    def __init__(self, config_yaml: yaml.YAML, validate_inputs: bool = True):
        """Initialises PipelineConfig with parsed (but not necessarily validated) YAML.

        Args:
            config_yaml (yaml.YAML): Input YAML, usually the output of `strictyaml.load`.
            validate_inputs (bool, optional): Validate the config input files. Ensures
                that the inputs match (e.g. each image has a catalogue), and that each
                path exists. Set to False to skip these checks. Defaults to True.

        Raises:
            PipelineConfigError: The input YAML config violates the schema.
        """
        self._yaml: yaml.YAML = config_yaml
        # The epoch_based parameter below is for if the user has entered just lists we
        # don't have access to the dates until the Image instances are created. So we
        # flag this as true so that we can reorder the epochs once the date information
        # is available. It is also recorded in the database such that there is a record
        # of the fact that the run was processed in an epoch based mode.
        self.epoch_based: bool

        # Determine if epoch-based association should be used based on input files.
        # If inputs have been parsed to dicts, then the user has defined their own epochs.
        # If inputs have been parsed to lists, we must convert to dicts and auto-fill
        # the epochs.

        # ensure the inputs are valid in case .from_file(..., validate=False) was used
        if not validate_inputs:
            return

        try:
            self._validate_inputs()
        except yaml.YAMLValidationError as e:
            raise PipelineConfigError(e)

        # detect simple list inputs and convert them to epoch-mode inputs
        yaml_inputs = self._yaml["inputs"]
        inputs = yaml_inputs.data
        for input_file_type in self._REQUIRED_INPUT_TYPES:
            # skip missing optional input types, e.g. background
            if (
                not self._REQUIRED_INPUT_TYPES[input_file_type]
                and input_file_type not in self["inputs"]
            ):
                continue

            input_files = inputs[input_file_type]

            # resolve glob expressions if present
            if isinstance(input_files, dict):
                # must be either a glob expression, list of glob expressions, or epoch-mode
                if "glob" in input_files:
                    # resolve the glob expressions
                    self.epoch_based = False
                    file_list = self._resolve_glob_expressions(
                        yaml_inputs[input_file_type]
                    )
                    inputs[input_file_type] = self._create_input_epochs(file_list)
                else:
                    # epoch-mode with either a list of files or glob expressions
                    self.epoch_based = True
                    for epoch in input_files:
                        if "glob" in input_files[epoch]:
                            # resolve the glob expressions
                            file_list = self._resolve_glob_expressions(
                                yaml_inputs[input_file_type][epoch]
                            )
                            inputs[input_file_type][epoch] = file_list
            else:
                # Epoch-based association not requested and no globs present. Replace
                # input lists with dicts where each input file has it's own epoch.
                self.epoch_based = False
                inputs[input_file_type] = self._create_input_epochs(
                    input_files
                )
        self._yaml["inputs"] = inputs

    def __getitem__(self, name: str):
        """Retrieves the requested YAML chunk as a native Python object."""
        return self._yaml[name].data

    @staticmethod
    def _create_input_epochs(input_files: List[str]) -> Dict[str, List[str]]:
        """Convert a list of input files into a dict where each list element is placed
        into its own list of length 1 and mapped to by a unique key, a string that is a
        0-padded integer. For example, ["A", "B", "C", ..., "Z"] would be converted to
        {
            "01": ["A"],
            "02": ["B"],
            "03": ["C"],
            ...
            "26": ["Z"],
        }
        The keys are 0-padded to ensure the strings are sortable regardless of the
        length of `input_files`.
        This conversion is required for run configs that are not defined in "epoch mode"
        as after config validation, the pipeline assumes that there will be defined
        epochs.

        Args:
            input_files: the list of input file paths.

        Returns:
            The input file paths mapped to by unique epoch keys.
        """
        pad_width = len(str(len(input_files)))
        input_files_dict = {
            f"{i + 1:0{pad_width}}": [val] for i, val in enumerate(input_files)
        }
        return input_files_dict

    @classmethod
    def from_file(
        cls,
        yaml_path: str,
        label: str = "run config",
        validate: bool = True,
        validate_inputs: bool = True,
        add_defaults: bool = True,
    ) -> "PipelineConfig":
        """Create a PipelineConfig object from a run configuration YAML file.

        Args:
            yaml_path: Path to the run config YAML file.
            label: A label for the config object that will be used in error messages.
                Default is "run config".
            validate: Perform config schema validation immediately after loading
                the config file. If set to False, the full schema validation
                will not be performed until PipelineConfig.validate() is
                explicitly called. The inputs are always validated regardless.
                Defaults to True.
            validate_inputs: Validate the config input files. Ensures that the inputs
                match (e.g. each image has a catalogue), and that each path exists. Set
                to False to skip these checks. Defaults to True.
            add_defaults: Add missing configuration parameters using configured
                defaults. The defaults are read from the Django settings file.
                Defaults to True.

        Raises:
            PipelineConfigError: The run config YAML file fails schema validation.

        """
        schema = cls.SCHEMA if validate else yaml.Any()
        with open(yaml_path) as fh:
            config_str = fh.read()
        try:
            config_yaml = yaml.load(config_str, schema=schema, label=label)
        except yaml.YAMLValidationError as e:
            raise PipelineConfigError(e)

        if add_defaults:
            # make a template config based on defaults
            config_defaults_str = make_config_template(
                cls.TEMPLATE_PATH,
                **settings.PIPE_RUN_CONFIG_DEFAULTS,
            )
            config_defaults_dict: Dict[str, Any] = yaml.load(config_defaults_str).data

            # merge configs
            config_dict = dict_merge(config_defaults_dict, config_yaml.data)
            config_yaml = yaml.as_document(config_dict, schema=schema, label=label)
        return cls(config_yaml, validate_inputs=validate_inputs)

    @staticmethod
    def _resolve_glob_expressions(input_files: yaml.YAML) -> List[str]:
        """Resolve glob expressions in a YAML chunk, returning a list of sorted file
        paths.

        Args:
            input_files (yaml.YAML): A validated YAML chunk of input files that is a
                mapping of "glob" to either a single glob expression or a sequence of
                glob expressions. e.g.
                ---
                glob: /foo/*.fits
                ---
                or
                ---
                glob:
                - /foo/A/*.fits
                - /foo/B/*.fits
                ---

        Returns:
            The resolved file paths in lexicographical order.
        """
        file_list: List[str] = []
        if input_files["glob"].is_sequence():
            for glob_expr in input_files["glob"]:
                file_list.extend(sorted(list(glob(glob_expr.data))))
        else:
            file_list.extend(sorted(list(glob(input_files["glob"].data))))
        return file_list

    def _validate_inputs(self):
        """Validate the input files. Each input type (i.e. image, selavy, noise,
        background) may be given as one of the following:
            1. A list of files.
            2. A glob expression.
            3. A list of glob expressions.
            4. A mapping of epochs to any of the above.
        Each input type is validated individually. Extra input validation steps, e.g. to
        ensure each input type has the same number of files, are performed in
        `validate()`.

        Raises:
            PipelineConfigError: The run config inputs fail schema validation.
        """
        try:
            # first pass validation
            self._yaml["inputs"].revalidate(yaml.Map(self._SCHEMA_INPUTS))

            for input_type in self._yaml["inputs"]:
                input_yaml = self._yaml["inputs"][input_type]
                if input_yaml.is_mapping():
                    # inputs are either epoch-mode, glob expressions, or both
                    if "glob" in input_yaml:
                        # validate globs
                        input_yaml.revalidate(yaml.Map(self._SCHEMA_GLOB_INPUTS))
                    else:
                        # validate epoch mode which may also contain glob expressions
                        input_yaml.revalidate(
                            yaml.MapPattern(
                                yaml.Str(),
                                yaml.UniqueSeq(yaml.Str())
                                | yaml.Map(self._SCHEMA_GLOB_INPUTS),
                            )
                        )
        except yaml.YAMLValidationError as e:
            raise PipelineConfigError(e)

    def validate(self, user: User = None):
        """Perform extra validation steps not covered by the default schema validation.
        The following checks are performed in order. If a check fails, an exception is
        raised and no further checks are performed.

        1. All input files have the same number of epochs and the same number of files
            per epoch.
        2. The number of input files does not exceed the configured pipeline maximum.
            This is only enforced if a regular user (not staff/admin) created the run.
        3. There are at least two input images.
        4. Background input images are required if source monitoring is turned on.
        5. All input files exist.

        Args:
            user: Optional. The User of the request if made through the UI. Defaults to
                None.

        Raises:
            PipelineConfigError: a validation check failed.
        """
        # run standard base schema validation
        try:
            self._yaml.revalidate(self.SCHEMA)
        except yaml.YAMLValidationError as e:
            raise PipelineConfigError(e)

        inputs = self["inputs"]

        # epochs defined for images only, used as the reference list of epochs
        epochs_image = inputs["image"].keys()
        # map input type to a set of epochs
        epochs_by_input_type = {
            input_type: set(inputs[input_type].keys())
            for input_type in inputs.keys()
        }
        # map input type to total number of files from all epochs
        n_files_by_input_type: Dict[str, int] = {}
        epoch_n_files: Dict[str, Dict[str, int]] = {}
        n_files = 0
        for input_type, epochs_set in epochs_by_input_type.items():
            epoch_n_files[input_type] = {}
            n_files_by_input_type[input_type] = 0
            for epoch in epochs_set:
                n = len(inputs[input_type][epoch])
                n_files_by_input_type[input_type] += n
                epoch_n_files[input_type][epoch] = n
                n_files += n

        # Note by this point the input files have been converted to a mapping regardless
        # of the user's input format.
        # Ensure all input file types have the same epochs.
        try:
            schema = yaml.Map({epoch: yaml.Seq(yaml.Str()) for epoch in epochs_image})
            for input_type in inputs.keys():
                # Generate a new YAML object on-the-fly per input to avoid saving
                # a validation schema per file in the PipelineConfig object
                # (These can consume a lot of RAM for long lists of input files).
                yaml.load(self._yaml["inputs"][input_type].as_yaml(), schema=schema)
        except yaml.YAMLValidationError:
            # number of epochs could be different or the name of the epochs may not match
            # find out which by counting the number of unique epochs per input type
            n_epochs_per_input_type = [
                len(epochs_set) for epochs_set in epochs_by_input_type.values()
            ]
            if len(set(n_epochs_per_input_type)) > 1:
                if self.epoch_based:
                    error_msg = "The number of epochs must match for all input types.\n"
                else:
                    error_msg = "The number of files must match for all input types.\n"
            else:
                error_msg = "The name of the epochs must match for all input types.\n"
            counts_str = ""
            if self.epoch_based:
                for input_type in epoch_n_files.keys():
                    n = len(epoch_n_files[input_type])
                    counts_str += (
                        f"{input_type} has {n} epoch{'s' if n > 1 else ''}:"
                        f" {', '.join(epoch_n_files[input_type].keys())}\n"
                    )
            else:
                for input_type, n in n_files_by_input_type.items():
                    counts_str += f"{input_type} has {n} file{'s' if n > 1 else ''}\n"

            counts_str = counts_str[:-1]
            raise PipelineConfigError(error_msg + counts_str)

        # Ensure all input file type epochs have the same number of files per epoch.
        # This could be combined with the number of epochs validation above, but we want
        # to give specific feedback to the user on failure.
        try:
            schema = yaml.Map(
                {epoch: yaml.FixedSeq([yaml.Str()] * epoch_n_files["image"][epoch])
                for epoch in epochs_image})
            for input_type in inputs.keys():
                yaml.load(self._yaml["inputs"][input_type].as_yaml(), schema=schema)
        except yaml.YAMLValidationError:
            # map input type to a mapping of epoch to file count
            file_counts_str = ""
            for input_type in inputs.keys():
                file_counts_str += f"{input_type}:\n"
                for epoch in sorted(inputs[input_type].keys()):
                    file_counts_str += (
                        f"  {epoch}: {len(inputs[input_type][epoch])}\n"
                    )
            file_counts_str = file_counts_str[:-1]
            raise PipelineConfigError(
                "The number of files per epoch does not match between input types.\n"
                + file_counts_str
            )

        # ensure the number of input files is less than the user limit
        if user and n_files > settings.MAX_PIPERUN_IMAGES:
            if user.is_staff:
                logger.warning(
                    "Maximum number of images"
                    f" ({settings.MAX_PIPERUN_IMAGES}) rule bypassed with"
                    " admin status."
                )
            else:
                raise PipelineConfigError(
                    f"The number of images entered ({n_files})"
                    " exceeds the maximum number of images currently"
                    f" allowed ({settings.MAX_PIPERUN_IMAGES}). Please ask"
                    " an administrator for advice on processing your run."
                )

        # ensure at least two inputs are provided
        check = [n_files_by_input_type[input_type] < 2 for input_type in inputs.keys()]
        if any(check):
            raise PipelineConfigError(
                "Number of image files must to be larger than 1"
            )

        # ensure background files are provided if source monitoring is requested
        try:
            monitor = self["source_monitoring"]["monitor"]
        except KeyError:
            monitor = False

        if monitor:
            inputs_schema = yaml.Map(
                {
                    k: yaml.UniqueSeq(yaml.Str())
                    | yaml.MapPattern(yaml.Str(), yaml.UniqueSeq(yaml.Str()))
                    for k in self._REQUIRED_INPUT_TYPES
                }
            )
            try:
                self._yaml["inputs"].revalidate(inputs_schema)
            except yaml.YAMLValidationError:
                raise PipelineConfigError(
                    "Background files must be provided if source monitoring is enabled."
                )

        # ensure the input files all exist
        for input_type in inputs.keys():
            for epoch, file_list in inputs[input_type].items():
                for file in file_list:
                    if not os.path.exists(file):
                        raise PipelineConfigError(f"{file} does not exist.")

        # ensure num_workers and num_workers_io are
        # either None (from null in config yaml) or an integer
        for param_name in ('num_workers', 'num_workers_io'):
            param_value = self['processing'][param_name]
            if (param_value is not None) and (type(param_value) is not int):
                raise PipelineConfigError(f"{param_name} can only be an integer or 'null'")

    def check_prev_config_diff(self) -> bool:
        """
        Checks if the previous config file differs from the current config file. Used in
        add mode. Only returns true if the images are different and the other general
        settings are the same (the requirement for add mode). Otherwise False is returned.

        Returns:
            `True` if images are different but general settings are the same,
                otherwise `False` is returned.
        """
        prev_config = PipelineConfig.from_file(
            os.path.join(self["run"]["path"], "config_prev.yaml"),
            label="previous run config",
        )
        if self._yaml == prev_config._yaml:
            return True

        # are the input image files different?
        images_changed = self["inputs"]["image"] != prev_config["inputs"]["image"]

        # are all the non-input file configs the same?
        config_dict = self._yaml.data
        prev_config_dict = prev_config._yaml.data
        _ = config_dict.pop("inputs")
        _ = prev_config_dict.pop("inputs")
        settings_check = config_dict == prev_config_dict

        if images_changed and settings_check:
            return False
        return True

    def image_opts(self) -> Dict[str, Any]:
        """
        Get the config options required for image ingestion only.
        Namely:
            - selavy_local_rms_fill_value
            - condon_errors
            - ra_uncertainty
            - dec_uncertainty

        Returns:
            The relevant key value pairs
        """
        keys = [
            "selavy_local_rms_fill_value",
            "condon_errors",
            "ra_uncertainty",
            "dec_uncertainty"
        ]
        return {key: self["measurements"][key] for key in keys}

__getitem__(name)

Retrieves the requested YAML chunk as a native Python object.

Source code in vast_pipeline/pipeline/config.py
216
217
218
def __getitem__(self, name: str):
    """Retrieves the requested YAML chunk as a native Python object."""
    return self._yaml[name].data

__init__(config_yaml, validate_inputs=True)

Initialises PipelineConfig with parsed (but not necessarily validated) YAML.

Parameters:

Name Type Description Default
config_yaml YAML

Input YAML, usually the output of strictyaml.load.

required
validate_inputs bool

Validate the config input files. Ensures that the inputs match (e.g. each image has a catalogue), and that each path exists. Set to False to skip these checks. Defaults to True.

True

Raises:

Type Description
PipelineConfigError

The input YAML config violates the schema.

Source code in vast_pipeline/pipeline/config.py
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
def __init__(self, config_yaml: yaml.YAML, validate_inputs: bool = True):
    """Initialises PipelineConfig with parsed (but not necessarily validated) YAML.

    Args:
        config_yaml (yaml.YAML): Input YAML, usually the output of `strictyaml.load`.
        validate_inputs (bool, optional): Validate the config input files. Ensures
            that the inputs match (e.g. each image has a catalogue), and that each
            path exists. Set to False to skip these checks. Defaults to True.

    Raises:
        PipelineConfigError: The input YAML config violates the schema.
    """
    self._yaml: yaml.YAML = config_yaml
    # The epoch_based parameter below is for if the user has entered just lists we
    # don't have access to the dates until the Image instances are created. So we
    # flag this as true so that we can reorder the epochs once the date information
    # is available. It is also recorded in the database such that there is a record
    # of the fact that the run was processed in an epoch based mode.
    self.epoch_based: bool

    # Determine if epoch-based association should be used based on input files.
    # If inputs have been parsed to dicts, then the user has defined their own epochs.
    # If inputs have been parsed to lists, we must convert to dicts and auto-fill
    # the epochs.

    # ensure the inputs are valid in case .from_file(..., validate=False) was used
    if not validate_inputs:
        return

    try:
        self._validate_inputs()
    except yaml.YAMLValidationError as e:
        raise PipelineConfigError(e)

    # detect simple list inputs and convert them to epoch-mode inputs
    yaml_inputs = self._yaml["inputs"]
    inputs = yaml_inputs.data
    for input_file_type in self._REQUIRED_INPUT_TYPES:
        # skip missing optional input types, e.g. background
        if (
            not self._REQUIRED_INPUT_TYPES[input_file_type]
            and input_file_type not in self["inputs"]
        ):
            continue

        input_files = inputs[input_file_type]

        # resolve glob expressions if present
        if isinstance(input_files, dict):
            # must be either a glob expression, list of glob expressions, or epoch-mode
            if "glob" in input_files:
                # resolve the glob expressions
                self.epoch_based = False
                file_list = self._resolve_glob_expressions(
                    yaml_inputs[input_file_type]
                )
                inputs[input_file_type] = self._create_input_epochs(file_list)
            else:
                # epoch-mode with either a list of files or glob expressions
                self.epoch_based = True
                for epoch in input_files:
                    if "glob" in input_files[epoch]:
                        # resolve the glob expressions
                        file_list = self._resolve_glob_expressions(
                            yaml_inputs[input_file_type][epoch]
                        )
                        inputs[input_file_type][epoch] = file_list
        else:
            # Epoch-based association not requested and no globs present. Replace
            # input lists with dicts where each input file has it's own epoch.
            self.epoch_based = False
            inputs[input_file_type] = self._create_input_epochs(
                input_files
            )
    self._yaml["inputs"] = inputs

check_prev_config_diff()

Checks if the previous config file differs from the current config file. Used in add mode. Only returns true if the images are different and the other general settings are the same (the requirement for add mode). Otherwise False is returned.

Returns:

Type Description
bool

True if images are different but general settings are the same, otherwise False is returned.

Source code in vast_pipeline/pipeline/config.py
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
def check_prev_config_diff(self) -> bool:
    """
    Checks if the previous config file differs from the current config file. Used in
    add mode. Only returns true if the images are different and the other general
    settings are the same (the requirement for add mode). Otherwise False is returned.

    Returns:
        `True` if images are different but general settings are the same,
            otherwise `False` is returned.
    """
    prev_config = PipelineConfig.from_file(
        os.path.join(self["run"]["path"], "config_prev.yaml"),
        label="previous run config",
    )
    if self._yaml == prev_config._yaml:
        return True

    # are the input image files different?
    images_changed = self["inputs"]["image"] != prev_config["inputs"]["image"]

    # are all the non-input file configs the same?
    config_dict = self._yaml.data
    prev_config_dict = prev_config._yaml.data
    _ = config_dict.pop("inputs")
    _ = prev_config_dict.pop("inputs")
    settings_check = config_dict == prev_config_dict

    if images_changed and settings_check:
        return False
    return True

from_file(yaml_path, label='run config', validate=True, validate_inputs=True, add_defaults=True) classmethod

Create a PipelineConfig object from a run configuration YAML file.

Parameters:

Name Type Description Default
yaml_path str

Path to the run config YAML file.

required
label str

A label for the config object that will be used in error messages. Default is "run config".

'run config'
validate bool

Perform config schema validation immediately after loading the config file. If set to False, the full schema validation will not be performed until PipelineConfig.validate() is explicitly called. The inputs are always validated regardless. Defaults to True.

True
validate_inputs bool

Validate the config input files. Ensures that the inputs match (e.g. each image has a catalogue), and that each path exists. Set to False to skip these checks. Defaults to True.

True
add_defaults bool

Add missing configuration parameters using configured defaults. The defaults are read from the Django settings file. Defaults to True.

True

Raises:

Type Description
PipelineConfigError

The run config YAML file fails schema validation.

Source code in vast_pipeline/pipeline/config.py
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
@classmethod
def from_file(
    cls,
    yaml_path: str,
    label: str = "run config",
    validate: bool = True,
    validate_inputs: bool = True,
    add_defaults: bool = True,
) -> "PipelineConfig":
    """Create a PipelineConfig object from a run configuration YAML file.

    Args:
        yaml_path: Path to the run config YAML file.
        label: A label for the config object that will be used in error messages.
            Default is "run config".
        validate: Perform config schema validation immediately after loading
            the config file. If set to False, the full schema validation
            will not be performed until PipelineConfig.validate() is
            explicitly called. The inputs are always validated regardless.
            Defaults to True.
        validate_inputs: Validate the config input files. Ensures that the inputs
            match (e.g. each image has a catalogue), and that each path exists. Set
            to False to skip these checks. Defaults to True.
        add_defaults: Add missing configuration parameters using configured
            defaults. The defaults are read from the Django settings file.
            Defaults to True.

    Raises:
        PipelineConfigError: The run config YAML file fails schema validation.

    """
    schema = cls.SCHEMA if validate else yaml.Any()
    with open(yaml_path) as fh:
        config_str = fh.read()
    try:
        config_yaml = yaml.load(config_str, schema=schema, label=label)
    except yaml.YAMLValidationError as e:
        raise PipelineConfigError(e)

    if add_defaults:
        # make a template config based on defaults
        config_defaults_str = make_config_template(
            cls.TEMPLATE_PATH,
            **settings.PIPE_RUN_CONFIG_DEFAULTS,
        )
        config_defaults_dict: Dict[str, Any] = yaml.load(config_defaults_str).data

        # merge configs
        config_dict = dict_merge(config_defaults_dict, config_yaml.data)
        config_yaml = yaml.as_document(config_dict, schema=schema, label=label)
    return cls(config_yaml, validate_inputs=validate_inputs)

image_opts()

Get the config options required for image ingestion only. Namely: - selavy_local_rms_fill_value - condon_errors - ra_uncertainty - dec_uncertainty

Returns:

Type Description
Dict[str, Any]

The relevant key value pairs

Source code in vast_pipeline/pipeline/config.py
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
def image_opts(self) -> Dict[str, Any]:
    """
    Get the config options required for image ingestion only.
    Namely:
        - selavy_local_rms_fill_value
        - condon_errors
        - ra_uncertainty
        - dec_uncertainty

    Returns:
        The relevant key value pairs
    """
    keys = [
        "selavy_local_rms_fill_value",
        "condon_errors",
        "ra_uncertainty",
        "dec_uncertainty"
    ]
    return {key: self["measurements"][key] for key in keys}

validate(user=None)

Perform extra validation steps not covered by the default schema validation. The following checks are performed in order. If a check fails, an exception is raised and no further checks are performed.

  1. All input files have the same number of epochs and the same number of files per epoch.
  2. The number of input files does not exceed the configured pipeline maximum. This is only enforced if a regular user (not staff/admin) created the run.
  3. There are at least two input images.
  4. Background input images are required if source monitoring is turned on.
  5. All input files exist.

Parameters:

Name Type Description Default
user User

Optional. The User of the request if made through the UI. Defaults to None.

None

Raises:

Type Description
PipelineConfigError

a validation check failed.

Source code in vast_pipeline/pipeline/config.py
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
def validate(self, user: User = None):
    """Perform extra validation steps not covered by the default schema validation.
    The following checks are performed in order. If a check fails, an exception is
    raised and no further checks are performed.

    1. All input files have the same number of epochs and the same number of files
        per epoch.
    2. The number of input files does not exceed the configured pipeline maximum.
        This is only enforced if a regular user (not staff/admin) created the run.
    3. There are at least two input images.
    4. Background input images are required if source monitoring is turned on.
    5. All input files exist.

    Args:
        user: Optional. The User of the request if made through the UI. Defaults to
            None.

    Raises:
        PipelineConfigError: a validation check failed.
    """
    # run standard base schema validation
    try:
        self._yaml.revalidate(self.SCHEMA)
    except yaml.YAMLValidationError as e:
        raise PipelineConfigError(e)

    inputs = self["inputs"]

    # epochs defined for images only, used as the reference list of epochs
    epochs_image = inputs["image"].keys()
    # map input type to a set of epochs
    epochs_by_input_type = {
        input_type: set(inputs[input_type].keys())
        for input_type in inputs.keys()
    }
    # map input type to total number of files from all epochs
    n_files_by_input_type: Dict[str, int] = {}
    epoch_n_files: Dict[str, Dict[str, int]] = {}
    n_files = 0
    for input_type, epochs_set in epochs_by_input_type.items():
        epoch_n_files[input_type] = {}
        n_files_by_input_type[input_type] = 0
        for epoch in epochs_set:
            n = len(inputs[input_type][epoch])
            n_files_by_input_type[input_type] += n
            epoch_n_files[input_type][epoch] = n
            n_files += n

    # Note by this point the input files have been converted to a mapping regardless
    # of the user's input format.
    # Ensure all input file types have the same epochs.
    try:
        schema = yaml.Map({epoch: yaml.Seq(yaml.Str()) for epoch in epochs_image})
        for input_type in inputs.keys():
            # Generate a new YAML object on-the-fly per input to avoid saving
            # a validation schema per file in the PipelineConfig object
            # (These can consume a lot of RAM for long lists of input files).
            yaml.load(self._yaml["inputs"][input_type].as_yaml(), schema=schema)
    except yaml.YAMLValidationError:
        # number of epochs could be different or the name of the epochs may not match
        # find out which by counting the number of unique epochs per input type
        n_epochs_per_input_type = [
            len(epochs_set) for epochs_set in epochs_by_input_type.values()
        ]
        if len(set(n_epochs_per_input_type)) > 1:
            if self.epoch_based:
                error_msg = "The number of epochs must match for all input types.\n"
            else:
                error_msg = "The number of files must match for all input types.\n"
        else:
            error_msg = "The name of the epochs must match for all input types.\n"
        counts_str = ""
        if self.epoch_based:
            for input_type in epoch_n_files.keys():
                n = len(epoch_n_files[input_type])
                counts_str += (
                    f"{input_type} has {n} epoch{'s' if n > 1 else ''}:"
                    f" {', '.join(epoch_n_files[input_type].keys())}\n"
                )
        else:
            for input_type, n in n_files_by_input_type.items():
                counts_str += f"{input_type} has {n} file{'s' if n > 1 else ''}\n"

        counts_str = counts_str[:-1]
        raise PipelineConfigError(error_msg + counts_str)

    # Ensure all input file type epochs have the same number of files per epoch.
    # This could be combined with the number of epochs validation above, but we want
    # to give specific feedback to the user on failure.
    try:
        schema = yaml.Map(
            {epoch: yaml.FixedSeq([yaml.Str()] * epoch_n_files["image"][epoch])
            for epoch in epochs_image})
        for input_type in inputs.keys():
            yaml.load(self._yaml["inputs"][input_type].as_yaml(), schema=schema)
    except yaml.YAMLValidationError:
        # map input type to a mapping of epoch to file count
        file_counts_str = ""
        for input_type in inputs.keys():
            file_counts_str += f"{input_type}:\n"
            for epoch in sorted(inputs[input_type].keys()):
                file_counts_str += (
                    f"  {epoch}: {len(inputs[input_type][epoch])}\n"
                )
        file_counts_str = file_counts_str[:-1]
        raise PipelineConfigError(
            "The number of files per epoch does not match between input types.\n"
            + file_counts_str
        )

    # ensure the number of input files is less than the user limit
    if user and n_files > settings.MAX_PIPERUN_IMAGES:
        if user.is_staff:
            logger.warning(
                "Maximum number of images"
                f" ({settings.MAX_PIPERUN_IMAGES}) rule bypassed with"
                " admin status."
            )
        else:
            raise PipelineConfigError(
                f"The number of images entered ({n_files})"
                " exceeds the maximum number of images currently"
                f" allowed ({settings.MAX_PIPERUN_IMAGES}). Please ask"
                " an administrator for advice on processing your run."
            )

    # ensure at least two inputs are provided
    check = [n_files_by_input_type[input_type] < 2 for input_type in inputs.keys()]
    if any(check):
        raise PipelineConfigError(
            "Number of image files must to be larger than 1"
        )

    # ensure background files are provided if source monitoring is requested
    try:
        monitor = self["source_monitoring"]["monitor"]
    except KeyError:
        monitor = False

    if monitor:
        inputs_schema = yaml.Map(
            {
                k: yaml.UniqueSeq(yaml.Str())
                | yaml.MapPattern(yaml.Str(), yaml.UniqueSeq(yaml.Str()))
                for k in self._REQUIRED_INPUT_TYPES
            }
        )
        try:
            self._yaml["inputs"].revalidate(inputs_schema)
        except yaml.YAMLValidationError:
            raise PipelineConfigError(
                "Background files must be provided if source monitoring is enabled."
            )

    # ensure the input files all exist
    for input_type in inputs.keys():
        for epoch, file_list in inputs[input_type].items():
            for file in file_list:
                if not os.path.exists(file):
                    raise PipelineConfigError(f"{file} does not exist.")

    # ensure num_workers and num_workers_io are
    # either None (from null in config yaml) or an integer
    for param_name in ('num_workers', 'num_workers_io'):
        param_value = self['processing'][param_name]
        if (param_value is not None) and (type(param_value) is not int):
            raise PipelineConfigError(f"{param_name} can only be an integer or 'null'")

RelatedSource

Bases: Model

Association table for the many to many Source relationship with itself Django doc https://docs.djangoproject.com/en/3.1/ref/models/fields/#django.db.models.ManyToManyField.through

Source code in vast_pipeline/models.py
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
class RelatedSource(models.Model):
    '''
    Association table for the many to many Source relationship with itself
    Django doc
    https://docs.djangoproject.com/en/3.1/ref/models/fields/#django.db.models.ManyToManyField.through
    '''
    from_source = models.ForeignKey(Source, on_delete=models.CASCADE)
    to_source = models.ForeignKey(
        Source,
        on_delete=models.CASCADE,
        related_name='related_sources'
    )

    class Meta:
        constraints = [
            models.UniqueConstraint(
                name='%(app_label)s_%(class)s_unique_pair',
                fields=['from_source', 'to_source']
            )
        ]

Run

Bases: CommentableModel

A Run is essentially a pipeline run/processing istance over a set of images

Source code in vast_pipeline/models.py
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
class Run(CommentableModel):
    """
    A Run is essentially a pipeline run/processing istance over a set of
    images
    """
    user = models.ForeignKey(
        User,
        on_delete=models.SET_NULL,
        null=True,
        blank=True
    )

    name = models.CharField(
        max_length=64,
        unique=True,
        validators=[
            RegexValidator(
                regex=r'[\[@!#$%^&*()<>?/\|}{~:\] ]',
                message='Name contains not allowed characters!',
                inverse_match=True
            ),
        ],
        help_text='name of the pipeline run'
    )
    description = models.CharField(
        max_length=240,
        blank=True,
        help_text="A short description of the pipeline run."
    )
    time = models.DateTimeField(
        auto_now=True,
        help_text='Datetime of a pipeline run.'
    )
    path = models.FilePathField(
        max_length=200,
        help_text='path to the pipeline run'
    )
    STATUS_CHOICES = [
        ('INI', 'Initialised'),
        ('QUE', 'Queued'),
        ('RUN', 'Running'),
        ('END', 'Completed'),
        ('ERR', 'Error'),
        ('RES', 'Restoring'),
        ('DEL', 'Deleting'),
    ]
    status = models.CharField(
        max_length=3,
        choices=STATUS_CHOICES,
        default='INI',
        help_text='Status of the pipeline run.'
    )
    n_images = models.IntegerField(
        default=0,
        help_text='number of images processed in this run'
    )
    n_sources = models.IntegerField(
        default=0,
        help_text='number of sources extracted in this run'
    )
    n_selavy_measurements = models.IntegerField(
        default=0,
        help_text='number of selavy measurements in this run'
    )
    n_forced_measurements = models.IntegerField(
        default=0,
        help_text='number of forced measurements in this run'
    )
    n_new_sources = models.IntegerField(
        default=0,
        help_text='number of new sources in this run'
    )
    epoch_based = models.BooleanField(
        default=False,
        help_text=(
            'Whether the run was processed using epoch based association'
            ', i.e. the user passed in groups of images defining epochs'
            ' rather than every image being treated individually.'
        )
    )

    objects = RunManager()  # used instead of RunQuerySet.as_manager() so mypy checks work

    class Meta:
        ordering = ['name']

    def __str__(self):
        return self.name

    def save(self, *args, **kwargs):
        # enforce the full model validation on save
        self.full_clean()
        super(Run, self).save(*args, **kwargs)

    def get_config(
        self, validate: bool = True, validate_inputs: bool = True, prev: bool = False
    ) -> PipelineConfig:
        """Read, parse, and optionally validate the run configuration file.

        Args:
            validate: Validate the run configuration. Defaults to False.
            validate_inputs: Validate the config input files. Ensures
                that the inputs match (e.g. each image has a catalogue), and that each
                path exists. Set to False to skip these checks. Defaults to True.
            prev: Get the previous config file instead of the current config. The
                previous config is the one used for the last successfully completed run.
                The current config may have been modified since the run was executed.

        Returns:
            PipelineConfig: The run configuration object.
        """
        config_name = "config_prev.yaml" if prev else "config.yaml"
        config = PipelineConfig.from_file(
            str(Path(self.path) / config_name),
            validate=validate,
            validate_inputs=validate_inputs,
        )
        return config

get_config(validate=True, validate_inputs=True, prev=False)

Read, parse, and optionally validate the run configuration file.

Parameters:

Name Type Description Default
validate bool

Validate the run configuration. Defaults to False.

True
validate_inputs bool

Validate the config input files. Ensures that the inputs match (e.g. each image has a catalogue), and that each path exists. Set to False to skip these checks. Defaults to True.

True
prev bool

Get the previous config file instead of the current config. The previous config is the one used for the last successfully completed run. The current config may have been modified since the run was executed.

False

Returns:

Name Type Description
PipelineConfig PipelineConfig

The run configuration object.

Source code in vast_pipeline/models.py
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
def get_config(
    self, validate: bool = True, validate_inputs: bool = True, prev: bool = False
) -> PipelineConfig:
    """Read, parse, and optionally validate the run configuration file.

    Args:
        validate: Validate the run configuration. Defaults to False.
        validate_inputs: Validate the config input files. Ensures
            that the inputs match (e.g. each image has a catalogue), and that each
            path exists. Set to False to skip these checks. Defaults to True.
        prev: Get the previous config file instead of the current config. The
            previous config is the one used for the last successfully completed run.
            The current config may have been modified since the run was executed.

    Returns:
        PipelineConfig: The run configuration object.
    """
    config_name = "config_prev.yaml" if prev else "config.yaml"
    config = PipelineConfig.from_file(
        str(Path(self.path) / config_name),
        validate=validate,
        validate_inputs=validate_inputs,
    )
    return config

RunAdmin

Bases: ModelAdmin

The RunAdmin class.

Source code in vast_pipeline/admin.py
18
19
20
21
22
23
24
class RunAdmin(admin.ModelAdmin):
    """
    The RunAdmin class.
    """
    list_display = ('name', 'time', 'status')
    list_filter = ('time', 'status')
    exclude = ('path',)

RunQuerySet

Bases: QuerySet

Source code in vast_pipeline/models.py
71
72
73
74
75
76
77
78
79
80
81
82
83
class RunQuerySet(models.QuerySet):

    def check_max_runs(self, max_runs: int = 5) -> int:
        """
        Check if number of running pipeline runs is above threshold.

        Args:
            max_runs: The maximum number of processing runs allowed.

        Returns:
            The count of the current pipeline runs with a status of `RUN`.
        """
        return self.filter(status='RUN').count() >= max_runs

check_max_runs(max_runs=5)

Check if number of running pipeline runs is above threshold.

Parameters:

Name Type Description Default
max_runs int

The maximum number of processing runs allowed.

5

Returns:

Type Description
int

The count of the current pipeline runs with a status of RUN.

Source code in vast_pipeline/models.py
73
74
75
76
77
78
79
80
81
82
83
def check_max_runs(self, max_runs: int = 5) -> int:
    """
    Check if number of running pipeline runs is above threshold.

    Args:
        max_runs: The maximum number of processing runs allowed.

    Returns:
        The count of the current pipeline runs with a status of `RUN`.
    """
    return self.filter(status='RUN').count() >= max_runs

SkyRegionAdmin

Bases: ModelAdmin

The SkyRegionAdmin class.

Source code in vast_pipeline/admin.py
43
44
45
46
47
class SkyRegionAdmin(admin.ModelAdmin):
    """
    The SkyRegionAdmin class.
    """
    list_display = ('__str__', 'centre_ra', 'centre_dec')

Source

Bases: CommentableModel

Source code in vast_pipeline/models.py
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
class Source(CommentableModel):
    run = models.ForeignKey(Run, on_delete=models.CASCADE, null=True,)
    related = models.ManyToManyField(
        'self',
        through='RelatedSource',
        symmetrical=False,
        through_fields=('from_source', 'to_source')
    )

    name = models.CharField(max_length=100)
    new = models.BooleanField(default=False, help_text='New Source.')
    tags = TagField(
        space_delimiter=False,
        autocomplete_view="vast_pipeline:source_tags_autocomplete",
        autocomplete_settings={"width": "100%"},
    )

    # average fields calculated from the source measurements
    wavg_ra = models.FloatField(
        help_text='The weighted average right ascension (Deg).'
    )
    wavg_dec = models.FloatField(
        help_text='The weighted average declination (Deg).'
    )
    wavg_uncertainty_ew = models.FloatField(
        help_text=(
            'The weighted average uncertainty in the east-'
            'west (RA) direction (Deg).'
        )
    )
    wavg_uncertainty_ns = models.FloatField(
        help_text=(
            'The weighted average uncertainty in the north-'
            'south (Dec) direction (Deg).'
        )
    )
    avg_flux_int = models.FloatField(
        help_text='The average integrated flux value.'
    )
    avg_flux_peak = models.FloatField(
        help_text='The average peak flux value.'
    )
    max_flux_peak = models.FloatField(
        help_text='The maximum peak flux value.'
    )
    min_flux_peak = models.FloatField(
        help_text='The minimum peak flux value.'
    )
    max_flux_int = models.FloatField(
        help_text='The maximum integrated flux value.'
    )
    min_flux_int = models.FloatField(
        help_text='The minimum integrated flux value.'
    )
    min_flux_int_isl_ratio = models.FloatField(
        help_text='The minimum integrated island flux ratio value.'
    )
    min_flux_peak_isl_ratio = models.FloatField(
        help_text='The minimum peak island flux ratio value.'
    )
    avg_compactness = models.FloatField(
        help_text='The average compactness.'
    )
    min_snr = models.FloatField(
        help_text='The minimum signal-to-noise ratio value of the detections.'
    )
    max_snr = models.FloatField(
        help_text='The maximum signal-to-noise ratio value of the detections.'
    )

    # metrics
    v_int = models.FloatField(
        help_text='V metric for int flux.'
    )
    v_peak = models.FloatField(
        help_text='V metric for peak flux.'
    )
    eta_int = models.FloatField(
        help_text='Eta metric for int flux.'
    )
    eta_peak = models.FloatField(
        help_text='Eta metric for peak flux.'
    )
    new_high_sigma = models.FloatField(
        help_text=(
            'The largest sigma value for the new source'
            ' if it was placed in previous image.'
        )
    )
    n_neighbour_dist = models.FloatField(
        help_text='Distance to the nearest neighbour (deg)'
    )
    vs_abs_significant_max_int = models.FloatField(
        default=0.0,
        help_text=(
            'Maximum value of all measurement pair variability t-statistics for int'
            ' flux that exceed variability.source_aggregate_pair_metrics_min_abs_vs in'
            ' the pipeline run configuration.'
        )
    )
    m_abs_significant_max_int = models.FloatField(
        default=0.0,
        help_text=(
            'Maximum absolute value of all measurement pair modulation indices for int'
            ' flux that exceed variability.source_aggregate_pair_metrics_min_abs_vs in'
            ' the pipeline run configuration.'
        )
    )
    vs_abs_significant_max_peak = models.FloatField(
        default=0.0,
        help_text=(
            'Maximum absolute value of all measurement pair variability t-statistics for'
            ' peak flux that exceed variability.source_aggregate_pair_metrics_min_abs_vs'
            ' in the pipeline run configuration.'
        )
    )
    m_abs_significant_max_peak = models.FloatField(
        default=0.0,
        help_text=(
            'Maximum absolute value of all measurement pair modulation indices for '
            ' peak flux that exceed variability.source_aggregate_pair_metrics_min_abs_vs'
            ' in the pipeline run configuration.'
        )
    )

    # total metrics to report in UI
    n_meas = models.IntegerField(
        help_text='total measurements of the source'
    )
    n_meas_sel = models.IntegerField(
        help_text='total selavy extracted measurements of the source'
    )
    n_meas_forced = models.IntegerField(
        help_text='total force extracted measurements of the source'
    )
    n_rel = models.IntegerField(
        help_text='total relations of the source with other sources'
    )
    n_sibl = models.IntegerField(
        help_text='total siblings of the source'
    )

    objects = SourceQuerySet.as_manager()

    def __str__(self):
        return self.name

    def get_measurement_pairs(self) -> List[MeasurementPair]:
        """Calculate the measurement pair metrics for the source. If the run config
        set variability.pair_metrics to false, then no pairs are calculated and an empty
        list is returned.

        Returns:
            List[MeasurementPair]: The list of measurement pairs and their metrics.
        """
        # do not calculate pair metrics if it was disabled in the run config
        config = self.run.get_config(validate=False, validate_inputs=False, prev=True)
        # validate the config schema only, not the full validation executed by
        # PipelineConfig.validate.
        config._yaml.revalidate(PipelineConfig.SCHEMA)
        if not config["variability"]["pair_metrics"]:
            return []

        measurements = (
            Measurement.objects.filter(source=self)
            .select_related("image")
            .order_by("image__datetime")
        )
        measurement_pairs: List[MeasurementPair] = []
        for meas_a, meas_b in combinations(measurements, 2):
            # ensure the measurements are in time order
            if meas_a.image.datetime > meas_b.image.datetime:
                meas_a, meas_b = meas_b, meas_a
            # calculate metrics
            vs_peak = calculate_vs_metric(
                meas_a.flux_peak, meas_b.flux_peak, meas_a.flux_peak_err, meas_b.flux_peak_err
            )
            m_int = calculate_m_metric(meas_a.flux_int, meas_b.flux_int)
            vs_int = calculate_vs_metric(
                meas_a.flux_int, meas_b.flux_int, meas_a.flux_int_err, meas_b.flux_int_err
            )
            m_peak = calculate_m_metric(meas_a.flux_peak, meas_b.flux_peak)
            measurement_pairs.append(
                MeasurementPair(self.id, meas_a.id, meas_b.id, vs_peak, m_peak, vs_int, m_int)
            )
        return measurement_pairs

get_measurement_pairs()

Calculate the measurement pair metrics for the source. If the run config set variability.pair_metrics to false, then no pairs are calculated and an empty list is returned.

Returns:

Type Description
List[MeasurementPair]

List[MeasurementPair]: The list of measurement pairs and their metrics.

Source code in vast_pipeline/models.py
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
def get_measurement_pairs(self) -> List[MeasurementPair]:
    """Calculate the measurement pair metrics for the source. If the run config
    set variability.pair_metrics to false, then no pairs are calculated and an empty
    list is returned.

    Returns:
        List[MeasurementPair]: The list of measurement pairs and their metrics.
    """
    # do not calculate pair metrics if it was disabled in the run config
    config = self.run.get_config(validate=False, validate_inputs=False, prev=True)
    # validate the config schema only, not the full validation executed by
    # PipelineConfig.validate.
    config._yaml.revalidate(PipelineConfig.SCHEMA)
    if not config["variability"]["pair_metrics"]:
        return []

    measurements = (
        Measurement.objects.filter(source=self)
        .select_related("image")
        .order_by("image__datetime")
    )
    measurement_pairs: List[MeasurementPair] = []
    for meas_a, meas_b in combinations(measurements, 2):
        # ensure the measurements are in time order
        if meas_a.image.datetime > meas_b.image.datetime:
            meas_a, meas_b = meas_b, meas_a
        # calculate metrics
        vs_peak = calculate_vs_metric(
            meas_a.flux_peak, meas_b.flux_peak, meas_a.flux_peak_err, meas_b.flux_peak_err
        )
        m_int = calculate_m_metric(meas_a.flux_int, meas_b.flux_int)
        vs_int = calculate_vs_metric(
            meas_a.flux_int, meas_b.flux_int, meas_a.flux_int_err, meas_b.flux_int_err
        )
        m_peak = calculate_m_metric(meas_a.flux_peak, meas_b.flux_peak)
        measurement_pairs.append(
            MeasurementPair(self.id, meas_a.id, meas_b.id, vs_peak, m_peak, vs_int, m_int)
        )
    return measurement_pairs

SourceAdmin

Bases: ModelAdmin

The SourceAdmin class.

Source code in vast_pipeline/admin.py
52
53
54
55
56
57
58
class SourceAdmin(admin.ModelAdmin):
    """
    The SourceAdmin class.
    """
    list_display = ('name', 'wavg_ra', 'wavg_dec', 'new')
    list_filter = ('new',)
    search_fields = ('name',)

SourceFavAdmin

Bases: ModelAdmin

The SourceFavAdmin class.

Source code in vast_pipeline/admin.py
74
75
76
77
78
79
80
class SourceFavAdmin(admin.ModelAdmin):
    """
    The SourceFavAdmin class.
    """
    list_display = ('user', 'source', 'comment')
    list_filter = ('user',)
    search_fields = ('user','source')

SourceQuerySet

Bases: QuerySet

Source code in vast_pipeline/models.py
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
class SourceQuerySet(models.QuerySet):

    def cone_search(
        self, ra: float, dec: float, radius_deg: float
    ) -> models.QuerySet:
        """
        Return all the Sources withing radius_deg of (ra,dec).
        Returns a QuerySet of Sources, ordered by distance from
        (ra,dec) ascending.

        Args:
            ra: The right ascension value of the cone search central
                coordinate.
            dec: The declination value of the cone search central coordinate.
            radius_deg: The radius over which to perform the cone search.

        Returns:
            Sources found withing the cone search area.
        """
        return (
            self.extra(
                select={
                    "distance": "q3c_dist(wavg_ra, wavg_dec, %s, %s) * 3600"
                },
                select_params=[ra, dec],
                where=["q3c_radial_query(wavg_ra, wavg_dec, %s, %s, %s)"],
                params=[ra, dec, radius_deg],
            )
            .order_by("distance")
        )

Return all the Sources withing radius_deg of (ra,dec). Returns a QuerySet of Sources, ordered by distance from (ra,dec) ascending.

Parameters:

Name Type Description Default
ra float

The right ascension value of the cone search central coordinate.

required
dec float

The declination value of the cone search central coordinate.

required
radius_deg float

The radius over which to perform the cone search.

required

Returns:

Type Description
QuerySet

Sources found withing the cone search area.

Source code in vast_pipeline/models.py
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
def cone_search(
    self, ra: float, dec: float, radius_deg: float
) -> models.QuerySet:
    """
    Return all the Sources withing radius_deg of (ra,dec).
    Returns a QuerySet of Sources, ordered by distance from
    (ra,dec) ascending.

    Args:
        ra: The right ascension value of the cone search central
            coordinate.
        dec: The declination value of the cone search central coordinate.
        radius_deg: The radius over which to perform the cone search.

    Returns:
        Sources found withing the cone search area.
    """
    return (
        self.extra(
            select={
                "distance": "q3c_dist(wavg_ra, wavg_dec, %s, %s) * 3600"
            },
            select_params=[ra, dec],
            where=["q3c_radial_query(wavg_ra, wavg_dec, %s, %s, %s)"],
            params=[ra, dec, radius_deg],
        )
        .order_by("distance")
    )

calculate_m_metric(flux_a, flux_b)

Calculate the m variability metric which is the modulation index between two fluxes. This is proportional to the fractional variability. See Section 5 of Mooley et al. (2016) for details, DOI: 10.3847/0004-637X/818/2/105.

Parameters:

Name Type Description Default
flux_a float

flux value "A".

required
flux_b float

flux value "B".

required

Returns:

Name Type Description
float float

the m metric for flux values "A" and "B".

Source code in vast_pipeline/pipeline/pairs.py
33
34
35
36
37
38
39
40
41
42
43
44
45
def calculate_m_metric(flux_a: float, flux_b: float) -> float:
    """Calculate the m variability metric which is the modulation index between two fluxes.
    This is proportional to the fractional variability.
    See Section 5 of Mooley et al. (2016) for details, DOI: 10.3847/0004-637X/818/2/105.

    Args:
        flux_a (float): flux value "A".
        flux_b (float): flux value "B".

    Returns:
        float: the m metric for flux values "A" and "B".
    """
    return 2 * ((flux_a - flux_b) / (flux_a + flux_b))

calculate_vs_metric(flux_a, flux_b, flux_err_a, flux_err_b)

Calculate the Vs variability metric which is the t-statistic that the provided fluxes are variable. See Section 5 of Mooley et al. (2016) for details, DOI: 10.3847/0004-637X/818/2/105.

Parameters:

Name Type Description Default
flux_a float

flux value "A".

required
flux_b float

flux value "B".

required
flux_err_a float

error of flux_a.

required
flux_err_b float

error of flux_b.

required

Returns:

Name Type Description
float float

the Vs metric for flux values "A" and "B".

Source code in vast_pipeline/pipeline/pairs.py
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
def calculate_vs_metric(
    flux_a: float, flux_b: float, flux_err_a: float, flux_err_b: float
) -> float:
    """Calculate the Vs variability metric which is the t-statistic that the provided
    fluxes are variable. See Section 5 of Mooley et al. (2016) for details,
    DOI: 10.3847/0004-637X/818/2/105.

    Args:
        flux_a (float): flux value "A".
        flux_b (float): flux value "B".
        flux_err_a (float): error of `flux_a`.
        flux_err_b (float): error of `flux_b`.

    Returns:
        float: the Vs metric for flux values "A" and "B".
    """
    return (flux_a - flux_b) / np.hypot(flux_err_a, flux_err_b)