Source code for astrodata.provenance

from astropy.table import Table
from datetime import datetime


PROVENANCE_DATE_FORMAT = "%Y-%m-%d %H:%M:%S.%f"


[docs]def add_provenance(ad, filename, md5, primitive, timestamp=None): """ Add the given provenance entry to the full set of provenance records on this object. Provenance is not added if the incoming md5 is None or ''. These indicate source data for the provenance that are not on disk, so not useful as provenance. Parameters ---------- ad : `astrodata.AstroData` filename : str md5 : str primitive : str timestamp : `datetime.datetime` """ if md5 is None or md5 == '': # not a real input, we can ignore this one return if timestamp is None: timestamp = datetime.now().strftime(PROVENANCE_DATE_FORMAT) if hasattr(ad, 'PROVENANCE'): existing_provenance = ad.PROVENANCE for row in existing_provenance: if row[1] == filename and \ row[2] == md5 and \ row[3] == primitive: # nothing needed, we already have it return if not hasattr(ad, 'PROVENANCE'): timestamp_data = [timestamp] filename_data = [filename] md5_data = [md5] provenance_added_by_data = [primitive] ad.PROVENANCE = Table( [timestamp_data, filename_data, md5_data, provenance_added_by_data], names=('timestamp', 'filename', 'md5', 'provenance_added_by'), dtype=('S28', 'S128', 'S128', 'S128')) else: provenance = ad.PROVENANCE provenance.add_row((timestamp, filename, md5, primitive))
[docs]def add_provenance_history(ad, timestamp_start, timestamp_stop, primitive, args): """ Add the given ProvenanceHistory entry to the full set of history records on this object. Parameters ---------- ad : `astrodata.AstroData` AstroData object to add history record to. timestamp_start : `datetime.datetime` Date of the start of this operation. timestamp_stop : `datetime.datetime` Date of the end of this operation. primitive : str Name of the primitive performed. args : str Arguments used for the primitive call. """ if hasattr(ad, 'PROVENANCE_HISTORY'): for row in ad.PROVENANCE_HISTORY: if timestamp_start == row[0] and \ timestamp_stop == row[1] and \ primitive == row[2] and \ args == row[3]: # already in the history, skip return colsize = len(args)+1 if hasattr(ad, 'PROVENANCE_HISTORY'): colsize = max(colsize, max(len(ph[3]) for ph in ad.PROVENANCE_HISTORY) + 1) timestamp_start_arr = [ph[0] for ph in ad.PROVENANCE_HISTORY] timestamp_stop_arr = [ph[1] for ph in ad.PROVENANCE_HISTORY] primitive_arr = [ph[2] for ph in ad.PROVENANCE_HISTORY] args_arr = [ph[3] for ph in ad.PROVENANCE_HISTORY] else: timestamp_start_arr = [] timestamp_stop_arr = [] primitive_arr = [] args_arr = [] timestamp_start_arr.append(timestamp_start) timestamp_stop_arr.append(timestamp_stop) primitive_arr.append(primitive) args_arr.append(args) ad.PROVENANCE_HISTORY = Table( [timestamp_start_arr, timestamp_stop_arr, primitive_arr, args_arr], names=('timestamp_start', 'timestamp_stop', 'primitive', 'args'), dtype=("S28", "S28", "S128", "S%d" % colsize) )
[docs]def clone_provenance(provenance_data, ad): """ For a single input's provenance, copy it into the output `AstroData` object as appropriate. This takes a dictionary with a source filename, md5 and both it's original provenance and provenance_history information. It duplicates the provenance data into the outgoing `AstroData` ad object. Parameters ---------- provenance_data : Pointer to the `~astrodata.AstroData` table with the provenance information. *Note* this may be the output `~astrodata.AstroData` as well, so we need to handle that. ad : `astrodata.AstroData` Outgoing `~astrodata.AstroData` object to add provenance data to. """ pd = [(prov[1], prov[2], prov[3], prov[0]) for prov in provenance_data] for p in pd: add_provenance(ad, p[0], p[1], p[2], timestamp=p[3])
[docs]def clone_provenance_history(provenance_history_data, ad): """ For a single input's provenance history, copy it into the output `AstroData` object as appropriate. This takes a dictionary with a source filename, md5 and both it's original provenance and provenance_history information. It duplicates the provenance data into the outgoing `AstroData` ad object. Parameters ---------- provenance_history_data : pointer to the `AstroData` table with the history information. *Note* this may be the output `~astrodata.AstroData` as well, so we need to handle that. ad : `astrodata.AstroData` Outgoing `~astrodata.AstroData` object to add provenance history data to. """ phd = [(prov_hist[0], prov_hist[1], prov_hist[2], prov_hist[3]) for prov_hist in provenance_history_data] for ph in phd: add_provenance_history(ad, ph[0], ph[1], ph[2], ph[3])