Source code for tkp.steps.forced_fitting

import logging
from tkp.accessors import sourcefinder_image_from_accessor
from tkp.db import general as dbgen
from tkp.db import monitoringlist as dbmon
from tkp.db import nulldetections as dbnd

logger = logging.getLogger(__name__)


def get_forced_fit_requests(image, expiration):
    nd_requested_fits = dbnd.get_nulldetections(image.id, expiration)
    logger.debug("Found %s null detections" % len(nd_requested_fits))
    mon_entries = dbmon.get_monitor_entries(image.dataset.id)

    all_fit_positions = []
    all_fit_ids = []

    all_fit_positions.extend([(nd[1],nd[2]) for nd in nd_requested_fits])
    all_fit_ids.extend([ ('ff_nd', nd[0]) for nd in nd_requested_fits])

    all_fit_positions.extend([(ms[1],ms[2]) for ms in mon_entries])
    all_fit_ids.extend([('ff_ms',ms[0]) for ms in mon_entries])

    return all_fit_positions, all_fit_ids


def insert_and_associate_forced_fits(image_id,successful_fits,successful_ids):
    assert len(successful_ids) == len(successful_fits)

    nd_extractions=[]
    nd_runcats=[]
    ms_extractions=[]
    ms_ids = []

    for idx, id in enumerate(successful_ids):
        if id[0] == 'ff_nd':
            nd_extractions.append(successful_fits[idx])
            nd_runcats.append(id[1])
        elif id[0] == 'ff_ms':
            ms_extractions.append(successful_fits[idx])
            ms_ids.append(id[1])
        else:
            raise ValueError("Forced fit type id not recognised:" + id[0])

    if nd_extractions:
        logger.debug("adding null detections")
        dbgen.insert_extracted_sources(image_id, nd_extractions,
                                       extract_type='ff_nd',
                                       ff_runcat_ids=nd_runcats)
        dbnd.associate_nd(image_id)
    else:
        logger.debug("No successful nulldetection fits")

    if ms_extractions:
        dbgen.insert_extracted_sources(image_id, ms_extractions,
                                       extract_type='ff_ms',
                                       ff_monitor_ids=ms_ids)
        logger.debug("adding monitoring sources")
        dbmon.associate_ms(image_id)
    else:
        logger.debug("No successful monitor fits")


[docs]def perform_forced_fits(fit_posns, fit_ids, accessor, extraction_params): """ Perform forced source measurements on an image based on a list of positions. Args: fit_posns (tuple): List of (RA, Dec) tuples: Positions to be fit. fit_ids: List of identifiers for each requested fit position. image_path (str): path to image for measurements. extraction_params (dict): source extraction parameters, as a dictionary. Returns: tuple: A matched pair of lists (serialized_fits, ids), corresponding to successfully fitted positions. NB returned lists may be shorter than input lists if some fits are unsuccessful. """ logger.debug("Forced fitting in image: %s" % (accessor.url)) if not len(fit_ids): logging.debug("nothing to force fit") return [], [] margin = extraction_params['margin'] radius = extraction_params['extraction_radius_pix'] back_size_x = extraction_params['back_size_x'] back_size_y = extraction_params['back_size_y'] data_image = sourcefinder_image_from_accessor(accessor, margin=margin, radius=radius, back_size_x=back_size_x, back_size_y=back_size_y) box_in_beampix = extraction_params['box_in_beampix'] boxsize = box_in_beampix * max(data_image.beam[0], data_image.beam[1]) fits = data_image.fit_fixed_positions( fit_posns, boxsize, ids=fit_ids) successful_fits, successful_ids = fits if successful_fits: serialized = [ f.serialize( extraction_params['ew_sys_err'], extraction_params['ns_sys_err']) for f in successful_fits] return serialized, successful_ids else: return [], []