Repository: winter-telescope/mirar (language: python)

Issue:
[FEATURE] WINTER pipeline end-to-end
**Is your feature request related to a problem? Please describe.**
I'm always frustrated when the WINTER data is not fully processed.

**Describe the solution you'd like**
WINTER data is fully processed.

[FEATURE] Upload photometry to SkyPortal
**Is your feature request related to a problem? Please describe.**
I'm always frustrated when I perform a reduction and the results just sit on my computer.

**Describe the solution you'd like**
I'd like for the photometry results to be uploaded to Fritz!
[FEATURE] Reorganise Procesors
**Is your feature request related to a problem? Please describe.**
I'm always frustrated when I try and find a processor, because our structure is rather illogical.

**Describe the solution you'd like**
One example is `photcal.py`, which is not in the `photometry` directory. Or `send_to_fritz` and `xmatch`, which are not in the `candidates` or `avro` directories. The calibration processors are not grouped together. `reference` is split off from ZOGY. Maybe a hierarchy of the three classes of processor, and then logical subdirectories as needed?

**Describe alternatives you've considered**
Letting anarchy reign.



Discussion / hints:
@virajkaram, re astrometry.net : "we need only one series 5206"
We will need to install astrometry.net in the CI, which is also a bottleneck to #446 / #422 . We can do it with conda.

partially addressed by #286 

Fix patch:
diff --git a/.github/workflows/continuous_integration.yml b/.github/workflows/continuous_integration.yml
index 67c8633b8..4695350a8 100644
--- a/.github/workflows/continuous_integration.yml
+++ b/.github/workflows/continuous_integration.yml
@@ -69,6 +69,12 @@ jobs:
           curl -o /usr/share/miniconda/envs/test/data/index-5201-04.fits https://portal.nersc.gov/project/cosmo/temp/dstn/index-5200/LITE/index-5201-04.fits
           curl -o /usr/share/miniconda/envs/test/data/index-5204-13.fits https://portal.nersc.gov/project/cosmo/temp/dstn/index-5200/LITE/index-5204-13.fits
           curl -o /usr/share/miniconda/envs/test/data/index-5205-13.fits https://portal.nersc.gov/project/cosmo/temp/dstn/index-5200/LITE/index-5205-13.fits
+          # WINTER
+          curl -o /usr/share/miniconda/envs/test/data/index-5205-11.fits https://portal.nersc.gov/project/cosmo/temp/dstn/index-5200/LITE/index-5205-11.fits
+          curl -o /usr/share/miniconda/envs/test/data/index-5203-11.fits https://portal.nersc.gov/project/cosmo/temp/dstn/index-5200/LITE/index-5203-11.fits
+          curl -o /usr/share/miniconda/envs/test/data/index-5206-11.fits https://portal.nersc.gov/project/cosmo/temp/dstn/index-5200/LITE/index-5206-11.fits
+          curl -o /usr/share/miniconda/envs/test/data/index-5204-11.fits https://portal.nersc.gov/project/cosmo/temp/dstn/index-5200/LITE/index-5204-11.fits
+
 
       # Set up database
       - name: postgres
diff --git a/.github/workflows/pylint.yml b/.github/workflows/pylint.yml
index f3e6d7cb3..7d4bf1533 100644
--- a/.github/workflows/pylint.yml
+++ b/.github/workflows/pylint.yml
@@ -26,4 +26,4 @@ jobs:
           poetry install
       - name: Analysing the code with pylint
         run: |
-           poetry run pylint mirar --fail-under=9.78
+           poetry run pylint mirar --fail-under=9.80
diff --git a/mirar/paths.py b/mirar/paths.py
index 6fba1b697..4b4bcf97e 100644
--- a/mirar/paths.py
+++ b/mirar/paths.py
@@ -291,7 +291,7 @@ def get_astrometry_keys() -> list:
 XPOS_KEY = "xpos"
 YPOS_KEY = "ypos"
 APFLUX_PREFIX_KEY = "fluxap"
-APFLUXUNC_PREFIX_KEY = "fluxucap"
+APFLUXUNC_PREFIX_KEY = "fluxuncap"
 APMAG_PREFIX_KEY = "magap"
 APMAGUNC_PREFIX_KEY = "sigmagap"
 
diff --git a/mirar/pipelines/sedmv2/blocks.py b/mirar/pipelines/sedmv2/blocks.py
index 356b9d2f3..b76c0d6cf 100644
--- a/mirar/pipelines/sedmv2/blocks.py
+++ b/mirar/pipelines/sedmv2/blocks.py
@@ -28,14 +28,9 @@
 from mirar.processors.csvlog import CSVLog
 from mirar.processors.mask import MaskPixelsFromPath
 from mirar.processors.photcal import PhotCalibrator
-from mirar.processors.photometry.aperture_photometry import (
-    ImageAperturePhotometry,
-    SourceAperturePhotometry,
-)
-from mirar.processors.photometry.psf_photometry import (  # ImagePSFPhotometry,
-    SourcePSFPhotometry,
-)
+from mirar.processors.photometry import AperturePhotometry, PSFPhotometry
 from mirar.processors.reference import ProcessReference
+from mirar.processors.sources import ForcedPhotometryDetector
 from mirar.processors.utils import (
     ImageBatcher,
     ImageDebatcher,
@@ -141,8 +136,9 @@
 process_stellar = reduce + resample_stellar + calibrate
 
 image_photometry = [  # imported from wirc/blocks.py
-    # ImageSelector(("OBSTYPE", "SCIENCE")),
-    ImageAperturePhotometry(
+    ImageSaver(output_dir_name="photometry"),
+    ForcedPhotometryDetector(ra_header_key="OBJRAD", dec_header_key="OBJDECD"),
+    AperturePhotometry(
         aper_diameters=[
             2 / SEDMV2_PIXEL_SCALE,
             3 / SEDMV2_PIXEL_SCALE,
@@ -166,29 +162,19 @@
         ],
         col_suffix_list=["2", "3", "4", "5", "10"],
         phot_cutout_size=100,
-        target_ra_key="OBJRAD",
-        target_dec_key="OBJDECD",
         zp_key="ZP_AUTO",
     ),
-    # Sextractor(**sextractor_reference_config, output_sub_dir="psf", cache=False),
-    # PSFex(config_path=psfex_config_path, output_sub_dir="psf", norm_fits=True),
-    # ImagePSFPhotometry(
-    #     target_ra_key="OBJRAD",
-    #     target_dec_key="OBJDECD",
-    #     zp_colname="ZP_AUTO",
-    # ),
-    ImageSaver(output_dir_name="photometry"),
 ]
 
 candidate_photometry = [  # imported from wirc/blocks.py
-    SourceAperturePhotometry(
+    AperturePhotometry(
         aper_diameters=[16, 70],
         phot_cutout_size=100,
         bkg_in_diameters=[25, 90],
         bkg_out_diameters=[40, 100],
         col_suffix_list=["", "big"],
     ),
-    SourcePSFPhotometry(),
+    PSFPhotometry(),
 ]
 
 
diff --git a/mirar/pipelines/summer/blocks.py b/mirar/pipelines/summer/blocks.py
index 74985f157..8f93122ab 100644
--- a/mirar/pipelines/summer/blocks.py
+++ b/mirar/pipelines/summer/blocks.py
@@ -42,8 +42,7 @@
 from mirar.processors.database.database_updater import ImageSequenceDatabaseUpdater
 from mirar.processors.mask import MaskPixelsFromPath
 from mirar.processors.photcal import PhotCalibrator
-from mirar.processors.photometry.aperture_photometry import SourceAperturePhotometry
-from mirar.processors.photometry.psf_photometry import SourcePSFPhotometry
+from mirar.processors.photometry import AperturePhotometry, PSFPhotometry
 from mirar.processors.reference import ProcessReference
 from mirar.processors.sources import SourceWriter
 from mirar.processors.sources.source_detector import ZOGYSourceDetector
@@ -229,8 +228,8 @@
 extract_candidates = [
     ZOGYSourceDetector(output_sub_dir="subtract", **sextractor_candidates_config),
     RegionsWriter(output_dir_name="candidates"),
-    SourcePSFPhotometry(),
-    SourceAperturePhotometry(
+    PSFPhotometry(),
+    AperturePhotometry(
         aper_diameters=[8, 40],
         phot_cutout_size=100,
         bkg_in_diameters=[25, 90],
diff --git a/mirar/pipelines/winter/blocks.py b/mirar/pipelines/winter/blocks.py
index d2b5cf879..2842efd47 100644
--- a/mirar/pipelines/winter/blocks.py
+++ b/mirar/pipelines/winter/blocks.py
@@ -51,7 +51,6 @@
     annotate_winter_subdet_headers,
     get_raw_winter_mask,
     load_stacked_winter_image,
-    load_test_stacked_winter_image,
     load_test_winter_image,
     load_winter_mef_image,
 )
@@ -99,12 +98,11 @@
     MaskPixelsFromFunction,
 )
 from mirar.processors.photcal import PhotCalibrator
-from mirar.processors.photometry.aperture_photometry import SourceAperturePhotometry
-from mirar.processors.photometry.psf_photometry import SourcePSFPhotometry
+from mirar.processors.photometry import AperturePhotometry, PSFPhotometry
 from mirar.processors.reference import GetReferenceImage, ProcessReference
 from mirar.processors.sources import (
     CandidateNamer,
-    CustomSourceModifier,
+    CustomSourceTableModifier,
     SourceLoader,
     SourceWriter,
     ZOGYSourceDetector,
@@ -128,11 +126,31 @@
         input_sub_dir="raw",
         load_image=load_winter_mef_image,
     ),
+    ImageBatcher("UTCTIME"),
+    CSVLog(
+        export_keys=[
+            "UTCTIME",
+            "PROGNAME",
+            DITHER_N_KEY,
+            MAX_DITHER_KEY,
+            "FILTER",
+            EXPTIME_KEY,
+            OBSCLASS_KEY,
+            "BOARD_ID",
+            "BASENAME",
+            TARGET_KEY,
+            "RADEG",
+            "DECDEG",
+            "T_ROIC",
+            "FIELDID",
+        ]
+    ),
     ImageSelector(
-        ("BOARD_ID", "4"),
+        ("BOARD_ID", "2"),
         (OBSCLASS_KEY, ["dark", "science"]),
         (EXPTIME_KEY, "120.0"),
-        ("FIELDID", ["9767", str(DEFAULT_FIELD)]),
+        ("filter", ["dark", "J"]),
+        ("FIELDID", ["3944", str(DEFAULT_FIELD)]),
     ),
     ImageSaver("testdata", output_dir=get_test_data_dir()),
 ]
@@ -404,16 +422,6 @@
 
 # Image subtraction
 
-load_test_stack = [
-    ImageLoader(
-        input_img_dir=get_test_data_dir(),
-        input_sub_dir="final",
-        load_image=load_test_stacked_winter_image,
-    ),
-    ImageBatcher(["BOARD_ID", "FILTER", TARGET_KEY, "SUBCOORD"]),
-    DatabaseImageInserter(db_table=Stack, duplicate_protocol="replace"),
-]
-
 load_stack = [
     ImageLoader(input_sub_dir="final", input_img_dir=base_output_dir),
     ImageBatcher(["BOARD_ID", "FILTER", TARGET_KEY, "SUBCOORD"]),
@@ -442,29 +450,32 @@
     ),
     ImageSaver(output_dir_name="diffs"),
     DatabaseImageInserter(db_table=Diff, duplicate_protocol="replace"),
+    ImageSaver(output_dir_name="subtract"),
 ]
 
+load_sub = [
+    ImageLoader(input_sub_dir="subtract"),
+]
 detect_candidates = [
     ZOGYSourceDetector(
         output_sub_dir="subtract",
         **sextractor_candidate_config,
-        copy_image_keywords=["stackid", "progname"],
     ),
-    SourcePSFPhotometry(),
-    SourceAperturePhotometry(
+    PSFPhotometry(),
+    AperturePhotometry(
         aper_diameters=[16, 70],
         phot_cutout_size=100,
         bkg_in_diameters=[25, 90],
         bkg_out_diameters=[40, 100],
         col_suffix_list=["", "big"],
     ),
-    CustomSourceModifier(winter_candidate_annotator_filterer),
+    CustomSourceTableModifier(winter_candidate_annotator_filterer),
     SourceWriter(output_dir_name="candidates"),
 ]
 #
 # candidate_colnames = get_column_names_from_schema(winter_candidate_config)
 
-load_candidates = [
+load_sources = [
     SourceLoader(input_dir_name="candidates"),
 ]
 
@@ -492,7 +503,9 @@
         db_output_columns=prv_candidate_cols + [CAND_NAME_KEY],
         additional_query_constraints=winter_history_deprecated_constraint,
     ),
-    CustomSourceModifier(modifier_function=winter_candidate_avro_fields_calculator),
+    CustomSourceTableModifier(
+        modifier_function=winter_candidate_avro_fields_calculator
+    ),
     DatabaseSourceInserter(
         db_table=Candidate,
         duplicate_protocol="fail",
diff --git a/mirar/pipelines/winter/generator.py b/mirar/pipelines/winter/generator.py
index 10d79694e..d5b0e704f 100644
--- a/mirar/pipelines/winter/generator.py
+++ b/mirar/pipelines/winter/generator.py
@@ -7,13 +7,13 @@
 
 import numpy as np
 import pandas as pd
-from astropy.io import fits
 from astropy.table import Table
+from astropy.time import Time
 
 from mirar.catalog import Gaia2Mass
 from mirar.catalog.base_catalog import CatalogFromFile
 from mirar.catalog.vizier import PS1
-from mirar.data import Image, ImageBatch
+from mirar.data import Image, ImageBatch, SourceBatch
 from mirar.data.utils.compress import decode_img
 from mirar.database.constraints import DBQueryConstraints
 from mirar.database.transactions import select_from_table
@@ -23,7 +23,7 @@
     OBSCLASS_KEY,
     REF_CAT_PATH_KEY,
     SATURATE_KEY,
-    SCI_IMG_KEY,
+    TIME_KEY,
     ZP_KEY,
     ZP_STD_KEY,
     get_output_dir,
@@ -293,92 +293,103 @@ def winter_stackid_annotator(batch: ImageBatch) -> ImageBatch:
     return batch
 
 
-def winter_candidate_annotator_filterer(src_df: pd.DataFrame):
+def winter_candidate_annotator_filterer(source_batch: SourceBatch) -> SourceBatch:
     """
     Function to perform basic filtering to weed out bad WIRC candidates with None
     magnitudes, to be added.
-    :param src_df: Source dataframe
-    :return: filtered dataframe
+    :param source_batch: Source batch
+    :return: updated batch
     """
 
-    none_mask = (
-        src_df["sigmapsf"].isnull()
-        | src_df["magpsf"].isnull()
-        | src_df["magap"].isnull()
-        | src_df["sigmagap"].isnull()
-    )
+    new_batch = SourceBatch([])
 
-    mask = none_mask.values
+    for source in source_batch:
+        src_df = source.get_data()
 
-    # Needing to do this because the dataframe is big-endian
-    mask_inds = np.where(~mask)[0]
-    filtered_df = pd.DataFrame([src_df.loc[x] for x in mask_inds]).reset_index(
-        drop=True
-    )
+        none_mask = (
+            src_df["sigmapsf"].isnull()
+            | src_df["magpsf"].isnull()
+            | src_df["magap"].isnull()
+            | src_df["sigmagap"].isnull()
+        )
+
+        mask = none_mask.values
+
+        # Needing to do this because the dataframe is big-endian
+        mask_inds = np.where(~mask)[0]
+        filtered_df = pd.DataFrame([src_df.loc[x] for x in mask_inds]).reset_index(
+            drop=True
+        )
 
-    if len(filtered_df) == 0:
-        filtered_df = pd.DataFrame(columns=src_df.columns)
+        if len(filtered_df) == 0:
+            filtered_df = pd.DataFrame(columns=src_df.columns)
 
-    # Pipeline (db) specific keywords
-    filtered_df.loc[:, "magzpsci"] = filtered_df.loc[:, ZP_KEY]
-    filtered_df.loc[:, "magzpsciunc"] = filtered_df.loc[:, ZP_STD_KEY]
-    filtered_df.loc[:, "diffmaglim"] = filtered_df.loc[:, MAGLIM_KEY]
+        # Pipeline (db) specific keywords
+        source["magzpsci"] = source[ZP_KEY]
+        source["magzpsciunc"] = source[ZP_STD_KEY]
+        source["diffmaglim"] = source[MAGLIM_KEY]
+        source["programpi"] = source["PROGPI"]
+        source["programid"] = source["PROGID"]
+        source["field"] = source["FIELDID"]
+        source["jd"] = Time(source[TIME_KEY]).jd
 
-    if len(filtered_df) > 0:
-        sci_resamp_image_path = filtered_df.loc[0, SCI_IMG_KEY]
-        filtered_df.loc[:, "field"] = fits.getval(sci_resamp_image_path, "FIELDID")
-        filtered_df.loc[:, "programpi"] = fits.getval(sci_resamp_image_path, "PROGPI")
-        filtered_df.loc[:, "programid"] = fits.getval(sci_resamp_image_path, "PROGID")
-        filtered_df.loc[:, "fid"] = fits.getval(sci_resamp_image_path, "FID")
-        filtered_df.loc[:, "deprecated"] = False
+        source.set_data(filtered_df)
+        new_batch.append(source)
 
-    return filtered_df
+    return new_batch
 
 
-def winter_candidate_avro_fields_calculator(src_df: pd.DataFrame) -> pd.DataFrame:
+def winter_candidate_avro_fields_calculator(source_table: SourceBatch) -> SourceBatch:
     """
     Function to calculate the AVRO fields for WINTER
     """
-    hist_dfs = [
-        pd.DataFrame(src_df["prv_candidates"].loc[x]) for x in range(len(src_df))
-    ]
 
-    jdstarthists, jdendhists = [], []
-    for ind, hist_df in enumerate(hist_dfs):
-        if len(hist_df) == 0:
-            jdstarthists.append(src_df.loc[ind]["jd"])
-            jdendhists.append(src_df.loc[ind]["jd"])
-        else:
-            jdstarthists.append(hist_df["jd"].min())
-            jdendhists.append(hist_df["jd"].max())
-    src_df["jdstarthist"] = jdstarthists
-    src_df["jdendhist"] = jdendhists
-    src_df["ndethist"] = [len(x) for x in hist_dfs]
-    src_df["magdiff"] = src_df["magpsf"] - src_df["magap"]
-    src_df["magfromlim"] = src_df["diffmaglim"] - src_df["magpsf"]
-    src_df["d_to_x"] = src_df["NAXIS1"] - src_df["xpos"]
-    src_df["d_to_y"] = src_df["NAXIS2"] - src_df["ypos"]
-    src_df["mindtoedge"] = src_df[["xpos", "ypos", "d_to_x", "d_to_y"]].min(axis=1)
-    src_df["cutout_science"] = src_df["cutoutScience"]
-    src_df["cutout_template"] = src_df["cutoutTemplate"]
-    src_df["cutout_difference"] = src_df["cutoutDifference"]
-    nnegs, nbads, sumrat = [], [], []
-    for _, src in src_df.iterrows():
-        diff_cutout_data = decode_img(src["cutout_difference"])
-        # Get central 5x5 pixels
-        nx, ny = diff_cutout_data.shape
-        diff_stamp = diff_cutout_data[
-            nx // 2 - 3 : nx // 2 + 2, ny // 2 - 3 : ny // 2 + 2
+    new_batch = SourceBatch([])
+
+    for source in source_table:
+        src_df = source.get_data()
+
+        src_df["magdiff"] = src_df["magpsf"] - src_df["magap"]
+        src_df["magfromlim"] = source["diffmaglim"] - src_df["magpsf"]
+
+        hist_dfs = [
+            pd.DataFrame(src_df["prv_candidates"].loc[x]) for x in range(len(src_df))
         ]
-        nnegs.append(np.sum(diff_stamp < 0))
-        nbads.append(np.sum(np.isnan(diff_stamp)))
-        sumrat.append(np.sum(diff_stamp) / np.sum(np.abs(diff_stamp)))
 
-    src_df["nneg"] = nnegs
-    src_df["nbad"] = nbads
-    src_df["sumrat"] = sumrat
+        jdstarthists, jdendhists = [], []
+        for _, hist_df in enumerate(hist_dfs):
+            if len(hist_df) == 0:
+                jdstarthists.append(source["jd"])
+                jdendhists.append(source["jd"])
+            else:
+                jdstarthists.append(hist_df["jd"].min())
+                jdendhists.append(hist_df["jd"].max())
+        src_df["jdstarthist"] = min(jdstarthists)
+        src_df["jdendhist"] = max(jdendhists)
+        src_df["ndethist"] = [len(x) for x in hist_dfs]
+        src_df["d_to_x"] = src_df["NAXIS1"] - src_df["xpos"]
+        src_df["d_to_y"] = src_df["NAXIS2"] - src_df["ypos"]
+        src_df["mindtoedge"] = src_df[["xpos", "ypos", "d_to_x", "d_to_y"]].min(axis=1)
+        nnegs, nbads, sumrat = [], [], []
+        for _, src in src_df.iterrows():
+            diff_cutout_data = decode_img(src["cutout_difference"])
+            # Get central 5x5 pixels
+            nx, ny = diff_cutout_data.shape
+            diff_stamp = diff_cutout_data[
+                nx // 2 - 3 : nx // 2 + 2, ny // 2 - 3 : ny // 2 + 2
+            ]
+            nnegs.append(np.sum(diff_stamp < 0))
+            nbads.append(np.sum(np.isnan(diff_stamp)))
+            sumrat.append(np.sum(diff_stamp) / np.sum(np.abs(diff_stamp)))
+
+        src_df["nneg"] = nnegs
+        src_df["nbad"] = nbads
+        src_df["sumrat"] = sumrat
+
+        source.set_data(src_df)
+        new_batch.append(source)
 
-    return src_df
+    return new_batch
 
 
 def winter_reference_stack_annotator(stacked_image: Image, image: Image) -> Image:
diff --git a/mirar/pipelines/winter/load_winter_image.py b/mirar/pipelines/winter/load_winter_image.py
index 4daedcab1..5f6b0d8a8 100644
--- a/mirar/pipelines/winter/load_winter_image.py
+++ b/mirar/pipelines/winter/load_winter_image.py
@@ -25,8 +25,6 @@
     BASE_NAME_KEY,
     COADD_KEY,
     GAIN_KEY,
-    LATEST_SAVE_KEY,
-    LATEST_WEIGHT_SAVE_KEY,
     OBSCLASS_KEY,
     PROC_FAIL_KEY,
     PROC_HISTORY_KEY,
@@ -61,6 +59,7 @@ def clean_header(header: fits.Header) -> fits.Header:
     date_t = Time(header["UTCTIME"])
 
     header["MJD-OBS"] = date_t.mjd
+    header["JD"] = date_t.jd
     header["DATE-OBS"] = date_t.isot
 
     header[OBSCLASS_KEY] = header["OBSTYPE"].lower()
@@ -231,24 +230,6 @@ def load_test_winter_image(
     return image
 
 
-def load_test_stacked_winter_image(path: str | Path) -> Image:
-    """
-    Load test stacked WINTER image
-
-    :param path: Path to image
-    :return: Image object
-    """
-    if isinstance(path, str):
-        path = Path(path)
-    image = open_raw_image(path)
-
-    image[LATEST_SAVE_KEY] = path.as_posix()
-    weight_image_name = Path(image[LATEST_WEIGHT_SAVE_KEY]).name
-    weight_path = path.parent / "weight" / weight_image_name
-    image[LATEST_WEIGHT_SAVE_KEY] = weight_path.as_posix()
-    return image
-
-
 def load_raw_winter_mef(
     path: str | Path,
 ) -> tuple[astropy.io.fits.Header, list[np.array], list[astropy.io.fits.Header]]:
diff --git a/mirar/pipelines/winter/winter_pipeline.py b/mirar/pipelines/winter/winter_pipeline.py
index 2dc273cbc..93c8bb0d5 100644
--- a/mirar/pipelines/winter/winter_pipeline.py
+++ b/mirar/pipelines/winter/winter_pipeline.py
@@ -12,11 +12,12 @@
     csvlog,
     detect_candidates,
     detrend_unpacked,
+    extract_all,
+    full_reduction,
     imsub,
-    load_candidates,
     load_stack,
     load_test,
-    load_test_stack,
+    mask_and_split,
     only_ref,
     photcal_stacks,
     process_candidates,
@@ -25,6 +26,8 @@
     reduce_unpacked,
     refbuild,
     reftest,
+    save_raw,
+    select_split_subset,
     unpack_all,
     unpack_subset,
 )
@@ -46,19 +49,26 @@ class WINTERPipeline(Pipeline):
         "unpack_subset": unpack_subset,
         "unpack_all": unpack_all,
         "detrend_unpacked": detrend_unpacked,
+        "imsub": load_stack + imsub + detect_candidates,
         "reduce": reduce,
         "reduce_unpacked": reduce_unpacked,
         "photcal_stacks": photcal_stacks,
         "buildtest": build_test,
-        "test": load_test + csvlog + realtime,
-        "test_imsub": load_test_stack + imsub + detect_candidates + process_candidates,
+        "test": load_test
+        + csvlog
+        + extract_all
+        + mask_and_split
+        + select_split_subset
+        + save_raw
+        + full_reduction
+        + imsub
+        + detect_candidates
+        + process_candidates,
         "refbuild": refbuild,
         "reftest": reftest,
         "only_ref": only_ref,
         "realtime": realtime,
-        "imsub": load_stack + imsub,
         "detect_candidates": load_stack + imsub + detect_candidates,
-        "process_candidates": load_candidates + process_candidates,
         "full_imsub": load_stack + imsub + detect_candidates + process_candidates,
     }
 
diff --git a/mirar/pipelines/wirc/blocks.py b/mirar/pipelines/wirc/blocks.py
index 886d2c57e..f3ccdd962 100644
--- a/mirar/pipelines/wirc/blocks.py
+++ b/mirar/pipelines/wirc/blocks.py
@@ -9,7 +9,9 @@
     LATEST_SAVE_KEY,
     OBSCLASS_KEY,
     RAW_IMG_KEY,
+    REF_IMG_KEY,
     SATURATE_KEY,
+    SCI_IMG_KEY,
 )
 from mirar.pipelines.wirc.generator import (
     wirc_astrometric_catalog_generator,
@@ -47,7 +49,7 @@
 from mirar.processors.astromatic.swarp import ReloadSwarpComponentImages
 from mirar.processors.astrometry.autoastrometry import AutoAstrometry
 from mirar.processors.astrometry.utils import AstrometryFromFile
-from mirar.processors.avro import IPACAvroExporter, SendToFritz
+from mirar.processors.avro import IPACAvroExporter
 from mirar.processors.csvlog import CSVLog
 from mirar.processors.dark import DarkCalibrator
 from mirar.processors.database.database_inserter import DatabaseSourceInserter
@@ -64,19 +66,17 @@
     WriteMaskedCoordsToFile,
 )
 from mirar.processors.photcal import PhotCalibrator
-from mirar.processors.photometry.aperture_photometry import (
-    ImageAperturePhotometry,
-    SourceAperturePhotometry,
-)
-from mirar.processors.photometry.psf_photometry import (
-    ImagePSFPhotometry,
-    SourcePSFPhotometry,
-)
+from mirar.processors.photometry import AperturePhotometry, PSFPhotometry
 from mirar.processors.reference import ProcessReference
 from mirar.processors.sky import NightSkyMedianCalibrator
-from mirar.processors.sources import CandidateNamer, SourceWriter, ZOGYSourceDetector
-from mirar.processors.sources.source_table_builder import ForcedPhotometryCandidateTable
-from mirar.processors.sources.source_table_modifier import CustomSourceModifier
+from mirar.processors.skyportal import SkyportalCandidateUploader
+from mirar.processors.sources import (
+    CandidateNamer,
+    ForcedPhotometryDetector,
+    SourceWriter,
+    ZOGYSourceDetector,
+)
+from mirar.processors.sources.source_table_modifier import CustomSourceTableModifier
 from mirar.processors.sources.utils import RegionsWriter
 from mirar.processors.utils import (
     HeaderAnnotator,
@@ -197,51 +197,48 @@
     ImageSaver(output_dir_name="diffs"),
 ]
 
-image_photometry = [
-    ImageAperturePhotometry(
-        aper_diameters=[16],
-        bkg_in_diameters=[25],
-        bkg_out_diameters=[40],
-        col_suffix_list=[""],
-        phot_cutout_size=100,
-        target_ra_key="TARGRA",
-        target_dec_key="TARGDEC",
-    ),
-    Sextractor(**sextractor_reference_config, output_sub_dir="subtract", cache=False),
-    PSFex(config_path=psfex_path, output_sub_dir="photometry", norm_fits=True),
-    ImagePSFPhotometry(target_ra_key="TARGRA", target_dec_key="TARGDEC"),
-    ImageSaver(output_dir_name="photometry"),
-]
-
 export_candidates_from_header = [
-    ForcedPhotometryCandidateTable(
-        ra_header_key="TARGRA", dec_header_key="TARGDEC", name_header_key="TARGNAME"
-    ),
+    ForcedPhotometryDetector(ra_header_key="TARGRA", dec_header_key="TARGDEC"),
 ]
 
 candidate_photometry = [
-    SourceAperturePhotometry(
+    AperturePhotometry(
         aper_diameters=[16, 70],
         phot_cutout_size=100,
         bkg_in_diameters=[25, 90],
         bkg_out_diameters=[40, 100],
         col_suffix_list=["", "big"],
     ),
-    SourcePSFPhotometry(),
+    PSFPhotometry(),
 ]
 
 detect_candidates = [
     ZOGYSourceDetector(
         output_sub_dir="subtract",
         **sextractor_candidate_config,
-        copy_image_keywords=["PROGID", "PROGPI"],
     ),
 ]
+#
+image_photometry = [
+    Sextractor(**sextractor_reference_config, output_sub_dir="subtract", cache=False),
+    PSFex(config_path=psfex_path, output_sub_dir="photometry", norm_fits=True),
+    ImageSaver(output_dir_name="photometry"),
+    # AperturePhotometry(
+    #     aper_diameters=[16],
+    #     bkg_in_diameters=[25],
+    #     bkg_out_diameters=[40],
+    #     col_suffix_list=[""],
+    #     phot_cutout_size=100,
+    #     target_ra_key="TARGRA",
+    #     target_dec_key="TARGDEC",
+    # ),
+    # PSFPhotometry(),
+]
 
 process_candidates = [
     RegionsWriter(output_dir_name="candidates"),
-    SourcePSFPhotometry(),
-    SourceAperturePhotometry(
+    PSFPhotometry(),
+    AperturePhotometry(
         aper_diameters=[16, 70],
         phot_cutout_size=100,
         bkg_in_diameters=[25, 90],
@@ -249,7 +246,7 @@
         col_suffix_list=["", "big"],
     ),
     # SourceWriter(output_dir_name="candidates"),
-    CustomSourceModifier(modifier_function=wirc_source_table_filter_annotator),
+    CustomSourceTableModifier(modifier_function=wirc_source_table_filter_annotator),
     XMatch(catalog=TMASS(num_sources=3, search_radius_arcmin=0.5)),
     XMatch(catalog=PS1(num_sources=3, search_radius_arcmin=0.5)),
     SourceWriter(output_dir_name="kowalski"),
@@ -266,11 +263,13 @@
     ),
     DatabaseHistorySelector(
         crossmatch_radius_arcsec=2.0,
-        time_field_name="jd",
         history_duration_days=500.0,
         db_table=Candidate,
         db_output_columns=[CAND_NAME_KEY] + prv_candidate_cols,
     ),
+    HeaderAnnotator(input_keys=[LATEST_SAVE_KEY], output_key="diffimgname"),
+    HeaderAnnotator(input_keys=[SCI_IMG_KEY], output_key="sciimgname"),
+    HeaderAnnotator(input_keys=[REF_IMG_KEY], output_key="refimgname"),
     DatabaseSourceInserter(db_table=Candidate, duplicate_protocol="fail"),
     SourceWriter(output_dir_name="candidates"),
     # EdgeCandidatesMask(edge_boundary_size=100)
@@ -278,12 +277,13 @@
 ]
 
 package_candidates = [
+    HeaderAnnotator(input_keys=["MJD-OBS"], output_key="mjd"),
     IPACAvroExporter(
         base_name="WIRC",
         avro_schema_path=wirc_avro_schema_path,
     ),
-    SendToFritz(
-        base_name="WIRCTEST",
+    SkyportalCandidateUploader(
+        origin="WIRCTEST",
         group_ids=[1431],
         fritz_filter_id=74,
         instrument_id=5,
diff --git a/mirar/pipelines/wirc/generator.py b/mirar/pipelines/wirc/generator.py
index bb48d0fa5..c1f79ec15 100644
--- a/mirar/pipelines/wirc/generator.py
+++ b/mirar/pipelines/wirc/generator.py
@@ -10,9 +10,7 @@
 from astropy.table import Table
 
 from mirar.catalog import Gaia2Mass
-from mirar.data import Image
-from mirar.paths import DIFF_IMG_KEY, FILTER_KEY, MAGLIM_KEY, REF_IMG_KEY, SCI_IMG_KEY
-from mirar.pipelines.wirc.load_wirc_image import wirc_filter_dict
+from mirar.data import Image, SourceBatch
 from mirar.pipelines.wirc.wirc_files import (
     psfex_path,
     sextractor_reference_config,
@@ -24,34 +22,36 @@
 logger = logging.getLogger(__name__)
 
 
-def wirc_source_table_filter_annotator(src_df: pd.DataFrame) -> pd.DataFrame:
+def wirc_source_table_filter_annotator(source_table: SourceBatch) -> SourceBatch:
     """
     Function to remove bad candidates with None in sigmapsf, magpsf, magap, sigmagap,
     and update the source table with the keys required for the WIRC database
-    :param src_df: Source dataframe
-    :return: annotated dataframe
+    :param source_table: source table
+    :return: updated source table
     """
-    none_mask = (
-        src_df.loc[:, "sigmapsf"].isnull()
-        | src_df.loc[:, "magpsf"].isnull()
-        | src_df.loc[:, "magap"].isnull()
-        | src_df.loc[:, "sigmagap"].isnull()
-    )
 
-    mask = none_mask.values
+    new_batch = SourceBatch([])
+
+    for source in source_table:
+        src_df = source.get_data()
+
+        none_mask = (
+            src_df.loc[:, "sigmapsf"].isnull()
+            | src_df.loc[:, "magpsf"].isnull()
+            | src_df.loc[:, "magap"].isnull()
+            | src_df.loc[:, "sigmagap"].isnull()
+        )
+
+        mask = none_mask.values
+
+        # Needing to do this because the dataframe is big-endian
+        mask_inds = np.where(~mask)[0]
+        src_df = pd.DataFrame([src_df.loc[x] for x in mask_inds]).reset_index(drop=True)
 
-    # Needing to do this because the dataframe is big-endian
-    mask_inds = np.where(~mask)[0]
-    src_df = pd.DataFrame([src_df.loc[x] for x in mask_inds]).reset_index(drop=True)
+        source.set_data(src_df)
+        new_batch.append(source)
 
-    src_df.loc[:, "sciimgname"] = src_df.loc[:, SCI_IMG_KEY]
-    src_df.loc[:, "refimgname"] = src_df.loc[:, REF_IMG_KEY]
-    src_df.loc[:, "diffimgname"] = src_df.loc[:, DIFF_IMG_KEY]
-    src_df.loc[:, "fid"] = [wirc_filter_dict[x] for x in src_df.loc[:, FILTER_KEY]]
-    src_df.loc[:, "programid"] = src_df.loc[:, "progid"]
-    src_df.loc[:, "programpi"] = src_df.loc[:, "progpi"]
-    src_df.loc[:, "diffmaglim"] = src_df.loc[:, MAGLIM_KEY]
-    return src_df
+    return new_batch
 
 
 def wirc_photometric_img_catalog_purifier(catalog, image):
diff --git a/mirar/pipelines/wirc/load_wirc_image.py b/mirar/pipelines/wirc/load_wirc_image.py
index c99b1f737..a2e7b30c0 100644
--- a/mirar/pipelines/wirc/load_wirc_image.py
+++ b/mirar/pipelines/wirc/load_wirc_image.py
@@ -21,6 +21,7 @@
     ZP_KEY,
     ZP_STD_KEY,
 )
+from mirar.processors.skyportal import SNCOSMO_KEY
 
 wirc_filter_dict = {"J": 1, "H": 2, "Ks": 3}
 
@@ -28,6 +29,12 @@
 
 WIRC_NONLINEAR_LEVEL = 30000
 
+sncosmo_filters = {
+    "j": "2massj",
+    "h": "2massh",
+    "ks": "2massks",
+}
+
 
 def load_raw_wirc_fits(path: str | Path) -> tuple[np.array, astropy.io.fits.Header]:
     """
@@ -40,6 +47,9 @@ def load_raw_wirc_fits(path: str | Path) -> tuple[np.array, astropy.io.fits.Head
     if GAIN_KEY not in header.keys():
         header[GAIN_KEY] = 1.2
     header["FILTER"] = header["AFT"].split("__")[0]
+
+    header[SNCOSMO_KEY] = sncosmo_filters[header["FILTER"].lower()]
+
     if "COADDS" in header.keys():
         header["DETCOADD"] = header["COADDS"]
     if SATURATE_KEY not in header:
@@ -57,6 +67,9 @@ def load_raw_wirc_fits(path: str | Path) -> tuple[np.array, astropy.io.fits.Head
     else:
         header["DATE-OBS"] = header["UTSHUT"]
         header["MJD-OBS"] = Time(header["UTSHUT"]).mjd
+
+    header["JD"] = Time(header["DATE-OBS"]).jd
+
     if COADD_KEY not in header.keys():
         logger.debug(f"No {COADD_KEY} entry. Setting coadds to 1.")
         header[COADD_KEY] = 1
diff --git a/mirar/pipelines/wirc/wirc_files/avro_schema/wirc.alert.avsc b/mirar/pipelines/wirc/wirc_files/avro_schema/wirc.alert.avsc
index 582dcb09c..6614d3c40 100644
--- a/mirar/pipelines/wirc/wirc_files/avro_schema/wirc.alert.avsc
+++ b/mirar/pipelines/wirc/wirc_files/avro_schema/wirc.alert.avsc
@@ -12,8 +12,8 @@
         {"name": "prv_candidates", "type": [{
             "type": "array",
             "items": "wirc.alert.prv_candidate"}, "null"]},
-        {"name": "cutoutScience", "type": "bytes", "doc": "science cutout image [gzipped fits bytestream]"},
-        {"name": "cutoutTemplate", "type": "bytes", "doc": "cutout of template image [gzipped fits bytestream]"},
-        {"name": "cutoutDifference", "type": "bytes", "doc": "difference image cutout [gzipped fits bytestream]"}
+        {"name": "cutout_science", "type": "bytes", "doc": "science cutout image [gzipped fits bytestream]"},
+        {"name": "cutout_template", "type": "bytes", "doc": "cutout of template image [gzipped fits bytestream]"},
+        {"name": "cutout_difference", "type": "bytes", "doc": "difference image cutout [gzipped fits bytestream]"}
         ]
 }
diff --git a/mirar/pipelines/wirc/wirc_files/avro_schema/wirc.alert.candidate.avsc b/mirar/pipelines/wirc/wirc_files/avro_schema/wirc.alert.candidate.avsc
index e43b35953..460536f65 100644
--- a/mirar/pipelines/wirc/wirc_files/avro_schema/wirc.alert.candidate.avsc
+++ b/mirar/pipelines/wirc/wirc_files/avro_schema/wirc.alert.candidate.avsc
@@ -8,7 +8,6 @@
 		{"name": "diffmaglim", "type": ["float", "null"], "doc": "Expected 5-sigma mag limit in difference image based on global noise estimate [mag]"},
 		{"name": "fid", "type": "int", "doc": "Filter ID (0=Y; 1=J; 2=H; 3=K)"},
 		{"name": "programpi", "type": ["string", "null"], "doc": "Principal investigator attached to program ID"},
-		{"name": "programid", "type": "int", "doc": "Program ID: encodes either public, collab, or caltech mode"},
 		{"name": "candid", "type": "long", "doc": "Candidate ID from operations DB"},
 		{"name": "isdiffpos", "type": "boolean", "doc": "true => candidate is from positive (sci minus ref) subtraction; f or 0 => candidate is from negative (ref minus sci) subtraction"},
 		{"name": "field", "type": ["int", "null"], "doc": "ZTF field ID"}, 
diff --git a/mirar/pipelines/wirc/wirc_files/models.py b/mirar/pipelines/wirc/wirc_files/models.py
index 921271830..b8d3677cc 100644
--- a/mirar/pipelines/wirc/wirc_files/models.py
+++ b/mirar/pipelines/wirc/wirc_files/models.py
@@ -115,9 +115,9 @@ class Candidate(BaseDB):
 
     fid: int = Field(ge=0)
 
-    diffimgname: str | None = Field(max_length=255, default=None)
-    sciimgname: str | None = Field(max_length=255, default=None)
-    refimgname: str | None = Field(max_length=255, default=None)
+    diffimgname: str = Field(max_length=255)
+    sciimgname: str = Field(max_length=255)
+    refimgname: str = Field(max_length=255)
 
     magpsf: float = Field()
     sigmapsf: float = Field(ge=0)
diff --git a/mirar/processors/astromatic/swarp/component_images.py b/mirar/processors/astromatic/swarp/component_images.py
index cda06375e..343ea4b60 100644
--- a/mirar/processors/astromatic/swarp/component_images.py
+++ b/mirar/processors/astromatic/swarp/component_images.py
@@ -12,7 +12,7 @@
 from mirar.io import open_fits
 from mirar.paths import STACKED_COMPONENT_IMAGES_KEY
 from mirar.processors.astromatic.swarp.swarp import Swarp
-from mirar.processors.base_processor import BaseImageProcessor
+from mirar.processors.base_processor import BaseImageProcessor, PrerequisiteError
 from mirar.processors.utils.image_saver import ImageSaver
 
 logger = logging.getLogger(__name__)
@@ -79,7 +79,7 @@ def check_prerequisites(
                 f"However, the following steps were found: {self.preceding_steps}."
             )
             logger.error(err)
-            raise ValueError(err)
+            raise PrerequisiteError(err)
 
         index = np.argmax(mask)
 
@@ -92,4 +92,4 @@ def check_prerequisites(
                 f"However, the following steps were found: {self.preceding_steps}."
             )
             logger.error(err)
-            raise ValueError(err)
+            raise PrerequisiteError(err)
diff --git a/mirar/processors/avro/__init__.py b/mirar/processors/avro/__init__.py
index 72ada667c..10343c1b9 100644
--- a/mirar/processors/avro/__init__.py
+++ b/mirar/processors/avro/__init__.py
@@ -2,4 +2,3 @@
 Module for producing, and broadcasting, avro alerts
 """
 from mirar.processors.avro.ipac_avro_exporter import IPACAvroExporter
-from mirar.processors.avro.send_to_fritz import SendToFritz
diff --git a/mirar/processors/avro/ipac_avro_exporter.py b/mirar/processors/avro/ipac_avro_exporter.py
index d776507b8..b63849033 100644
--- a/mirar/processors/avro/ipac_avro_exporter.py
+++ b/mirar/processors/avro/ipac_avro_exporter.py
@@ -68,8 +68,12 @@ def fill_schema(schema: Schema, row: pd.Series, metadata: dict) -> dict:
             key = field["name"]
             if key in row.keys():
                 new[key] = row[key]
+            elif key.upper() in row.keys():
+                new[key] = row[key.upper()]
             elif key in metadata.keys():
                 new[key] = metadata[key]
+            elif key.upper() in metadata.keys():
+                new[key] = metadata[key.upper()]
 
         return new
 
diff --git a/mirar/processors/avro/send_to_fritz.py b/mirar/processors/avro/send_to_fritz.py
deleted file mode 100644
index 5f4a14054..000000000
--- a/mirar/processors/avro/send_to_fritz.py
+++ /dev/null
@@ -1,830 +0,0 @@
-"""
-Module for sending candidates to Fritz.
-"""
-import base64
-import gzip
-import io
-import logging
-import os
-import time
-from copy import deepcopy
-from datetime import datetime
-from typing import Mapping, Optional
-
-import matplotlib
-import matplotlib.pyplot as plt
-import numpy as np
-import pandas as pd
-import requests
-from astropy.io import fits
-from astropy.stats import sigma_clipped_stats
-from astropy.time import Time
-from astropy.visualization import (
-    AsymmetricPercentileInterval,
-    ImageNormalize,
-    LinearStretch,
-    LogStretch,
-)
-from requests.adapters import HTTPAdapter
-from urllib3.util import Retry
-
-from mirar.data import SourceBatch
-from mirar.paths import CAND_NAME_KEY, PACKAGE_NAME, __version__
-from mirar.processors.base_processor import BaseSourceProcessor
-
-matplotlib.use("agg")
-
-logger = logging.getLogger(__name__)
-
-DEFAULT_TIMEOUT = 5  # seconds
-
-
-class TimeoutHTTPAdapter(HTTPAdapter):
-    """
-    HTTP adapter that sets a default timeout for all requests.
-    """
-
-    def __init__(self, *args, **kwargs):
-        self.timeout = DEFAULT_TIMEOUT
-        if "timeout" in kwargs:
-            self.timeout = kwargs["timeout"]
-            del kwargs["timeout"]
-        super().__init__(*args, **kwargs)
-
-    def send(self, request, *args, **kwargs):
-        """
-        Send a request with a default timeout.
-
-        :param request: request to send
-        :param args: args to pass to super().send
-        :param kwargs: kwargs to pass to super().send
-        :return: response from request
-        """
-        try:
-            timeout = kwargs.get("timeout")
-            if timeout is None:
-                kwargs["timeout"] = self.timeout
-            return super().send(request, *args, **kwargs)
-        except AttributeError:
-            kwargs["timeout"] = DEFAULT_TIMEOUT
-
-
-class SendToFritz(BaseSourceProcessor):
-    """
-    Processor for sending candidates to Fritz.
-    """
-
-    base_key = "fritzsender"
-
-    def __init__(
-        self,
-        base_name: str,
-        group_ids: list[int],
-        fritz_filter_id: int,
-        instrument_id: int,
-        stream_id: int,
-        update_thumbnails: bool = True,
-        protocol: str = "http",
-    ):
-        super().__init__()
-        self.token = None
-        self.group_ids = group_ids
-        self.fritz_filter_id = fritz_filter_id
-        self.instrument_id = instrument_id
-        self.origin = base_name  # used for sending updates to Fritz
-        self.stream_id = stream_id
-        self.protocol = protocol
-        self.update_thumbnails = update_thumbnails
-
-        self._session = None
-        self.session_headers = None
-
-    def set_up_session(self):
-        """
-        Set up a session for sending requests to Fritz.
-
-        :return: None
-        """
-        # session to talk to SkyPortal/Fritz
-        self._session = requests.Session()
-        self.session_headers = {
-            "Authorization": f"token {self._get_fritz_token()}",
-            "User-Agent": "mirar",
-        }
-
-        retries = Retry(
-            total=5,
-            backoff_factor=2,
-            status_forcelist=[405, 429, 500, 502, 503, 504],
-            method_whitelist=["HEAD", "GET", "PUT", "POST", "PATCH"],
-        )
-        adapter = TimeoutHTTPAdapter(timeout=5, max_retries=retries)
-        self._session.mount("https://", adapter)
-        self._session.mount("http://", adapter)
-
-    def get_session(self) -> requests.Session:
-        """
-        Wrapper for getting the session.
-        If the session is not set up, it will be set up.
-
-        :return: Session
-        """
-        if self._session is None:
-            self.set_up_session()
-
-        return self._session
-
-    @staticmethod
-    def _get_fritz_token():
-        """
-        Get Fritz token from environment variable.
-
-        :return: Fritz token
-        """
-        token_fritz = os.getenv("FRITZ_TOKEN")
-
-        if token_fritz is None:
-            err = (
-                "No Fritz token specified. Run 'export FRITZ_TOKEN=<token>' to "
-                "set. The Fritz token will need to be specified manually "
-                "for Fritz API queries."
-            )
-            logger.error(err)
-            raise ValueError(err)
-
-        return token_fritz
-
-    def _apply_to_sources(
-        self,
-        batch: SourceBatch,
-    ) -> SourceBatch:
-        """
-        Apply the processor to a batch of candidates.
-
-        :param batch: SourceBatch to process
-        :return: SourceBatch after processing
-        """
-        for source_table in batch:
-            candidate_table = source_table.get_data()
-            self.make_alert(candidate_table)
-        return batch
-
-    @staticmethod
-    def _get_author_id():
-        """
-        Fritz author id is used in update calls. Can be found
-
-        :return: Fritz author id
-        """
-        authid_fritz = os.getenv("FRITZ_AUTHID")
-
-        if authid_fritz is None:
-            err = (
-                "No Fritz author id specified. Run 'export FRITZ_AUTHID=<id>' to set. "
-                "Author id needs to be specified for updates sent by Fritz API queries."
-            )
-            logger.error(err)
-            raise ValueError(err)
-
-        return authid_fritz
-
-    @staticmethod
-    def read_input_df(candidate_df: pd.DataFrame):
-        """Takes a DataFrame, which has multiple candidate
-        and creates list of dictionaries, each dictionary
-        representing a single candidate.
-
-        Args:
-            candidate_df (pandas.core.frame.DataFrame): dataframe of all candidates.
-
-        Returns:
-            (list[dict]): list of dictionaries, each a candidate.
-        """
-        all_candidates = []
-
-        for i in range(0, len(candidate_df)):
-            candidate = {}
-            for key in candidate_df.keys():
-                try:
-                    if isinstance(candidate_df.iloc[i].get(key), (list, str)):
-                        candidate[key] = candidate_df.iloc[i].get(key)
-                    else:
-                        # change to native python type
-                        candidate[key] = candidate_df.iloc[i].get(key).item()
-                except AttributeError:  # for IOBytes objs
-                    candidate[key] = candidate_df.iloc[i].get(key)
-
-            all_candidates.append(candidate)
-
-        return all_candidates
-
-    def api(
-        self, method: str, endpoint: str, data: Optional[Mapping] = None
-    ) -> requests.Response:
-        """Make an API call to a SkyPortal instance
-
-        headers = {'Authorization': f'token {self.token}'}
-        response = requests.request(method, endpoint, json_dict=data, headers=headers)
-
-        :param method: HTTP method
-        :param endpoint: API endpoint
-        :param data: JSON data to send
-        :return: response from API call
-        """
-        method = method.lower()
-
-        session = self.get_session()
-
-        methods = {
-            "head": session.head,
-            "get": session.get,
-            "post": session.post,
-            "put": session.put,
-            "patch": session.patch,
-            "delete": session.delete,
-        }
-
-        if endpoint is None:
-            raise ValueError("Endpoint not specified")
-        if method not in ["head", "get", "post", "put", "patch", "delete"]:
-            raise ValueError(f"Unsupported method: {method}")
-
-        if method == "get":
-            response = methods[method](
-                f"{endpoint}",
-                params=data,
-                headers=self.session_headers,
-            )
-        else:
-            response = methods[method](
-                f"{endpoint}",
-                json=data,
-                headers=self.session_headers,
-            )
-
-        return response
-
-    def alert_post_source(self, alert: dict, group_ids: Optional[list[int]] = None):
-        """Add a new source to SkyPortal
-
-        :param alert: dict of source info
-        :param group_ids: list of group_ids to post source to, defaults to None
-        :return: None
-        """
-        if group_ids is None:
-            group_ids = self.group_ids
-
-        data = {
-            "ra": alert["ra"],
-            "dec": alert["dec"],
-            "id": alert[CAND_NAME_KEY],
-            "group_ids": group_ids,
-            "origin": self.origin,
-        }
-
-        logger.debug(
-            f"Saving {alert[CAND_NAME_KEY]} {alert['candid']} as a Source on SkyPortal"
-        )
-        response = self.api("POST", "https://fritz.science/api/sources", data)
-
-        if response.json()["status"] == "success":
-            logger.debug(
-                f"Saved {alert[CAND_NAME_KEY]} {alert['candid']} "
-                f"as a Source on SkyPortal"
-            )
-        else:
-            err = (
-                f"Failed to save {alert[CAND_NAME_KEY]} {alert['candid']} "
-                f"as a Source on SkyPortal"
-            )
-            logger.error(err)
-            logger.error(response.json())
-
-    def alert_post_candidate(self, alert):
-        """
-        Post a candidate on SkyPortal. Creates new candidate(s) (one per filter)
-
-        :param alert: dict of alert info
-        :return: None
-        """
-
-        data = {
-            "id": alert[CAND_NAME_KEY],
-            "ra": alert["ra"],
-            "dec": alert["dec"],
-            "filter_ids": [self.fritz_filter_id],
-            "passing_alert_id": self.fritz_filter_id,
-            "passed_at": Time(datetime.utcnow()).isot,
-            "origin": f"{PACKAGE_NAME}:{__version__}",
-        }
-
-        logger.debug(
-            f"Posting metadata of {alert[CAND_NAME_KEY]} {alert['candid']} to SkyPortal"
-        )
-        response = self.api("POST", "https://fritz.science/api/candidates", data)
-
-        if response.json()["status"] == "success":
-            logger.debug(
-                f"Posted {alert[CAND_NAME_KEY]} {alert['candid']} metadata to SkyPortal"
-            )
-        else:
-            logger.error(
-                f"Failed to post {alert[CAND_NAME_KEY]} {alert['candid']} "
-                f"metadata to SkyPortal"
-            )
-            logger.error(response.json())
-
-    def make_thumbnail(self, alert, skyportal_type: str, alert_packet_type: str):
-        """
-        Convert lossless FITS cutouts from ZTF-like alerts into PNGs.
-        Make thumbnail for pushing to SkyPortal.
-
-        :param alert: ZTF-like alert packet/dict
-        :param skyportal_type: <new|ref|sub> thumbnail type expected by SkyPortal
-        :param alert_packet_type: <Science|Template|Difference> survey naming
-        :return:
-        """
-        alert = deepcopy(alert)
-        cutout_data = alert[f"cutout{alert_packet_type}"]
-
-        with gzip.open(io.BytesIO(cutout_data), "rb") as cutout:
-            with fits.open(
-                io.BytesIO(cutout.read()), ignore_missing_simple=True
-            ) as hdu:
-                image_data = hdu[0].data  # pylint: disable=no-member
-
-        buff = io.BytesIO()
-        plt.close("all")
-        fig = plt.figure()
-        fig.set_size_inches(4, 4, forward=False)
-        ax_1 = plt.Axes(fig, [0.0, 0.0, 1.0, 1.0])
-        ax_1.set_axis_off()
-        fig.add_axes(ax_1)
-
-        # replace nans with median:
-        img = np.array(image_data)
-        # replace dubiously large values
-        xl_mask = np.greater(np.abs(img), 1e20, where=~np.isnan(img))
-        if img[xl_mask].any():
-            img[xl_mask] = np.nan
-        if np.isnan(img).any():
-            median = float(np.nanmean(img.flatten()))
-            img = np.nan_to_num(img, nan=median)
-
-        norm = ImageNormalize(
-            img,
-            stretch=LinearStretch()
-            if alert_packet_type == "Difference"
-            else LogStretch(),
-        )
-        img_norm = norm(img)
-        normalizer = AsymmetricPercentileInterval(
-            lower_percentile=1, upper_percentile=100
-        )
-        vmin, vmax = normalizer.get_limits(img_norm)
-        ax_1.imshow(img_norm, cmap="bone", origin="lower", vmin=vmin, vmax=vmax)
-        plt.savefig(buff, dpi=42)
-
-        buff.seek(0)
-        plt.close("all")
-
-        thumbnail_dict = {
-            "obj_id": alert[CAND_NAME_KEY],
-            "data": base64.b64encode(buff.read()).decode("utf-8"),
-            "ttype": skyportal_type,
-        }
-
-        return thumbnail_dict
-
-    def alert_post_thumbnails(self, alert):
-        """Post alert Science, Reference, and Subtraction thumbnails to SkyPortal
-
-        :param alert: dict of source/candidate information
-        :return:
-        """
-        for ttype, instrument_type in [
-            ("new", "Science"),
-            ("ref", "Template"),
-            ("sub", "Difference"),
-        ]:
-            logger.debug(
-                f"Making {instrument_type} thumbnail for {alert[CAND_NAME_KEY]} "
-                f"{alert['candid']}",
-            )
-            thumb = self.make_thumbnail(alert, ttype, instrument_type)
-
-            logger.debug(
-                f"Posting {instrument_type} thumbnail for {alert[CAND_NAME_KEY]} "
-                f"{alert['candid']} to SkyPortal",
-            )
-            response = self.api("POST", "https://fritz.science/api/thumbnail", thumb)
-
-            if response.json()["status"] == "success":
-                logger.debug(
-                    f"Posted {alert[CAND_NAME_KEY]} {alert['candid']} "
-                    f"{instrument_type} cutout to SkyPortal"
-                )
-            else:
-                logger.error(
-                    f"Failed to post {alert[CAND_NAME_KEY]} {alert['candid']} "
-                    f"{instrument_type} cutout to SkyPortal"
-                )
-                logger.error(response.json())
-
-    def upload_thumbnail(self, alert):
-        """Post new thumbnail to Fritz.
-
-        NOTE: this is the original WINTER method for sending thumbnails,
-        not full sized but higher contrast, similar to alert_make_thumbnail
-
-        Format of thumbnail payload:
-        { "obj_id": "string",  "data": "string",  "ttype": "string"}
-        """
-        fritz_to_cand = {"new": "SciBitIm", "ref": "RefBitIm", "sub": "DiffBitIm"}
-
-        for fritz_key, cand_key in fritz_to_cand.items():
-            cutout = alert[cand_key]
-
-            buffer = io.BytesIO()
-            plt.figure(figsize=(3, 3))
-            mean, median, std = sigma_clipped_stats(cutout)
-            plt.imshow(
-                cutout,
-                origin="lower",
-                cmap="gray",
-                vmin=mean - 1 * std,
-                vmax=median + 3 * std,
-            )
-            plt.xticks([])
-            plt.yticks([])
-
-            plt.savefig(buffer, format="png")
-
-            cutoutb64 = base64.b64encode(buffer.getvalue())
-            cutoutb64_string = cutoutb64.decode("utf8")
-
-            data_payload = {
-                "obj_id": alert[CAND_NAME_KEY],
-                "data": cutoutb64_string,
-                "ttype": fritz_key,
-            }
-
-            response = self.api(
-                "POST", "https://fritz.science/api/thumbnail", data=data_payload
-            )
-
-            if response.json()["status"] == "success":
-                logger.debug(
-                    f"Posted {alert[CAND_NAME_KEY]} {alert['candid']} "
-                    f"{cand_key} cutout to SkyPortal"
-                )
-            else:
-                logger.error(
-                    f"Failed to post {alert[CAND_NAME_KEY]} {alert['candid']} "
-                    f"{cand_key} cutout to SkyPortal"
-                )
-                logger.error(response.json())
-
-    def make_photometry(self, alert, jd_start: Optional[float] = None):
-        """
-        Make a de-duplicated pandas.DataFrame with photometry of alert[CAND_NAME_KEY]
-        Modified from Kowalksi (https://github.com/dmitryduev/kowalski)
-
-        :param alert: candidate dictionary
-        :param jd_start: date from which to start photometry from
-        """
-        alert = deepcopy(alert)
-        top_level = [
-            "schemavsn",
-            "publisher",
-            CAND_NAME_KEY,
-            "candid",
-            "candidate",
-            "prv_candidates",
-            "cutoutScience",
-            "cutoutTemplate",
-            "cutoutDifference",
-        ]
-        alert["candidate"] = {}
-
-        # (keys having value in 3.)
-        delete = [key for key in alert.keys() if key not in top_level]
-
-        # delete the key/s
-        for key in delete:
-            alert["candidate"][key] = alert[key]
-            del alert[key]
-
-        alert["candidate"] = [alert["candidate"]]
-        df_candidate = pd.DataFrame(alert["candidate"], index=[0])
-
-        df_prv_candidates = pd.DataFrame(alert["prv_candidates"])
-
-        df_light_curve = pd.concat(
-            [df_candidate, df_prv_candidates], ignore_index=True, sort=False
-        )
-
-        # note: WNTR (like PGIR) uses 2massj, which is not in sncosmo as of
-        # 20210803, cspjs seems to be close/good enough as an approximation
-        df_light_curve["filter"] = "cspjs"
-
-        df_light_curve["magsys"] = "ab"
-        df_light_curve["mjd"] = df_light_curve["jd"] - 2400000.5
-
-        df_light_curve["mjd"] = df_light_curve["mjd"].astype(np.float64)
-        df_light_curve["magpsf"] = df_light_curve["magpsf"].astype(np.float32)
-        df_light_curve["sigmapsf"] = df_light_curve["sigmapsf"].astype(np.float32)
-
-        df_light_curve = (
-            df_light_curve.drop_duplicates(subset=["mjd", "magpsf"])
-            .reset_index(drop=True)
-            .sort_values(by=["mjd"])
-        )
-
-        # filter out bad data:
-        mask_good_diffmaglim = df_light_curve["diffmaglim"] > 0
-        df_light_curve = df_light_curve.loc[mask_good_diffmaglim]
-
-        # convert from mag to flux
-
-        # step 1: calculate the coefficient that determines whether the
-        # flux should be negative or positive
-        coeff = df_light_curve["isdiffpos"].apply(
-            lambda x: 1.0 if x in [True, 1, "y", "Y", "t", "1"] else -1.0
-        )
-
-        # step 2: calculate the flux normalized to an arbitrary AB zeropoint of
-        # 23.9 (results in flux in uJy)
-        df_light_curve["flux"] = coeff * 10 ** (
-            -0.4 * (df_light_curve["magpsf"] - 23.9)
-        )
-
-        # step 3: separate detections from non detections
-        detected = np.isfinite(df_light_curve["magpsf"])
-        undetected = ~detected
-
-        # step 4: calculate the flux error
-        df_light_curve["fluxerr"] = None  # initialize the column
-
-        # step 4a: calculate fluxerr for detections using sigmapsf
-        df_light_curve.loc[detected, "fluxerr"] = np.abs(
-            df_light_curve.loc[detected, "sigmapsf"]
-            * df_light_curve.loc[detected, "flux"]
-            * np.log(10)
-            / 2.5
-        )
-
-        # step 4b: calculate fluxerr for non detections using diffmaglim
-        df_light_curve.loc[undetected, "fluxerr"] = (
-            10 ** (-0.4 * (df_light_curve.loc[undetected, "diffmaglim"] - 23.9)) / 5.0
-        )  # as diffmaglim is the 5-sigma depth
-
-        # step 5: set the zeropoint and magnitude system
-        df_light_curve["zp"] = 23.9
-        df_light_curve["zpsys"] = "ab"
-
-        # only "new" photometry requested?
-        if jd_start is not None:
-            w_after_jd = df_light_curve["jd"] > jd_start
-            df_light_curve = df_light_curve.loc[w_after_jd]
-
-        return df_light_curve
-
-    def alert_put_photometry(self, alert):
-        """Send photometry to Fritz."""
-        logger.debug(
-            f"Making alert photometry of {alert[CAND_NAME_KEY]} {alert['candid']}"
-        )
-        df_photometry = self.make_photometry(alert)
-
-        # post photometry
-        photometry = {
-            "obj_id": alert[CAND_NAME_KEY],
-            "stream_ids": [int(self.stream_id)],
-            "instrument_id": self.instrument_id,
-            "mjd": df_photometry["mjd"].tolist(),
-            "flux": df_photometry["flux"].tolist(),
-            "fluxerr": df_photometry["fluxerr"].tolist(),
-            "zp": df_photometry["zp"].tolist(),
-            "magsys": df_photometry["zpsys"].tolist(),
-            "filter": df_photometry["filter"].tolist(),
-            "ra": df_photometry["ra"].tolist(),
-            "dec": df_photometry["dec"].tolist(),
-        }
-
-        if (len(photometry.get("flux", ())) > 0) or (
-            len(photometry.get("fluxerr", ())) > 0
-        ):
-            logger.debug(
-                f"Posting photometry of {alert[CAND_NAME_KEY]} {alert['candid']}, "
-                f"stream_id={self.stream_id} to SkyPortal"
-            )
-            response = self.api(
-                "PUT", "https://fritz.science/api/photometry", photometry
-            )
-            if response.json()["status"] == "success":
-                logger.debug(
-                    f"Posted {alert[CAND_NAME_KEY]} photometry stream_id={self.stream_id} "
-                    f"to SkyPortal"
-                )
-            else:
-                logger.error(
-                    f"Failed to post {alert[CAND_NAME_KEY]} photometry "
-                    f"stream_id={self.stream_id} to SkyPortal"
-                )
-                logger.error(response.json())
-
-    def alert_post_annotation(self, alert):
-        """Post an annotation. Works for both candidates and sources."""
-        data = {
-            "chipsf": alert["chipsf"],
-            "fwhm": alert["fwhm"],
-            "scorr": alert["scorr"],
-        }
-        payload = {"origin": self.origin, "data": data, "group_ids": self.group_ids}
-
-        path = (
-            f"https://fritz.science/api/sources/{str(alert[CAND_NAME_KEY])}/annotations"
-        )
-        response = self.api("POST", path, payload)
-
-        if response.json()["status"] == "success":
-            logger.debug(f"Posted {alert[CAND_NAME_KEY]} annotation to SkyPortal")
-        else:
-            logger.error(
-                f"Failed to post {alert[CAND_NAME_KEY]} annotation to SkyPortal"
-            )
-            logger.error(response.json())
-
-    def alert_put_annotation(self, alert):
-        """Retrieve an annotation to check if it exists already."""
-        response = self.api(
-            "GET",
-            f"https://fritz.science/api/sources/"
-            f"{str(alert[CAND_NAME_KEY])}/annotations",
-        )
-
-        if response.json()["status"] == "success":
-            logger.debug(f"Got {alert[CAND_NAME_KEY]} annotations from SkyPortal")
-        else:
-            logger.debug(
-                f"Failed to get {alert[CAND_NAME_KEY]} annotations from SkyPortal"
-            )
-            logger.debug(response.json())
-            return False
-
-        existing_annotations = {
-            annotation["origin"]: {
-                "annotation_id": annotation["id"],
-                "author_id": annotation["author_id"],
-            }
-            for annotation in response.json()["data"]
-        }
-
-        # no previous annotation exists on SkyPortal for this object? just post then
-        if self.origin not in existing_annotations:
-            self.alert_post_annotation(alert)
-        # annotation from this(WNTR) origin exists
-        else:
-            # annotation data
-            data = {
-                "fwhm": alert["fwhm"],
-                "scorr": alert["scorr"],
-                "chipsf": alert["chipsf"],
-            }
-            new_annotation = {
-                "author_id": existing_annotations[self.origin]["author_id"],
-                "obj_id": alert[CAND_NAME_KEY],
-                "origin": self.origin,
-                "data": data,
-                "group_ids": self.group_ids,
-            }
-
-            logger.debug(
-                f"Putting annotation for {alert[CAND_NAME_KEY]} {alert['candid']} "
-                f"to SkyPortal",
-            )
-            response = self.api(
-                "PUT",
-                f"https://fritz.science/api/sources/{alert[CAND_NAME_KEY]}"
-                f"/annotations/{existing_annotations[self.origin]['annotation_id']}",
-                new_annotation,
-            )
-            if response.json()["status"] == "success":
-                logger.debug(
-                    f"Posted updated {alert[CAND_NAME_KEY]} annotation to SkyPortal"
-                )
-            else:
-                logger.error(
-                    f"Failed to post updated {alert[CAND_NAME_KEY]} annotation "
-                    f"to SkyPortal"
-                )
-                logger.error(response.json())
-
-        return None
-
-    def alert_skyportal_manager(self, alert):
-        """Posts alerts to SkyPortal if criteria is met
-
-        :param alert: _description_
-        :type alert: _type_
-        """
-        # check if candidate exists in SkyPortal
-        logger.debug(f"Checking if {alert[CAND_NAME_KEY]} is candidate in SkyPortal")
-        response = self.api(
-            "HEAD", f"https://fritz.science/api/candidates/{alert[CAND_NAME_KEY]}"
-        )
-        is_candidate = response.status_code == 200
-        logger.debug(
-            f"{alert[CAND_NAME_KEY]} {'is' if is_candidate else 'is not'} "
-            f"candidate in SkyPortal"
-        )
-
-        # check if source exists in SkyPortal
-        logger.debug(f"Checking if {alert[CAND_NAME_KEY]} is source in SkyPortal")
-        response = self.api(
-            "HEAD", f"https://fritz.science/api/sources/{alert[CAND_NAME_KEY]}"
-        )
-        is_source = response.status_code == 200
-        logger.debug(
-            f"{alert[CAND_NAME_KEY]} "
-            f"{'is' if is_source else 'is not'} source in SkyPortal"
-        )
-
-        # object does not exist in SkyPortal: neither cand nor source
-        if (not is_candidate) and (not is_source):
-            # post candidate
-            self.alert_post_candidate(alert)
-
-            # post annotations
-            self.alert_post_annotation(alert)
-
-            # post full light curve
-            self.alert_put_photometry(alert)
-
-            # post thumbnails
-            self.alert_post_thumbnails(alert)
-
-            # TODO autosave stuff, necessary?
-
-        # obj already exists in SkyPortal
-        else:
-            # TODO passed_filters logic
-
-            # post candidate with new filter ids
-            self.alert_post_candidate(alert)
-
-            # put (*not* post) annotations
-            self.alert_put_annotation(alert)
-
-            # exists in SkyPortal & already saved as a source
-            if is_source:
-                # get info on the corresponding groups:
-                logger.debug(
-                    f"Getting source groups info on "
-                    f"{alert[CAND_NAME_KEY]} from SkyPortal",
-                )
-                response = self.api(
-                    "GET",
-                    f"https://fritz.science/api/sources/{alert[CAND_NAME_KEY]}/groups",
-                )
-                if response.json()["status"] == "success":
-                    existing_groups = response.json()["data"]
-                    existing_group_ids = [g["id"] for g in existing_groups]
-
-                    for existing_gid in existing_group_ids:
-                        if existing_gid in self.group_ids:
-                            self.alert_post_source(alert, [existing_gid])
-                else:
-                    logger.error(
-                        f"Failed to get source groups info on {alert[CAND_NAME_KEY]}"
-                    )
-            else:  # exists in SkyPortal but NOT saved as a source
-                self.alert_post_source(alert)
-
-            # post alert photometry in single call to /api/photometry
-            self.alert_put_photometry(alert)
-
-            if self.update_thumbnails:
-                self.alert_post_thumbnails(alert)
-
-        logger.debug(f"SendToFritz Manager complete for {alert[CAND_NAME_KEY]}")
-
-    def make_alert(self, candidate_df: pd.DataFrame):
-        """
-        Function to make an alert from a single row of a pandas DataFrame
-
-        :param candidate_df: Candidate DataFrame
-        :return: None
-        """
-        t_0 = time.time()
-        all_cands = self.read_input_df(candidate_df)
-        num_cands = len(all_cands)
-
-        for cand in all_cands:
-            self.alert_skyportal_manager(cand)
-
-        t_1 = time.time()
-        logger.info(
-            f"Took {(t_1 - t_0):.2f} seconds to Fritz process {num_cands} candidates."
-        )
diff --git a/mirar/processors/base_processor.py b/mirar/processors/base_processor.py
index cab6ae868..561309701 100644
--- a/mirar/processors/base_processor.py
+++ b/mirar/processors/base_processor.py
@@ -13,6 +13,7 @@
 from threading import Thread
 
 import numpy as np
+import pandas as pd
 from tqdm.auto import tqdm
 
 from mirar.data import DataBatch, Dataset, Image, ImageBatch, SourceBatch
@@ -471,6 +472,12 @@ def __init__(self, master_image_path: str | Path, *args, **kwargs):
         self.master_image_path = Path(master_image_path)
 
     def get_cache_path(self, images: ImageBatch) -> Path:
+        """
+        Gets path for saving/loading cached image
+
+        :param images: Images to process
+        :return: Path to cached image
+        """
         return self.master_image_path
 
 
@@ -496,6 +503,21 @@ def _apply(self, batch: ImageBatch) -> SourceBatch:
     def _apply_to_images(self, batch: ImageBatch) -> SourceBatch:
         raise NotImplementedError
 
+    def get_metadata(self, image: Image) -> dict:
+        """
+        Get metadata from image
+
+        :param image: Image to get metadata from
+        :return: Metadata dictionary
+        """
+        metadata = {}
+
+        for key in image.keys():
+            if key != "COMMENT":
+                metadata[key] = image[key]
+
+        return metadata
+
 
 class BaseSourceProcessor(BaseProcessor, ABC):
     """
@@ -515,3 +537,18 @@ def _apply_to_sources(
         batch: SourceBatch,
     ) -> SourceBatch:
         raise NotImplementedError
+
+    @staticmethod
+    def generate_super_dict(metadata: dict, source_row: pd.Series) -> dict:
+        """
+        Generate a dictionary of metadata and candidate row, with lower case keys
+
+        :param metadata: Metadata for the source table
+        :param source_row: Individual row of the source table
+        :return: Combined dictionary
+        """
+        super_dict = {key.lower(): val for key, val in metadata.items()}
+        super_dict.update(
+            {key.lower(): val for key, val in source_row.to_dict().items()}
+        )
+        return super_dict
diff --git a/mirar/processors/database/database_inserter.py b/mirar/processors/database/database_inserter.py
index 61032b295..01377b01e 100644
--- a/mirar/processors/database/database_inserter.py
+++ b/mirar/processors/database/database_inserter.py
@@ -9,7 +9,11 @@
 
 from mirar.data import Image, ImageBatch, SourceBatch
 from mirar.errors.exceptions import BaseProcessorError
-from mirar.processors.base_processor import BaseImageProcessor, BaseSourceProcessor
+from mirar.processors.base_processor import (
+    BaseImageProcessor,
+    BaseSourceProcessor,
+    PrerequisiteError,
+)
 from mirar.processors.database.base_database_processor import BaseDatabaseProcessor
 from mirar.processors.utils.image_selector import ImageBatcher
 
@@ -97,21 +101,6 @@ def _apply_to_sources(self, batch: SourceBatch) -> SourceBatch:
 
         return batch
 
-    @staticmethod
-    def generate_super_dict(metadata: dict, source_row: pd.Series) -> dict:
-        """
-        Generate a dictionary of metadata and candidate row, with lower case keys
-
-        :param metadata: Metadata for the source table
-        :param source_row: Individual row of the source table
-        :return: Combined dictionary
-        """
-        super_dict = {key.lower(): val for key, val in metadata.items()}
-        super_dict.update(
-            {key.lower(): val for key, val in source_row.to_dict().items()}
-        )
-        return super_dict
-
 
 class DatabaseImageBatchInserter(DatabaseImageInserter):
     """
@@ -170,4 +159,4 @@ def check_prerequisites(
                 f"However, the following steps were found: {self.preceding_steps}."
             )
             logger.error(err)
-            raise ValueError
+            raise PrerequisiteError(err)
diff --git a/mirar/processors/database/database_selector.py b/mirar/processors/database/database_selector.py
index 36bc423c2..40b80ccd3 100644
--- a/mirar/processors/database/database_selector.py
+++ b/mirar/processors/database/database_selector.py
@@ -32,7 +32,7 @@ def __init__(
         super().__init__(*args, **kwargs)
         self.boolean_match_key = boolean_match_key
 
-    def get_constraints(self, data: DataBlock) -> None | DBQueryConstraints:
+    def get_constraints(self, data: dict) -> None | DBQueryConstraints:
         """
         Get db query constraints for a given datablock object
 
@@ -107,7 +107,7 @@ def __init__(self, db_query_columns: str | list[str], *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.db_query_columns = db_query_columns
 
-    def get_accepted_values(self, data: DataBlock) -> list[str | float | int]:
+    def get_accepted_values(self, data: dict) -> list[str | float | int]:
         """
         Get list of accepted values for crossmatch query
 
@@ -117,7 +117,7 @@ def get_accepted_values(self, data: DataBlock) -> list[str | float | int]:
         accepted_values = [data[x.upper()] for x in self.db_query_columns]
         return accepted_values
 
-    def get_constraints(self, data: DataBlock) -> DBQueryConstraints:
+    def get_constraints(self, data: dict) -> DBQueryConstraints:
         """
         Get db query constraints for a datablock
 
@@ -169,10 +169,12 @@ def _apply_to_sources(
         batch: SourceBatch,
     ) -> SourceBatch:
         for source_table in batch:
+            metadata = source_table.get_metadata()
             candidate_table = source_table.get_data()
             results = []
-            for _, cand in candidate_table.iterrows():
-                query_constraints = self.get_constraints(cand)
+            for _, source in candidate_table.iterrows():
+                super_dict = self.generate_super_dict(metadata, source)
+                query_constraints = self.get_constraints(super_dict)
                 logger.debug(
                     f"Query constraints: " f"{query_constraints.parse_constraints()}"
                 )
diff --git a/mirar/processors/photometry/__init__.py b/mirar/processors/photometry/__init__.py
new file mode 100644
index 000000000..2370e5686
--- /dev/null
+++ b/mirar/processors/photometry/__init__.py
@@ -0,0 +1,2 @@
+from mirar.processors.photometry.aperture_photometry import AperturePhotometry
+from mirar.processors.photometry.psf_photometry import PSFPhotometry
diff --git a/mirar/processors/photometry/aperture_photometry.py b/mirar/processors/photometry/aperture_photometry.py
index ae0185beb..2e9438571 100644
--- a/mirar/processors/photometry/aperture_photometry.py
+++ b/mirar/processors/photometry/aperture_photometry.py
@@ -1,31 +1,25 @@
 """
 Module with processors to perform aperture photometry
 """
-from typing import Optional
-
 import numpy as np
 
-from mirar.data import ImageBatch, SourceBatch
+from mirar.data import SourceBatch
 from mirar.paths import (
     APFLUX_PREFIX_KEY,
     APFLUXUNC_PREFIX_KEY,
     APMAG_PREFIX_KEY,
     APMAGUNC_PREFIX_KEY,
 )
-from mirar.processors.photometry.base_photometry import (
-    AperturePhotometry,
-    BaseImagePhotometry,
-    BaseSourcePhotometry,
-)
-from mirar.processors.photometry.utils import get_mags_from_fluxes
+from mirar.processors.photometry.base_photometry import BasePhotometryProcessor
+from mirar.processors.photometry.utils import aper_photometry, get_mags_from_fluxes
 
 
-class SourceAperturePhotometry(BaseSourcePhotometry):
+class AperturePhotometry(BasePhotometryProcessor):
     """
     Processor to run aperture photometry on all candidates in candidate table
     """
 
-    base_key = "APERPHOTDF"
+    base_key = "APERPHOT"
 
     def __init__(
         self,
@@ -38,14 +32,36 @@ def __init__(
     ):
         super().__init__(*args, **kwargs)
 
-        self.aperture_photometer = AperturePhotometry(
-            aper_diameters=aper_diameters,
-            bkg_in_diameters=bkg_in_diameters,
-            bkg_out_diameters=bkg_out_diameters,
-        )
+        if not isinstance(aper_diameters, list):
+            aper_diameters = [aper_diameters]
+        if not isinstance(bkg_in_diameters, list):
+            bkg_in_diameters = [bkg_in_diameters]
+        if not isinstance(bkg_out_diameters, list):
+            bkg_out_diameters = [bkg_out_diameters]
+
+        self.aper_diameters = aper_diameters
+        self.bkg_in_diameters = bkg_in_diameters
+        self.bkg_out_diameters = bkg_out_diameters
+
         self.col_suffix_list = col_suffix_list
         if self.col_suffix_list is None:
-            self.col_suffix_list = self.aperture_photometer.aper_diameters
+            self.col_suffix_list = self.aper_diameters
+
+    def perform_photometry(
+        self, image_cutout: np.array, unc_image_cutout: np.array
+    ) -> tuple[list[float], list[float]]:
+        fluxes, fluxuncs = [], []
+        for ind, aper_diam in enumerate(self.aper_diameters):
+            flux, fluxunc = aper_photometry(
+                image_cutout,
+                unc_image_cutout,
+                aper_diam,
+                self.bkg_in_diameters[ind],
+                self.bkg_out_diameters[ind],
+            )
+            fluxes.append(flux)
+            fluxuncs.append(fluxunc)
+        return fluxes, fluxuncs
 
     def _apply_to_sources(
         self,
@@ -53,17 +69,22 @@ def _apply_to_sources(
     ) -> SourceBatch:
         for source_table in batch:
             candidate_table = source_table.get_data()
+
+            metadata = source_table.get_metadata()
+
             all_fluxes, all_fluxuncs = [], []
+
             for cand_ind in range(len(candidate_table)):
                 row = candidate_table.iloc[cand_ind]
 
-                image_cutout, unc_image_cutout = self.generate_cutouts(row)
+                image_cutout, unc_image_cutout = self.generate_cutouts(row, metadata)
 
-                fluxes, fluxuncs = self.aperture_photometer.perform_photometry(
+                fluxes, fluxuncs = self.perform_photometry(
                     image_cutout=image_cutout, unc_image_cutout=unc_image_cutout
                 )
                 all_fluxes.append(fluxes)
                 all_fluxuncs.append(fluxuncs)
+
             all_fluxes = np.array(all_fluxes).T
             all_fluxuncs = np.array(all_fluxuncs).T
 
@@ -73,12 +94,10 @@ def _apply_to_sources(
                 candidate_table[f"{APFLUXUNC_PREFIX_KEY}{suffix}"] = fluxunc
 
                 magnitudes, magnitudes_unc = get_mags_from_fluxes(
-                    flux_list=flux,
-                    fluxunc_list=fluxunc,
-                    zeropoint_list=np.array(candidate_table[self.zp_key], dtype=float),
-                    zeropoint_unc_list=np.array(
-                        candidate_table[f"{self.zp_std_key}"], dtype=float
-                    ),
+                    flux_list=np.array(flux, dtype=float),
+                    fluxunc_list=np.array(fluxunc, dtype=float),
+                    zeropoint=float(source_table[self.zp_key]),
+                    zeropoint_unc=float(source_table[self.zp_std_key]),
                 )
                 candidate_table[f"{APMAG_PREFIX_KEY}{suffix}"] = magnitudes
                 candidate_table[f"{APMAGUNC_PREFIX_KEY}{suffix}"] = magnitudes_unc
@@ -86,60 +105,3 @@ def _apply_to_sources(
             source_table.set_data(candidate_table)
 
         return batch
-
-
-class ImageAperturePhotometry(BaseImagePhotometry):
-    """
-    Processor to run aperture photometry at the RA/Dec specified in the header
-    """
-
-    base_key = "APERPHOTIM"
-
-    def __init__(
-        self,
-        *args,
-        aper_diameters: float | list[float] = 10.0,
-        bkg_in_diameters: float | list[float] = 25.0,
-        bkg_out_diameters: float | list[float] = 40.0,
-        col_suffix_list: Optional[list[str]] = None,
-        **kwargs,
-    ):
-        super().__init__(*args, **kwargs)
-
-        self.aperture_photometer = AperturePhotometry(
-            aper_diameters=aper_diameters,
-            bkg_in_diameters=bkg_in_diameters,
-            bkg_out_diameters=bkg_out_diameters,
-        )
-        self.col_suffix_list = col_suffix_list
-        if self.col_suffix_list is None:
-            self.col_suffix_list = self.aperture_photometer.aper_diameters
-
-    def _apply_to_images(
-        self,
-        batch: ImageBatch,
-    ) -> ImageBatch:
-        for image in batch:
-            image_cutout, unc_image_cutout = self.generate_cutouts(image)
-
-            fluxes, fluxuncs = self.aperture_photometer.perform_photometry(
-                image_cutout, unc_image_cutout
-            )
-
-            for ind, flux in enumerate(fluxes):
-                fluxunc = fluxuncs[ind]
-                suffix = self.col_suffix_list[ind]
-                image[f"fluxap{suffix}"] = flux
-                image[f"fluxunc{suffix}"] = fluxunc
-
-                magnitudes, magnitudes_unc = get_mags_from_fluxes(
-                    flux_list=[flux],
-                    fluxunc_list=[fluxunc],
-                    zeropoint_list=[float(image[self.zp_key])],
-                    zeropoint_unc_list=[float(image[self.zp_std_key])],
-                )
-
-                image[f"magap{suffix}"] = magnitudes[0]
-                image[f"magerrap{suffix}"] = magnitudes_unc[0]
-
-        return batch
diff --git a/mirar/processors/photometry/base_photometry.py b/mirar/processors/photometry/base_photometry.py
index 6e3ba0350..27409a7c0 100644
--- a/mirar/processors/photometry/base_photometry.py
+++ b/mirar/processors/photometry/base_photometry.py
@@ -1,13 +1,13 @@
 """
-Module with classes to perform photometry on an image or candidates
+Module with classes to perform photometry on sources
 """
 import logging
+from abc import ABC
 from pathlib import Path
 
 import numpy as np
 import pandas as pd
 from astropy.io import fits
-from astropy.wcs import WCS
 
 from mirar.data import Image
 from mirar.paths import (
@@ -21,108 +21,13 @@
     ZP_STD_KEY,
     get_output_dir,
 )
-from mirar.processors.base_processor import (
-    BaseImageProcessor,
-    BaseProcessor,
-    BaseSourceProcessor,
-    ImageHandler,
-)
-from mirar.processors.photometry.utils import (
-    aper_photometry,
-    get_rms_image,
-    make_cutouts,
-    make_psf_shifted_array,
-    psf_photometry,
-)
+from mirar.processors.base_processor import BaseSourceProcessor, ImageHandler
+from mirar.processors.photometry.utils import get_rms_image, make_cutouts
 
 logger = logging.getLogger(__name__)
 
 
-class BasePhotometry:
-    """
-    Parent class to run photometry on cutouts
-    """
-
-    def __init__(self):
-        pass
-
-    def perform_photometry(self, image_cutout: np.array, unc_image_cutout: np.array):
-        """
-        Function to perform photometry
-        Args:
-            image_cutout: 2D image cutout array
-            unc_image_cutout: 2D image uncertainty cutout
-
-        Returns:
-
-        """
-        raise NotImplementedError
-
-
-class PSFPhotometry(BasePhotometry):
-    """
-    Class to run PSF photometry on cutouts
-    """
-
-    def __init__(self, psf_filename: str):
-        super().__init__()
-        self.psf_filename = psf_filename
-
-    def perform_photometry(
-        self, image_cutout: np.array, unc_image_cutout: np.array
-    ) -> tuple[float, float, float, float, float]:
-        psfmodels = make_psf_shifted_array(
-            psf_filename=self.psf_filename,
-            cutout_size_psf_phot=int(image_cutout.shape[0] / 2),
-        )
-
-        flux, fluxunc, minchi2, xshift, yshift = psf_photometry(
-            image_cutout, unc_image_cutout, psfmodels
-        )
-        return flux, fluxunc, minchi2, xshift, yshift
-
-
-class AperturePhotometry(BasePhotometry):
-    """
-    Class to run aperture photometry on cutouts
-    """
-
-    def __init__(
-        self,
-        aper_diameters: float | list[float] = 10.0,
-        bkg_in_diameters: float | list[float] = 25.0,
-        bkg_out_diameters: float | list[float] = 40.0,
-    ):
-        if not isinstance(aper_diameters, list):
-            aper_diameters = [aper_diameters]
-        if not isinstance(bkg_in_diameters, list):
-            bkg_in_diameters = [bkg_in_diameters]
-        if not isinstance(bkg_out_diameters, list):
-            bkg_out_diameters = [bkg_out_diameters]
-        super().__init__()
-
-        self.aper_diameters = aper_diameters
-        self.bkg_in_diameters = bkg_in_diameters
-        self.bkg_out_diameters = bkg_out_diameters
-
-    def perform_photometry(
-        self, image_cutout: np.array, unc_image_cutout: np.array
-    ) -> tuple[list[float], list[float]]:
-        fluxes, fluxuncs = [], []
-        for ind, aper_diam in enumerate(self.aper_diameters):
-            flux, fluxunc = aper_photometry(
-                image_cutout,
-                unc_image_cutout,
-                aper_diam,
-                self.bkg_in_diameters[ind],
-                self.bkg_out_diameters[ind],
-            )
-            fluxes.append(flux)
-            fluxuncs.append(fluxunc)
-        return fluxes, fluxuncs
-
-
-class BasePhotometryProcessor(BaseProcessor, ImageHandler):
+class BasePhotometryProcessor(BaseSourceProcessor, ABC, ImageHandler):
     """
     Parent processor to run photometry
     """
@@ -133,48 +38,29 @@ def __init__(
         temp_output_sub_dir: str = "photometry",
         zp_key: str = ZP_KEY,
         zp_std_key: str = ZP_STD_KEY,
+        img_key: str = LATEST_SAVE_KEY,
+        unc_image_key: str = UNC_IMG_KEY,
+        psf_file_key: str = NORM_PSFEX_KEY,
+        x_colname: str = XPOS_KEY,
+        y_colname: str = YPOS_KEY,
     ):
-        """
-        Args:
-            phot_cutout_size: size of cutout to perform photometry on
-            temp_output_sub_dir: subdirectory to save temporary photometry files
-        """
         super().__init__()
         self.phot_cutout_size = phot_cutout_size
         self.temp_output_sub_dir = temp_output_sub_dir
         self.zp_key = zp_key
         self.zp_std_key = zp_std_key
-
-    def save_temp_image_uncimage(
-        self, data_item: Image | pd.Series
-    ) -> tuple[Path, Path]:
-        """
-        Create temporary image and uncertainty images and return their paths.
-        Args:
-            data_item: pandas DataFrame Series or astropy fits Header
-
-        Returns:
-            tuple: image name, uncertainty image name
-        """
-        raise NotImplementedError
-
-    def get_physical_coordinates(
-        self, data_item: Image | pd.Series
-    ) -> tuple[float, float]:
-        """
-        Get physical coordinates (x, y) at which to perform photometry
-        Args:
-            data_item: pandas DataFrame Series or astropy fits Header
-
-        Returns:
-            tuple: x, y coordinates at which photometry is performed
-
-        """
-        raise NotImplementedError
+        self.image_key = img_key
+        self.unc_image_key = unc_image_key
+        self.psf_file_key = psf_file_key
+        self.xpos_key = x_colname
+        self.ypos_key = y_colname
 
     def save_temp_image(self, image) -> Path:
         """
         Save a temporary image and return its path
+
+        :param image: Image object
+        :return: Path to the temporary image
         """
         photometry_out_temp_dir = get_output_dir(
             self.temp_output_sub_dir, self.night_sub_dir
@@ -189,6 +75,9 @@ def save_uncertainty_image(self, image: Image) -> Path:
         """
         Create an uncertainty image from the image and return the filenames of the
         image and uncertainty image
+
+        :param image: Image object
+        :return: Path to the uncertainty image
         """
         photometry_out_temp_dir = get_output_dir(
             self.temp_output_sub_dir, self.night_sub_dir
@@ -203,7 +92,7 @@ def save_uncertainty_image(self, image: Image) -> Path:
         return unc_filename
 
     def generate_cutouts(
-        self, data_item: Image | pd.Series
+        self, data_item: Image | pd.Series, metadata: dict
     ) -> tuple[np.ndarray, np.ndarray]:
         """
         Generate image and uncertainty image cutouts. This function first saves the
@@ -211,11 +100,13 @@ def generate_cutouts(
         then deletes the temporary files.
         Args:
             data_item: pandas DataFrame Series or astropy fits Header
+            metadata: Metadata dictionary
 
         Returns:
             tuple: 2D numpy arrays of the image cutout and uncertainty image cutout
         """
-        temp_imagename, temp_unc_imagename = self.save_temp_image_uncimage(data_item)
+
+        temp_imagename, temp_unc_imagename = self.save_temp_image_uncimage(metadata)
         x, y = self.get_physical_coordinates(data_item)
         image_cutout, unc_image_cutout = make_cutouts(
             image_paths=[temp_imagename, temp_unc_imagename],
@@ -226,64 +117,15 @@ def generate_cutouts(
         temp_unc_imagename.unlink()
         return image_cutout, unc_image_cutout
 
+    def save_temp_image_uncimage(self, metadata: dict) -> tuple[Path, Path]:
+        """
+        Function to save the image and uncertainty image to temporary files
 
-class BaseImagePhotometry(BasePhotometryProcessor, BaseImageProcessor):
-    """
-    Processor to run photometry on an image
-    """
-
-    def __init__(
-        self,
-        *args,
-        target_ra_key: str = "TARGRA",
-        target_dec_key: str = "TARGDEC",
-        **kwargs,
-    ):
-        super().__init__(*args, **kwargs)
-        self.target_ra_key = target_ra_key
-        self.target_dec_key = target_dec_key
-
-    def save_temp_image_uncimage(self, data_item: Image):
-        image = data_item
-        image_filename = self.save_temp_image(image)
-        unc_filename = self.save_uncertainty_image(image)
-        return image_filename, unc_filename
-
-    def get_physical_coordinates(self, data_item: Image) -> (int, int):
-        image = data_item
-        ra, dec = image[self.target_ra_key], image[self.target_dec_key]
-        wcs = WCS(image.header)
-        x, y = wcs.all_world2pix(ra, dec, 0)
-        return int(np.round(x)), int(np.round(y))
-
-
-class BaseSourcePhotometry(BasePhotometryProcessor, BaseSourceProcessor):
-    """
-    Processor to run photometry on a candidates table
-    """
-
-    base_key = "base_source_photometry"
-
-    def __init__(
-        self,
-        *args,
-        image_colname=LATEST_SAVE_KEY,
-        unc_image_colname=UNC_IMG_KEY,
-        psf_file_colname=NORM_PSFEX_KEY,
-        x_colname=XPOS_KEY,
-        y_colname=YPOS_KEY,
-        **kwargs,
-    ):
-        super().__init__(*args, **kwargs)
-        self.image_key = image_colname
-        self.unc_image_key = unc_image_colname
-        self.psf_file_key = psf_file_colname
-        self.xpos_key = x_colname
-        self.ypos_key = y_colname
+        :param metadata: Metadata dictionary
+        :return: Tuple of image and uncertainty image filenames
+        """
+        imagename = metadata[self.image_key]
 
-    def save_temp_image_uncimage(self, data_item: pd.Series):
-        row = data_item
-        imagename = row[self.image_key]
         image = Image(header=fits.getheader(imagename), data=fits.getdata(imagename))
 
         image_filename = self.save_temp_image(image)
@@ -291,7 +133,13 @@ def save_temp_image_uncimage(self, data_item: pd.Series):
 
         return image_filename, unc_filename
 
-    def get_physical_coordinates(self, data_item: pd.Series):
+    def get_physical_coordinates(self, data_item: pd.Series) -> tuple[int, int]:
+        """
+        Get the physical coordinates of the source from the data item
+
+        :param data_item: Series from the data table
+        :return: X and Y coordinates of the source
+        """
         row = data_item
         x, y = row[self.xpos_key], row[self.ypos_key]
         return int(x), int(y)
diff --git a/mirar/processors/photometry/psf_photometry.py b/mirar/processors/photometry/psf_photometry.py
index 1001db80e..8344fa86b 100644
--- a/mirar/processors/photometry/psf_photometry.py
+++ b/mirar/processors/photometry/psf_photometry.py
@@ -2,55 +2,43 @@
 Module with processors to perform point-spread-function photometry
 """
 import logging
+from pathlib import Path
 
 import numpy as np
 
-from mirar.data import Image, ImageBatch, SourceBatch
-from mirar.paths import (
-    MAG_PSF_KEY,
-    MAGERR_PSF_KEY,
-    NORM_PSFEX_KEY,
-    PSF_FLUX_KEY,
-    PSF_FLUXUNC_KEY,
-)
+from mirar.data import SourceBatch
+from mirar.paths import MAG_PSF_KEY, MAGERR_PSF_KEY, PSF_FLUX_KEY, PSF_FLUXUNC_KEY
 from mirar.processors.astromatic.psfex import PSFex
 from mirar.processors.base_processor import PrerequisiteError
-from mirar.processors.photometry.base_photometry import (
-    BaseImagePhotometry,
-    BaseSourcePhotometry,
-    PSFPhotometry,
+from mirar.processors.photometry.base_photometry import BasePhotometryProcessor
+from mirar.processors.photometry.utils import (
+    get_mags_from_fluxes,
+    make_psf_shifted_array,
+    psf_photometry,
 )
-from mirar.processors.photometry.utils import get_mags_from_fluxes
 
 logger = logging.getLogger(__name__)
 
 
-def check_psf_phot_prerequisites(processor):
+class PSFPhotometry(BasePhotometryProcessor):
     """
-    Function to check prerequisites for running PSF photometry
-    Args:
-        processor: PSF photometry processor
-
+    Processor to run PSF photometry on a source table
     """
-    mask = [isinstance(x, PSFex) for x in processor.preceding_steps]
-    if np.sum(mask) < 1:
-        err = (
-            f"{processor.__module__} requires {PSFex} as a prerequisite. "
-            f"However, the following steps were found: {processor.preceding_steps}."
-        )
-        logger.error(err)
-        raise PrerequisiteError(err)
 
+    base_key = "PSFPHOT"
 
-class SourcePSFPhotometry(BaseSourcePhotometry):
-    """
-    Processor to run PSF photometry on all candidates in candidate table
-    """
-
-    base_key = "PSFPHOTDF"
+    def perform_photometry(
+        self, image_cutout: np.array, unc_image_cutout: np.array, psf_filename: Path
+    ) -> tuple[float, float, float, float, float]:
+        psfmodels = make_psf_shifted_array(
+            psf_filename=psf_filename,
+            cutout_size_psf_phot=int(image_cutout.shape[0] / 2),
+        )
 
-    def __init__(self, **kwargs):
-        super().__init__(**kwargs)
+        flux, fluxunc, minchi2, xshift, yshift = psf_photometry(
+            image_cutout, unc_image_cutout, psfmodels
+        )
+        return flux, fluxunc, minchi2, xshift, yshift
 
     def get_psf_filename(self, row):
         """
@@ -71,21 +59,23 @@ def _apply_to_sources(
         for source_table in batch:
             candidate_table = source_table.get_data()
 
+            metadata = source_table.get_metadata()
+
             fluxes, fluxuncs, minchi2s, xshifts, yshifts = [], [], [], [], []
 
-            for ind in range(len(candidate_table)):
-                row = candidate_table.iloc[ind]
+            psf_filename = source_table[self.psf_file_key]
 
-                image_cutout, unc_image_cutout = self.generate_cutouts(row)
-                psf_filename = self.get_psf_filename(row)
-                psf_photometer = PSFPhotometry(psf_filename=psf_filename)
+            for _, row in candidate_table.iterrows():
+                image_cutout, unc_image_cutout = self.generate_cutouts(row, metadata)
                 (
                     flux,
                     fluxunc,
                     minchi2,
                     xshift,
                     yshift,
-                ) = psf_photometer.perform_photometry(image_cutout, unc_image_cutout)
+                ) = self.perform_photometry(
+                    image_cutout, unc_image_cutout, psf_filename=psf_filename
+                )
                 fluxes.append(flux)
                 fluxuncs.append(fluxunc)
                 minchi2s.append(minchi2)
@@ -99,12 +89,10 @@ def _apply_to_sources(
             candidate_table["yshift"] = yshifts
 
             magnitudes, magnitudes_unc = get_mags_from_fluxes(
-                flux_list=fluxes,
-                fluxunc_list=fluxuncs,
-                zeropoint_list=np.array(candidate_table[self.zp_key], dtype=float),
-                zeropoint_unc_list=np.array(
-                    candidate_table[self.zp_std_key], dtype=float
-                ),
+                flux_list=np.array(fluxes, dtype=float),
+                fluxunc_list=np.array(fluxuncs, dtype=float),
+                zeropoint=float(source_table[self.zp_key]),
+                zeropoint_unc=float(source_table[self.zp_std_key]),
             )
 
             candidate_table[MAG_PSF_KEY] = magnitudes
@@ -117,63 +105,11 @@ def _apply_to_sources(
     def check_prerequisites(
         self,
     ):
-        check_psf_phot_prerequisites(self)
-
-
-class ImagePSFPhotometry(BaseImagePhotometry):
-    """
-    Processor to run PSF photometry at the RA/Dec specified in the header
-    """
-
-    base_key = "PSFPHOTIM"
-
-    def __init__(
-        self,
-        *args,
-        **kwargs,
-    ):
-        super().__init__(*args, **kwargs)
-
-    def get_psf_filename(self, image: Image):
-        """
-        Function to get PSF file name of an image
-        Args:
-            image: Image
-
-        Returns:
-
-        """
-        psf_filename = image[NORM_PSFEX_KEY]
-        return psf_filename
-
-    def _apply_to_images(
-        self,
-        batch: ImageBatch,
-    ) -> ImageBatch:
-        for image in batch:
-            image_cutout, unc_image_cutout = self.generate_cutouts(image)
-            psf_filename = self.get_psf_filename(image)
-
-            psf_photometer = PSFPhotometry(psf_filename=psf_filename)
-
-            flux, fluxunc, _, _, _ = psf_photometer.perform_photometry(
-                image_cutout, unc_image_cutout
+        mask = [isinstance(x, PSFex) for x in self.preceding_steps]
+        if np.sum(mask) < 1:
+            err = (
+                f"{self.__module__} requires {PSFex} as a prerequisite. "
+                f"However, the following steps were found: {self.preceding_steps}."
             )
-
-            magnitudes, magnitudes_unc = get_mags_from_fluxes(
-                flux_list=[flux],
-                fluxunc_list=[fluxunc],
-                zeropoint_list=[float(image[self.zp_key])],
-                zeropoint_unc_list=[float(image[self.zp_std_key])],
-            )
-            image[PSF_FLUX_KEY] = flux
-            image[PSF_FLUXUNC_KEY] = fluxunc
-            image[MAG_PSF_KEY] = magnitudes[0]
-            image[MAGERR_PSF_KEY] = magnitudes_unc[0]
-
-        return batch
-
-    def check_prerequisites(
-        self,
-    ):
-        check_psf_phot_prerequisites(self)
+            logger.error(err)
+            raise PrerequisiteError(err)
diff --git a/mirar/processors/photometry/utils.py b/mirar/processors/photometry/utils.py
index 17041f205..22a2ec18d 100644
--- a/mirar/processors/photometry/utils.py
+++ b/mirar/processors/photometry/utils.py
@@ -285,34 +285,25 @@ def get_rms_image(image: Image) -> Image:
 
 
 def get_mags_from_fluxes(
-    flux_list: list[float] | np.ndarray,
-    fluxunc_list: list[float] | np.ndarray,
-    zeropoint_list: list[float] | np.ndarray,
-    zeropoint_unc_list: list[float] | np.ndarray,
+    flux_list: np.ndarray[float],
+    fluxunc_list: np.ndarray[float],
+    zeropoint: float,
+    zeropoint_unc: float,
 ) -> tuple[np.ndarray, np.ndarray]:
     """Convert fluxes and flux uncertainties to magnitudes and magnitude uncertainties
 
     :param flux_list: List of fluxes
     :param fluxunc_list: List of flux uncertainties
-    :param zeropoint_list: List of zeropoints
-    :param zeropoint_unc_list: List of zeropoint uncertainties
+    :param zeropoint: Zeropoint
+    :param zeropoint_unc: Zeropoint uncertainties
     :return: magnitudes, magnitude uncertainties
     """
-    assert len(flux_list) == len(fluxunc_list) == len(zeropoint_list)
-
-    if isinstance(flux_list, list):
-        flux_list = np.array(flux_list)
-    if isinstance(fluxunc_list, list):
-        fluxunc_list = np.array(fluxunc_list)
-    if isinstance(zeropoint_list, list):
-        zeropoint_list = np.array(zeropoint_list)
-    if isinstance(zeropoint_unc_list, list):
-        zeropoint_unc_list = np.array(zeropoint_unc_list)
-
-    magnitudes = zeropoint_list - 2.5 * np.log10(flux_list)
+    assert len(flux_list) == len(fluxunc_list)
+
+    magnitudes = zeropoint - 2.5 * np.log10(flux_list)
     magnitudes_unc = 1.086 * fluxunc_list / flux_list
 
-    magnitudes_unc = np.sqrt(magnitudes_unc**2 + zeropoint_unc_list**2)
+    magnitudes_unc = np.sqrt(magnitudes_unc**2 + zeropoint_unc**2)
     magnitudes = magnitudes.astype(object)
     magnitudes_unc = magnitudes_unc.astype(object)
     magnitudes[flux_list <= 0] = None
diff --git a/mirar/processors/skyportal/__init__.py b/mirar/processors/skyportal/__init__.py
new file mode 100644
index 000000000..a9fd1988b
--- /dev/null
+++ b/mirar/processors/skyportal/__init__.py
@@ -0,0 +1,10 @@
+"""
+Central module for the skyportal integration
+"""
+from mirar.processors.skyportal.client import SkyportalClient
+from mirar.processors.skyportal.skyportal_candidate import SkyportalCandidateUploader
+from mirar.processors.skyportal.skyportal_source import (
+    SNCOSMO_KEY,
+    SkyportalSourceUploader,
+)
+from mirar.processors.skyportal.thumbnail import make_thumbnail
diff --git a/mirar/processors/skyportal/client.py b/mirar/processors/skyportal/client.py
new file mode 100644
index 000000000..a67d1f7d6
--- /dev/null
+++ b/mirar/processors/skyportal/client.py
@@ -0,0 +1,162 @@
+"""
+Client for interacting with Skyportal API
+"""
+import logging
+import os
+from typing import Mapping, Optional
+from urllib.parse import urljoin
+
+import requests
+from requests.adapters import HTTPAdapter
+from urllib3.util import Retry
+
+DEFAULT_TIMEOUT = 5  # seconds
+
+logger = logging.getLogger(__name__)
+
+
+class TimeoutHTTPAdapter(HTTPAdapter):
+    """
+    HTTP adapter that sets a default timeout for all requests.
+    """
+
+    def __init__(self, *args, **kwargs):
+        self.timeout = DEFAULT_TIMEOUT
+        if "timeout" in kwargs:
+            self.timeout = kwargs["timeout"]
+            del kwargs["timeout"]
+        super().__init__(*args, **kwargs)
+
+    def send(self, request, *args, **kwargs):
+        """
+        Send a request with a default timeout.
+
+        :param request: request to send
+        :param args: args to pass to super().send
+        :param kwargs: kwargs to pass to super().send
+        :return: response from request
+        """
+        try:
+            timeout = kwargs.get("timeout")
+            if timeout is None:
+                kwargs["timeout"] = self.timeout
+            return super().send(request, *args, **kwargs)
+        except AttributeError:
+            kwargs["timeout"] = DEFAULT_TIMEOUT
+
+
+class SkyportalClient:
+    """
+    Basic Skyportal client class for executing functions
+    """
+
+    def __init__(
+        self,
+        base_url: str = "https://fritz.science/api/",
+    ):
+        self.base_url = base_url
+        self._session = None
+        self.session_headers = None
+
+    def set_up_session(self):
+        """
+        Set up a session for sending requests to Skyportal.
+
+        :return: None
+        """
+        # session to talk to SkyPortal
+        self._session = requests.Session()
+        self.session_headers = {
+            "Authorization": f"token {self._get_fritz_token()}",
+            "User-Agent": "mirar",
+        }
+
+        retries = Retry(
+            total=5,
+            backoff_factor=2,
+            status_forcelist=[405, 429, 500, 502, 503, 504],
+            allowed_methods=["HEAD", "GET", "PUT", "POST", "PATCH"],
+        )
+        adapter = TimeoutHTTPAdapter(timeout=5, max_retries=retries)
+        self._session.mount("https://", adapter)
+        self._session.mount("http://", adapter)
+
+    def get_session(self) -> requests.Session:
+        """
+        Wrapper for getting the session.
+        If the session is not set up, it will be set up.
+
+        :return: Session
+        """
+        if self._session is None:
+            self.set_up_session()
+
+        return self._session
+
+    @staticmethod
+    def _get_fritz_token():
+        """
+        Get Fritz token from environment variable.
+
+        :return: Fritz token
+        """
+        token_fritz = os.getenv("FRITZ_TOKEN")
+
+        if token_fritz is None:
+            err = (
+                "No Fritz token specified. Run 'export FRITZ_TOKEN=<token>' to "
+                "set. The Fritz token will need to be specified manually "
+                "for Fritz API queries."
+            )
+            logger.error(err)
+            raise ValueError(err)
+
+        return token_fritz
+
+    def api(
+        self, method: str, endpoint: str, data: Optional[Mapping] = None
+    ) -> requests.Response:
+        """Make an API call to a SkyPortal instance
+
+        headers = {'Authorization': f'token {self.token}'}
+        response = requests.request(method, endpoint, json_dict=data, headers=headers)
+
+        :param method: HTTP method
+        :param endpoint: API endpoint
+        :param data: JSON data to send
+        :return: response from API call
+        """
+        method = method.lower()
+
+        session = self.get_session()
+
+        methods = {
+            "head": session.head,
+            "get": session.get,
+            "post": session.post,
+            "put": session.put,
+            "patch": session.patch,
+            "delete": session.delete,
+        }
+
+        if endpoint is None:
+            raise ValueError("Endpoint not specified")
+        if method not in ["head", "get", "post", "put", "patch", "delete"]:
+            raise ValueError(f"Unsupported method: {method}")
+
+        url = urljoin(self.base_url, endpoint)
+
+        if method == "get":
+            response = methods[method](
+                url,
+                params=data,
+                headers=self.session_headers,
+            )
+        else:
+            response = methods[method](
+                url,
+                json=data,
+                headers=self.session_headers,
+            )
+
+        return response
diff --git a/mirar/processors/skyportal/skyportal_candidate.py b/mirar/processors/skyportal/skyportal_candidate.py
new file mode 100644
index 000000000..91c2cf828
--- /dev/null
+++ b/mirar/processors/skyportal/skyportal_candidate.py
@@ -0,0 +1,234 @@
+"""
+Module for sending candidates to Fritz.
+"""
+import logging
+from datetime import datetime
+
+from astropy.time import Time
+
+from mirar.paths import CAND_NAME_KEY, PACKAGE_NAME, __version__
+from mirar.processors.skyportal.skyportal_source import SkyportalSourceUploader
+
+logger = logging.getLogger(__name__)
+
+
+class SkyportalCandidateUploader(SkyportalSourceUploader):
+    """
+    Processor for sending candidates to Fritz.
+    """
+
+    def skyportal_post_candidate(self, alert):
+        """
+        Post a candidate on SkyPortal. Creates new candidate(s) (one per filter)
+
+        :param alert: dict of alert info
+        :return: None
+        """
+
+        data = {
+            "id": alert[CAND_NAME_KEY],
+            "ra": alert["ra"],
+            "dec": alert["dec"],
+            "filter_ids": [self.fritz_filter_id],
+            "passing_alert_id": self.fritz_filter_id,
+            "passed_at": Time(datetime.utcnow()).isot,
+            "origin": f"{PACKAGE_NAME}:{__version__}",
+        }
+
+        logger.debug(
+            f"Posting metadata of {alert[CAND_NAME_KEY]} {alert['candid']} to SkyPortal"
+        )
+        response = self.api("POST", "candidates", data)
+
+        if response.json()["status"] == "success":
+            logger.debug(
+                f"Posted {alert[CAND_NAME_KEY]} {alert['candid']} metadata to SkyPortal"
+            )
+        else:
+            logger.error(
+                f"Failed to post {alert[CAND_NAME_KEY]} {alert['candid']} "
+                f"metadata to SkyPortal"
+            )
+            logger.error(response.json())
+
+    def skyportal_post_annotation(self, alert):
+        """
+        Post an annotation. Works for both candidates and sources.
+
+        :param alert: alert data
+        :return: None
+        """
+        data = {
+            "chipsf": alert["chipsf"],
+            "fwhm": alert["fwhm"],
+            "scorr": alert["scorr"],
+        }
+        payload = {"origin": self.origin, "data": data, "group_ids": self.group_ids}
+
+        path = f"sources/{str(alert[CAND_NAME_KEY])}/annotations"
+        response = self.api("POST", path, payload)
+
+        if response.json()["status"] == "success":
+            logger.debug(f"Posted {alert[CAND_NAME_KEY]} annotation to SkyPortal")
+        else:
+            logger.error(
+                f"Failed to post {alert[CAND_NAME_KEY]} annotation to SkyPortal"
+            )
+            logger.error(response.json())
+
+    def skyportal_put_annotation(self, source):
+        """
+        Retrieve an annotation to check if it exists already.
+
+        :param source: detection data
+        :return: None
+        """
+        response = self.api(
+            "GET",
+            f"sources/{str(source[CAND_NAME_KEY])}/annotations",
+        )
+
+        if response.json()["status"] == "success":
+            logger.debug(f"Got {source[CAND_NAME_KEY]} annotations from SkyPortal")
+        else:
+            logger.debug(
+                f"Failed to get {source[CAND_NAME_KEY]} annotations from SkyPortal"
+            )
+            logger.debug(response.json())
+            return
+
+        existing_annotations = {
+            annotation["origin"]: {
+                "annotation_id": annotation["id"],
+                "author_id": annotation["author_id"],
+            }
+            for annotation in response.json()["data"]
+        }
+
+        # no previous annotation exists on SkyPortal for this object? just post then
+        if self.origin not in existing_annotations:
+            self.skyportal_post_annotation(source)
+        # annotation from this(WNTR) origin exists
+        else:
+            # annotation data
+            data = {
+                "fwhm": source["fwhm"],
+                "scorr": source["scorr"],
+                "chipsf": source["chipsf"],
+            }
+            new_annotation = {
+                "author_id": existing_annotations[self.origin]["author_id"],
+                "obj_id": source[CAND_NAME_KEY],
+                "origin": self.origin,
+                "data": data,
+                "group_ids": self.group_ids,
+            }
+
+            logger.debug(
+                f"Putting annotation for {source[CAND_NAME_KEY]} {source['candid']} "
+                f"to SkyPortal",
+            )
+            response = self.api(
+                "PUT",
+                f"sources/{source[CAND_NAME_KEY]}"
+                f"/annotations/{existing_annotations[self.origin]['annotation_id']}",
+                new_annotation,
+            )
+            if response.json()["status"] == "success":
+                logger.debug(
+                    f"Posted updated {source[CAND_NAME_KEY]} annotation to SkyPortal"
+                )
+            else:
+                logger.error(
+                    f"Failed to post updated {source[CAND_NAME_KEY]} annotation "
+                    f"to SkyPortal"
+                )
+                logger.error(response.json())
+
+    def export_to_skyportal(self, alert):
+        """
+        Posts a candidate to SkyPortal.
+
+        :param alert: _description_
+        :type alert: _type_
+        """
+        # check if candidate exists in SkyPortal
+        logger.debug(f"Checking if {alert[CAND_NAME_KEY]} is candidate in SkyPortal")
+        response = self.api("HEAD", f"candidates/{alert[CAND_NAME_KEY]}")
+
+        if response.status_code not in [200, 404]:
+            response.raise_for_status()
+
+        is_candidate = response.status_code == 200
+        logger.debug(
+            f"{alert[CAND_NAME_KEY]} {'is' if is_candidate else 'is not'} "
+            f"candidate in SkyPortal"
+        )
+
+        # check if source exists in SkyPortal
+        logger.debug(f"Checking if {alert[CAND_NAME_KEY]} is source in SkyPortal")
+        response = self.api("HEAD", f"sources/{alert[CAND_NAME_KEY]}")
+
+        if response.status_code not in [200, 404]:
+            response.raise_for_status()
+
+        is_source = response.status_code == 200
+        logger.debug(
+            f"{alert[CAND_NAME_KEY]} "
+            f"{'is' if is_source else 'is not'} source in SkyPortal"
+        )
+
+        # object does not exist in SkyPortal: neither cand nor source
+        if (not is_candidate) and (not is_source):
+            # post candidate
+            self.skyportal_post_candidate(alert)
+
+            # post annotations
+            self.skyportal_post_annotation(alert)
+
+            # post full light curve
+            self.skyportal_put_photometry(alert)
+
+            # post thumbnails
+            self.skyportal_post_thumbnails(alert)
+
+        # obj already exists in SkyPortal
+        else:
+            # post candidate with new filter ids
+            self.skyportal_post_candidate(alert)
+
+            # put (*not* post) annotations
+            self.skyportal_put_annotation(alert)
+
+            # exists in SkyPortal & already saved as a source
+            if is_source:
+                # get info on the corresponding groups:
+                logger.debug(
+                    f"Getting source groups info on "
+                    f"{alert[CAND_NAME_KEY]} from SkyPortal",
+                )
+                response = self.api(
+                    "GET",
+                    f"sources/{alert[CAND_NAME_KEY]}/groups",
+                )
+                if response.json()["status"] == "success":
+                    existing_groups = response.json()["data"]
+                    existing_group_ids = [g["id"] for g in existing_groups]
+
+                    for existing_gid in existing_group_ids:
+                        if existing_gid in self.group_ids:
+                            self.skyportal_post_source(alert, [existing_gid])
+                else:
+                    logger.error(
+                        f"Failed to get source groups info on {alert[CAND_NAME_KEY]}"
+                    )
+            else:  # exists in SkyPortal but NOT saved as a source
+                self.skyportal_post_source(alert)
+
+            # post alert photometry in single call to /api/photometry
+            self.skyportal_put_photometry(alert)
+
+            if self.update_thumbnails:
+                self.skyportal_post_thumbnails(alert)
+
+        logger.debug(f"SendToSkyportal Manager complete for {alert[CAND_NAME_KEY]}")
diff --git a/mirar/processors/skyportal/skyportal_source.py b/mirar/processors/skyportal/skyportal_source.py
new file mode 100644
index 000000000..9846162a8
--- /dev/null
+++ b/mirar/processors/skyportal/skyportal_source.py
@@ -0,0 +1,315 @@
+"""
+Module for sending candidates to Fritz.
+"""
+import logging
+from copy import deepcopy
+from typing import Mapping, Optional
+
+import matplotlib
+import numpy as np
+import pandas as pd
+import requests
+
+from mirar.data import SourceBatch
+from mirar.data.utils import decode_img
+from mirar.paths import CAND_NAME_KEY, SOURCE_HISTORY_KEY
+from mirar.processors.base_processor import BaseSourceProcessor
+from mirar.processors.skyportal.client import SkyportalClient
+from mirar.processors.skyportal.thumbnail import make_thumbnail
+
+matplotlib.use("agg")
+
+logger = logging.getLogger(__name__)
+
+SNCOSMO_KEY = "sncosmof"
+
+
+class SkyportalSourceUploader(BaseSourceProcessor):
+    """
+    Processor for sending source photometry to Skyportal.
+    """
+
+    base_key = "skyportalsender"
+
+    def __init__(
+        self,
+        origin: str,
+        group_ids: list[int],
+        fritz_filter_id: int,
+        instrument_id: int,
+        stream_id: int,
+        update_thumbnails: bool = False,
+    ):
+        super().__init__()
+        self.group_ids = group_ids
+        self.fritz_filter_id = fritz_filter_id
+        self.instrument_id = instrument_id
+        self.origin = origin  # used for sending updates to Fritz
+        self.stream_id = stream_id
+        self.update_thumbnails = update_thumbnails
+        self.skyportal_client = SkyportalClient()
+
+    def _apply_to_sources(
+        self,
+        batch: SourceBatch,
+    ) -> SourceBatch:
+        """
+        Apply the processor to a batch of candidates.
+
+        :param batch: SourceBatch to process
+        :return: SourceBatch after processing
+        """
+        for source_table in batch:
+            candidate_df = source_table.get_data()
+
+            metadata = source_table.get_metadata()
+
+            for _, src in candidate_df.iterrows():
+                super_dict = self.generate_super_dict(metadata, src)
+                self.export_to_skyportal(deepcopy(super_dict))
+
+        return batch
+
+    def skyportal_post_source(self, alert: dict, group_ids: Optional[list[int]] = None):
+        """Add a new source to SkyPortal
+
+        :param alert: dict of source info
+        :param group_ids: list of group_ids to post source to, defaults to None
+        :return: None
+        """
+        if group_ids is None:
+            group_ids = self.group_ids
+
+        data = {
+            "ra": alert["ra"],
+            "dec": alert["dec"],
+            "id": alert[CAND_NAME_KEY],
+            "group_ids": group_ids,
+            "origin": self.origin,
+        }
+
+        logger.debug(
+            f"Saving {alert[CAND_NAME_KEY]} {alert['candid']} as a Source on SkyPortal"
+        )
+        response = self.api("POST", "sources", data)
+
+        if response.json()["status"] == "success":
+            logger.debug(
+                f"Saved {alert[CAND_NAME_KEY]} {alert['candid']} "
+                f"as a Source on SkyPortal"
+            )
+        else:
+            err = (
+                f"Failed to save {alert[CAND_NAME_KEY]} {alert['candid']} "
+                f"as a Source on SkyPortal"
+            )
+            logger.error(err)
+            logger.error(response.json())
+
+    def api(
+        self, method: str, endpoint: str, data: Optional[Mapping] = None
+    ) -> requests.Response:
+        """Make an API call to a SkyPortal instance
+
+        headers = {'Authorization': f'token {self.token}'}
+        response = requests.request(method, endpoint, json_dict=data, headers=headers)
+
+        :param method: HTTP method
+        :param endpoint: API endpoint e.g sources
+        :param data: JSON data to send
+        :return: response from API call
+        """
+        return self.skyportal_client.api(method, endpoint, data)
+
+    def make_thumbnail(
+        self, source: pd.Series, skyportal_type: str, alert_packet_type: str
+    ):
+        """
+        Convert lossless FITS cutouts from ZTF-like alerts into PNGs.
+        Make thumbnail for pushing to SkyPortal.
+
+        :param source: ZTF-like alert packet/dict
+        :param skyportal_type: <new|ref|sub> thumbnail type expected by SkyPortal
+        :param alert_packet_type: <Science|Template|Difference> survey naming
+        :return:
+        """
+        source = deepcopy(source)
+        cutout_data = source[f"cutout_{alert_packet_type}"]
+
+        linear_stretch = alert_packet_type.lower() in ["difference"]
+
+        skyportal_thumbnal = make_thumbnail(
+            image_data=decode_img(cutout_data),
+            linear_stretch=linear_stretch,
+        )
+
+        thumbnail_dict = {
+            "obj_id": source[CAND_NAME_KEY],
+            "data": skyportal_thumbnal,
+            "ttype": skyportal_type,
+        }
+
+        return thumbnail_dict
+
+    def skyportal_post_thumbnails(self, alert):
+        """Post alert Science, Reference, and Subtraction thumbnails to SkyPortal
+
+        :param alert: dict of source/candidate information
+        :return: None
+        """
+        for ttype, instrument_type in [
+            ("new", "science"),
+            ("ref", "template"),
+            ("sub", "difference"),
+        ]:
+            logger.debug(
+                f"Making {instrument_type} thumbnail for {alert[CAND_NAME_KEY]} "
+                f"{alert['candid']}",
+            )
+            thumb = self.make_thumbnail(alert, ttype, instrument_type)
+
+            logger.debug(
+                f"Posting {instrument_type} thumbnail for {alert[CAND_NAME_KEY]} "
+                f"{alert['candid']} to SkyPortal",
+            )
+            response = self.api("POST", "thumbnail", thumb)
+
+            if response.json()["status"] == "success":
+                logger.debug(
+                    f"Posted {alert[CAND_NAME_KEY]} {alert['candid']} "
+                    f"{instrument_type} cutout to SkyPortal"
+                )
+            else:
+                logger.error(
+                    f"Failed to post {alert[CAND_NAME_KEY]} {alert['candid']} "
+                    f"{instrument_type} cutout to SkyPortal"
+                )
+                logger.error(response.json())
+
+    def make_photometry(self, source: pd.Series) -> pd.DataFrame:
+        """
+        Make a de-duplicated pandas.DataFrame with photometry of alert[CAND_NAME_KEY]
+        Modified from Kowalksi (https://github.com/dmitryduev/kowalski)
+
+        :param source: source row
+        :return: pandas.DataFrame with photometry
+        """
+
+        photometry_table = [
+            {
+                "mjd": source["mjd"],
+                "mag": source["magpsf"],
+                "magerr": source["sigmapsf"],
+                "filter": source[SNCOSMO_KEY],
+                "ra": source["ra"],
+                "dec": source["dec"],
+            }
+        ]
+
+        if len(source[SOURCE_HISTORY_KEY]) > 0:
+            prv_detections = pd.DataFrame.from_records(source[SOURCE_HISTORY_KEY])
+
+            for _, row in prv_detections.iterrows():
+                try:
+                    photometry_table.append(
+                        {
+                            "mjd": row["mjd"],
+                            "mag": row["magpsf"],
+                            "magerr": row["sigmapsf"],
+                            "filter": row[SNCOSMO_KEY],
+                            "ra": row["ra"],
+                            "dec": row["dec"],
+                        }
+                    )
+                except KeyError:
+                    logger.warning(
+                        f"Missing photometry information for previous "
+                        f"detection of {source[CAND_NAME_KEY]}"
+                    )
+
+        df_photometry = pd.DataFrame(photometry_table)
+
+        df_photometry["magsys"] = "ab"
+
+        # step 1: calculate the coefficient that determines whether the
+        # flux should be negative or positive
+        coeff = 1.0 if source["isdiffpos"] else -1.0
+
+        # step 2: calculate the flux normalized to an arbitrary AB zeropoint of
+        # 23.9 (results in flux in uJy)
+        df_photometry["flux"] = coeff * 10 ** (-0.4 * (df_photometry["mag"] - 23.9))
+
+        # step 4a: calculate fluxerr for detections using sigmapsf
+        df_photometry["fluxerr"] = (
+            df_photometry["magerr"] * df_photometry["flux"] * np.log(10) / 2.5
+        )
+
+        # step 5: set the zeropoint and magnitude system
+        df_photometry["zp"] = 23.9
+
+        df_photometry.drop(columns=["mag", "magerr"], inplace=True)
+
+        return df_photometry
+
+    def skyportal_put_photometry(self, alert):
+        """Send photometry to Fritz."""
+        logger.debug(
+            f"Making alert photometry of {alert[CAND_NAME_KEY]} {alert['candid']}"
+        )
+        df_photometry = self.make_photometry(alert)
+
+        if len(df_photometry) > 0:
+            photometry = df_photometry.to_dict("list")
+            photometry["obj_id"] = alert[CAND_NAME_KEY]
+            photometry["stream_ids"] = [int(self.stream_id)]
+            photometry["instrument_id"] = self.instrument_id
+
+            logger.debug(
+                f"Posting photometry of {alert[CAND_NAME_KEY]} {alert['candid']}, "
+                f"stream_id={self.stream_id} to SkyPortal"
+            )
+            response = self.api("PUT", "photometry", photometry)
+            if response.json()["status"] == "success":
+                logger.debug(
+                    f"Posted {alert[CAND_NAME_KEY]} photometry "
+                    f"stream_id={self.stream_id} to SkyPortal"
+                )
+            else:
+                logger.error(
+                    f"Failed to post {alert[CAND_NAME_KEY]} photometry "
+                    f"stream_id={self.stream_id} to SkyPortal"
+                )
+                logger.error(response.json())
+
+    def export_to_skyportal(self, alert):
+        """
+        Posts a source to SkyPortal.
+
+        :param alert: _description_
+        :type alert: _type_
+        """
+        # check if source exists in SkyPortal
+        logger.debug(f"Checking if {alert[CAND_NAME_KEY]} is source in SkyPortal")
+        response = self.api("HEAD", f"sources/{alert[CAND_NAME_KEY]}")
+
+        if response.status_code not in [200, 404]:
+            response.raise_for_status()
+
+        is_source = response.status_code == 200
+        logger.debug(
+            f"{alert[CAND_NAME_KEY]} "
+            f"{'is' if is_source else 'is not'} source in SkyPortal"
+        )
+
+        if not is_source:
+            self.skyportal_post_source(alert, group_ids=self.group_ids)
+            # post thumbnails
+            self.skyportal_post_thumbnails(alert)
+
+        # post full light curve
+        self.skyportal_put_photometry(alert)
+
+        if self.update_thumbnails:
+            self.skyportal_post_thumbnails(alert)
+
+        logger.debug(f"SendToSkyportal Manager complete for {alert[CAND_NAME_KEY]}")
diff --git a/mirar/processors/skyportal/thumbnail.py b/mirar/processors/skyportal/thumbnail.py
new file mode 100644
index 000000000..31338d59c
--- /dev/null
+++ b/mirar/processors/skyportal/thumbnail.py
@@ -0,0 +1,70 @@
+"""
+Function for converting a FITS image to a skyportal "thumbnail".
+"""
+import base64
+import io
+
+import matplotlib
+import matplotlib.pyplot as plt
+import numpy as np
+from astropy.visualization import (
+    AsymmetricPercentileInterval,
+    ImageNormalize,
+    LinearStretch,
+    LogStretch,
+)
+
+matplotlib.use("agg")
+
+
+def make_thumbnail(
+    image_data: np.ndarray,
+    linear_stretch: bool = False,
+) -> str:
+    """
+    Util function to convert a FITS image to a PNG image
+
+    Skyportal requires, to quote the API:
+
+        base64-encoded PNG image file contents.
+        Image size must be between 16px and 500px on a side.
+
+    :param image_data: Image data
+    :param linear_stretch: boolean whether to use a linear stretch (default is log)
+    :return: Skyportal-compliant PNG image string
+    """
+    with io.BytesIO() as buff:
+        fig = plt.figure()
+        fig.set_size_inches(4, 4, forward=False)
+        ax_1 = plt.Axes(fig, [0.0, 0.0, 1.0, 1.0])
+        ax_1.set_axis_off()
+        fig.add_axes(ax_1)
+
+        # replace nans with median:
+        img = np.array(image_data)
+        # replace dubiously large values
+        xl_mask = np.greater(np.abs(img), 1e20, where=~np.isnan(img))
+        if img[xl_mask].any():
+            img[xl_mask] = np.nan
+        if np.isnan(img).any():
+            median = float(np.nanmean(img.flatten()))
+            img = np.nan_to_num(img, nan=median)
+
+        norm = ImageNormalize(
+            img,
+            stretch=LinearStretch() if linear_stretch else LogStretch(),
+        )
+        img_norm = norm(img)
+        normalizer = AsymmetricPercentileInterval(
+            lower_percentile=1, upper_percentile=100
+        )
+        vmin, vmax = normalizer.get_limits(img_norm)
+        ax_1.imshow(img_norm, cmap="bone", origin="lower", vmin=vmin, vmax=vmax)
+        plt.savefig(buff, dpi=42)
+
+        buff.seek(0)
+        plt.close()
+
+        fritz_thumbnail = base64.b64encode(buff.read()).decode("utf-8")
+
+    return fritz_thumbnail
diff --git a/mirar/processors/sources/__init__.py b/mirar/processors/sources/__init__.py
index 91bf4ea2d..5bcf4d78e 100644
--- a/mirar/processors/sources/__init__.py
+++ b/mirar/processors/sources/__init__.py
@@ -2,13 +2,10 @@
 Central module for candidate detection and extraction.
 """
 from mirar.processors.sources.edge_mask import EdgeSourcesMask
+from mirar.processors.sources.forced_photometry import ForcedPhotometryDetector
 from mirar.processors.sources.namer import CandidateNamer
 from mirar.processors.sources.source_detector import ZOGYSourceDetector
 from mirar.processors.sources.source_exporter import SourceWriter
 from mirar.processors.sources.source_filter import BaseSourceFilter
 from mirar.processors.sources.source_loader import SourceLoader
-from mirar.processors.sources.source_table_builder import (
-    SourceTablefromCoordinates,
-    SourceTablefromHeader,
-)
-from mirar.processors.sources.source_table_modifier import CustomSourceModifier
+from mirar.processors.sources.source_table_modifier import CustomSourceTableModifier
diff --git a/mirar/processors/sources/forced_photometry.py b/mirar/processors/sources/forced_photometry.py
new file mode 100644
index 000000000..65a18a131
--- /dev/null
+++ b/mirar/processors/sources/forced_photometry.py
@@ -0,0 +1,65 @@
+"""
+Module to extract a candidates table from an image header
+"""
+import pandas as pd
+
+from mirar.data import ImageBatch, SourceBatch, SourceTable
+from mirar.data.utils import get_xy_from_wcs
+from mirar.errors import ProcessorError
+from mirar.paths import CAND_DEC_KEY, CAND_RA_KEY, XPOS_KEY, YPOS_KEY
+from mirar.processors.base_processor import BaseSourceGenerator
+
+
+class TableFromHeaderError(ProcessorError):
+    """Error relating to writing a source table from a header"""
+
+
+class HeaderKeyMissingError(ProcessorError):
+    """Error relating to missing keys in headers"""
+
+
+class ForcedPhotometryDetector(BaseSourceGenerator):
+    """
+    Class to create a candidates table for performing forced photometry
+    """
+
+    base_key = "forcedphot"
+
+    def __init__(
+        self,
+        calculate_image_coordinates: bool = True,
+        ra_header_key: str = CAND_RA_KEY,
+        dec_header_key: str = CAND_DEC_KEY,
+    ):
+        super().__init__()
+        self.calculate_image_coordinates = calculate_image_coordinates
+        self.ra_header_key = ra_header_key
+        self.dec_header_key = dec_header_key
+
+    def _apply_to_images(self, batch: ImageBatch) -> SourceBatch:
+        all_cands = SourceBatch()
+        for image in batch:
+            # Save the image here, to facilitate processing downstream
+            header = image.get_header()
+
+            new_dict = {
+                CAND_RA_KEY: image[self.ra_header_key],
+                CAND_DEC_KEY: image[self.dec_header_key],
+            }
+
+            if self.calculate_image_coordinates:
+                x, y = get_xy_from_wcs(
+                    ra_deg=image[self.ra_header_key],
+                    dec_deg=image[self.dec_header_key],
+                    header=header,
+                )
+                new_dict[XPOS_KEY] = x
+                new_dict[YPOS_KEY] = y
+
+            src_table = pd.DataFrame([new_dict])
+
+            metadata = self.get_metadata(image)
+
+            all_cands.append(SourceTable(src_table, metadata=metadata))
+
+        return all_cands
diff --git a/mirar/processors/sources/namer.py b/mirar/processors/sources/namer.py
index 898aa5614..f4dc0c3d6 100644
--- a/mirar/processors/sources/namer.py
+++ b/mirar/processors/sources/namer.py
@@ -9,7 +9,8 @@
 
 from mirar.data import SourceBatch
 from mirar.database.transactions.select import run_select
-from mirar.paths import CAND_NAME_KEY, SOURCE_XMATCH_KEY
+from mirar.paths import CAND_NAME_KEY, SOURCE_XMATCH_KEY, TIME_KEY
+from mirar.processors.base_processor import PrerequisiteError
 from mirar.processors.database import CrossmatchSourceWithDatabase
 from mirar.processors.database.database_selector import BaseDatabaseSourceSelector
 
@@ -30,7 +31,6 @@ def __init__(
         name_start: str = "aaaaa",
         db_name_field: str = CAND_NAME_KEY,
         db_order_field: str = "candid",
-        date_field: str = "jd",
         **kwargs,
     ):
         super().__init__(db_output_columns=[db_name_field], **kwargs)
@@ -38,7 +38,6 @@ def __init__(
         self.db_order_field = db_order_field
         self.base_name = base_name
         self.name_start = name_start
-        self.date_field = date_field
 
     @staticmethod
     def increment_string(string: str):
@@ -77,15 +76,15 @@ def increment_string(string: str):
 
         return new_string
 
-    def get_next_name(self, cand_jd: float, last_name: str = None) -> str:
+    def get_next_name(self, detection_time: Time, last_name: str = None) -> str:
         """
         Function to get a new candidate name
 
-        :param cand_jd: jd of detection
+        :param detection_time: detection time (Astropy Time object)
         :param last_name: last name
         :return: new name
         """
-        cand_year = Time(cand_jd, format="jd").datetime.year % 1000
+        cand_year = detection_time.datetime.year % 1000
         if last_name is None:
             res = run_select(
                 query=select(getattr(self.db_table.sql_model, self.db_name_field))
@@ -124,14 +123,13 @@ def _apply_to_sources(
 
             names = []
             lastname = None
+
+            detection_time = Time(source_table[TIME_KEY])
             for _, source in sources.iterrows():
                 if len(source[SOURCE_XMATCH_KEY]) > 0:
                     source_name = source[SOURCE_XMATCH_KEY][0][self.db_name_field]
-
                 else:
-                    source_name = self.get_next_name(
-                        source[self.date_field], last_name=lastname
-                    )
+                    source_name = self.get_next_name(detection_time, last_name=lastname)
                     lastname = source_name
                 names.append(source_name)
 
@@ -153,4 +151,4 @@ def check_prerequisites(
                 f"However, the following steps were found: {self.preceding_steps}."
             )
             logger.error(err)
-            raise ValueError(err)
+            raise PrerequisiteError(err)
diff --git a/mirar/processors/sources/source_detector.py b/mirar/processors/sources/source_detector.py
index 8a980305b..67d84cc1e 100644
--- a/mirar/processors/sources/source_detector.py
+++ b/mirar/processors/sources/source_detector.py
@@ -9,7 +9,6 @@
 import numpy as np
 import pandas as pd
 from astropy.io import fits
-from astropy.time import Time
 
 from mirar.data import Image, ImageBatch, SourceBatch, SourceTable
 from mirar.data.utils import encode_img
@@ -17,27 +16,17 @@
     BASE_NAME_KEY,
     CAND_DEC_KEY,
     CAND_RA_KEY,
-    DIFF_IMG_KEY,
-    EXPTIME_KEY,
-    FILTER_KEY,
-    JD_KEY,
     LATEST_SAVE_KEY,
     LATEST_WEIGHT_SAVE_KEY,
-    MAGLIM_KEY,
-    NORM_PSFEX_KEY,
     REF_IMG_KEY,
     SCI_IMG_KEY,
     SCOR_IMG_KEY,
-    TIME_KEY,
-    UNC_IMG_KEY,
     XPOS_KEY,
     YPOS_KEY,
-    ZP_KEY,
-    ZP_STD_KEY,
     get_output_dir,
 )
 from mirar.processors.astromatic.sextractor.sourceextractor import run_sextractor_dual
-from mirar.processors.base_processor import BaseSourceGenerator
+from mirar.processors.base_processor import BaseSourceGenerator, PrerequisiteError
 from mirar.processors.photometry.utils import make_cutouts
 from mirar.processors.zogy.zogy import ZOGY
 from mirar.utils.ldac_tools import get_table_from_ldac
@@ -51,9 +40,6 @@ def generate_candidates_table(
     sci_resamp_image_path: str | Path,
     ref_resamp_image_path: str | Path,
     diff_scorr_path: str | Path,
-    diff_psf_path: str | Path,
-    diff_unc_path: str | Path,
-    copy_image_keywords: str | list[str] = None,
 ) -> pd.DataFrame:
     """
     Generate a candidates table from a difference image
@@ -62,9 +48,6 @@ def generate_candidates_table(
     :param sci_resamp_image_path: Path to the resampled science image
     :param ref_resamp_image_path: Path to the resampled reference image
     :param diff_scorr_path: Path to the scorr image
-    :param diff_psf_path: Path to the psf image
-    :param diff_unc_path: Path to the uncertainty image
-    :param copy_image_keywords: Keywords to copy from the image header
     :return: Candidates table
     """
     det_srcs = get_table_from_ldac(scorr_catalog_path)
@@ -120,35 +103,13 @@ def generate_candidates_table(
         display_ref_ims.append(display_ref_bit)
         display_diff_ims.append(display_diff_bit)
 
-    det_srcs["cutoutScience"] = display_sci_ims
-    det_srcs["cutoutTemplate"] = display_ref_ims
-    det_srcs["cutoutDifference"] = display_diff_ims
-
-    # Standard mirar keywords copied from images
-    det_srcs[ZP_KEY] = diff[ZP_KEY]
-    det_srcs[ZP_STD_KEY] = diff[ZP_STD_KEY]
-    det_srcs[DIFF_IMG_KEY] = diff[LATEST_SAVE_KEY]
-    det_srcs[SCI_IMG_KEY] = sci_resamp_image_path
-    det_srcs[REF_IMG_KEY] = ref_resamp_image_path
-    det_srcs[NORM_PSFEX_KEY] = diff_psf_path
-    det_srcs[LATEST_SAVE_KEY] = diff[LATEST_SAVE_KEY]
-    det_srcs[UNC_IMG_KEY] = diff_unc_path
-    det_srcs[JD_KEY.lower()] = Time(fits.getval(sci_resamp_image_path, TIME_KEY)).jd
-    det_srcs[EXPTIME_KEY.lower()] = diff[EXPTIME_KEY]
-    det_srcs[MAGLIM_KEY] = diff[MAGLIM_KEY]
-    det_srcs[FILTER_KEY] = diff[FILTER_KEY]
+    det_srcs["cutout_science"] = display_sci_ims
+    det_srcs["cutout_template"] = display_ref_ims
+    det_srcs["cutout_difference"] = display_diff_ims
 
     # TODO - make isdiffpos a parameter
     det_srcs["isdiffpos"] = True
 
-    logger.info(f"Keywords to copy are {copy_image_keywords}")
-    # Any additional keywords to be copied from the image header
-    if copy_image_keywords is not None:
-        if isinstance(copy_image_keywords, str):
-            copy_image_keywords = [copy_image_keywords]
-        for keyword in copy_image_keywords:
-            det_srcs[keyword.lower()] = fits.getval(diff_path, keyword)
-
     return det_srcs
 
 
@@ -167,7 +128,6 @@ def __init__(
         cand_det_sextractor_nnw: str,
         cand_det_sextractor_params: str,
         output_sub_dir: str = "candidates",
-        copy_image_keywords: str | list[str] = None,
     ):
         super().__init__()
         self.output_sub_dir = output_sub_dir
@@ -175,9 +135,6 @@ def __init__(
         self.cand_det_sextractor_filter = cand_det_sextractor_filter
         self.cand_det_sextractor_nnw = cand_det_sextractor_nnw
         self.cand_det_sextractor_params = cand_det_sextractor_params
-        self.copy_image_keywords = copy_image_keywords
-        if isinstance(copy_image_keywords, str):
-            self.copy_image_keywords = [self.copy_image_keywords]
 
     def __str__(self) -> str:
         return (
@@ -205,10 +162,6 @@ def _apply_to_images(
             diff_image_path = os.path.join(
                 self.get_sub_output_dir(), image[BASE_NAME_KEY]
             )
-            diff_psf_path = os.path.join(
-                self.get_sub_output_dir(), image[NORM_PSFEX_KEY]
-            )
-            diff_unc_path = os.path.join(self.get_sub_output_dir(), image[UNC_IMG_KEY])
 
             scorr_image = self.open_fits(scorr_image_path)
             scorr_mask_path = os.path.join(
@@ -237,9 +190,6 @@ def _apply_to_images(
                 sci_resamp_image_path=sci_image_path,
                 ref_resamp_image_path=ref_image_path,
                 diff_scorr_path=scorr_image_path,
-                diff_psf_path=diff_psf_path,
-                diff_unc_path=diff_unc_path,
-                copy_image_keywords=self.copy_image_keywords,
             )
 
             if len(srcs_table) > 0:
@@ -247,11 +197,7 @@ def _apply_to_images(
                 srcs_table["X_SHAPE"] = x_shape
                 srcs_table["Y_SHAPE"] = y_shape
 
-            metadata = {}
-
-            for key in image.keys():
-                if key != "COMMENT":
-                    metadata[key] = image[key]
+            metadata = self.get_metadata(image)
 
             if len(srcs_table) == 0:
                 msg = f"No sources found in image {image[BASE_NAME_KEY]}"
@@ -267,4 +213,4 @@ def _apply_to_images(
     def check_prerequisites(self):
         check = np.sum([isinstance(x, ZOGY) for x in self.preceding_steps])
         if check != 1:
-            raise ValueError("ZOGY must be run before ZOGYSourceDetector")
+            raise PrerequisiteError("ZOGY must be run before ZOGYSourceDetector")
diff --git a/mirar/processors/sources/source_exporter.py b/mirar/processors/sources/source_exporter.py
index 5ecd2a624..408cd14bb 100644
--- a/mirar/processors/sources/source_exporter.py
+++ b/mirar/processors/sources/source_exporter.py
@@ -7,7 +7,7 @@
 from typing import Optional
 
 from mirar.data import SourceBatch, SourceTable
-from mirar.paths import DIFF_IMG_KEY, base_output_dir, get_output_dir, get_output_path
+from mirar.paths import BASE_NAME_KEY, base_output_dir, get_output_dir, get_output_path
 from mirar.processors.base_processor import BaseSourceProcessor
 
 logger = logging.getLogger(__name__)
@@ -60,11 +60,10 @@ def _apply_to_sources(
         )
         output_dir.mkdir(parents=True, exist_ok=True)
 
-        for source_list in batch:
-            source_table = source_list.get_data()
+        for source_table in batch:
             assert len(source_table) > 0, "Source table is empty"
 
-            old = Path(source_table.loc[0][DIFF_IMG_KEY])
+            old = Path(source_table[BASE_NAME_KEY])
             base_path = old.parent / f"{old.stem}{SOURCE_SUFFIX}"
 
             pkl_path = get_output_path(
@@ -76,6 +75,6 @@ def _apply_to_sources(
 
             logger.debug(f"Writing SourceTable to {pkl_path}")
 
-            save_source_table(source_list, pkl_path)
+            save_source_table(source_table, pkl_path)
 
         return batch
diff --git a/mirar/processors/sources/source_table_builder.py b/mirar/processors/sources/source_table_builder.py
deleted file mode 100644
index aefe59b18..000000000
--- a/mirar/processors/sources/source_table_builder.py
+++ /dev/null
@@ -1,219 +0,0 @@
-"""
-Module to extract a candidates table from an image header
-"""
-from typing import Optional
-
-import numpy as np
-import pandas as pd
-from astropy.io.fits import Header
-
-from mirar.data import ImageBatch, SourceBatch, SourceTable
-from mirar.data.utils import get_xy_from_wcs
-from mirar.errors import ProcessorError
-from mirar.paths import (
-    BASE_NAME_KEY,
-    CAND_DEC_KEY,
-    CAND_NAME_KEY,
-    CAND_RA_KEY,
-    LATEST_SAVE_KEY,
-    NORM_PSFEX_KEY,
-    UNC_IMG_KEY,
-    XPOS_KEY,
-    YPOS_KEY,
-    ZP_KEY,
-    ZP_STD_KEY,
-    base_output_dir,
-    core_fields,
-    get_output_path,
-)
-from mirar.processors.base_processor import BaseSourceGenerator
-
-
-class TableFromHeaderError(ProcessorError):
-    """Error relating to writing a source table from a header"""
-
-
-class HeaderKeyMissingError(ProcessorError):
-    """Error relating to missing keys in headers"""
-
-
-class SourceTablefromHeader(BaseSourceGenerator):
-    """
-    Class to create a source table from an image header with user specified header keys
-    """
-
-    base_key = "header_source_exporter"
-
-    def __init__(
-        self,
-        header_keys_list: list[str],
-        calculate_image_coordinates: bool = False,
-        ra_header_key: str = CAND_RA_KEY,
-        dec_header_key: str = CAND_DEC_KEY,
-        save: bool = True,
-        output_sub_dir: Optional[str] = "sourcetable",
-        output_dir: str = base_output_dir,
-        column_names_list: list[str] = None,
-    ):
-        super().__init__()
-        self.header_keys_list = np.array(header_keys_list)
-        self.calculate_image_coordinates = calculate_image_coordinates
-        self.ra_header_key = ra_header_key
-        self.dec_header_key = dec_header_key
-        self.save = save
-        self.output_sub_dir = output_sub_dir
-        self.output_dir = output_dir
-        self.column_names_list = column_names_list
-        if self.column_names_list is None:
-            self.column_names_list = np.array(header_keys_list)
-
-        assert len(self.column_names_list) == len(self.header_keys_list)
-        if np.logical_and(
-            self.calculate_image_coordinates,
-            np.logical_or(
-                CAND_RA_KEY not in self.column_names_list,
-                CAND_DEC_KEY not in self.column_names_list,
-            ),
-        ):
-            err = (
-                f"The default keys for ra/dec : {CAND_RA_KEY}, {CAND_DEC_KEY} are "
-                f"not present in the header_keys_list. Please provide ra_header_key"
-                f" and dec_header_key, or disable x/y coordinate calculation."
-            )
-            raise TableFromHeaderError(err)
-
-    def append_keys_to_header(self, header: Header):
-        """
-        Append additional keys to the header
-        Args:
-            header: header to append keys to
-
-        Returns:
-            header: Header with appended keys
-        """
-        return header
-
-    def _apply_to_images(self, batch: ImageBatch) -> SourceBatch:
-        all_cands = SourceBatch()
-        for image in batch:
-            # Save the image here, to facilitate processing downstream
-            header = image.get_header()
-            header = self.append_keys_to_header(header)
-            image.set_header(header)
-
-            if self.save:
-                path = get_output_path(
-                    base_name=image[BASE_NAME_KEY],
-                    dir_root=self.output_sub_dir,
-                    sub_dir=self.night_sub_dir,
-                )
-                if not path.parent.exists():
-                    path.parent.mkdir(parents=True)
-                self.save_fits(image, path)
-
-            key_absent = [x not in header.keys() for x in self.header_keys_list]
-
-            if np.sum(key_absent) > 0:
-                err = f"Keys {self.header_keys_list[key_absent]} absent from header"
-                raise HeaderKeyMissingError(err)
-
-            cands_table = pd.DataFrame()
-
-            for ind, key in enumerate(self.header_keys_list):
-                val = header[key]
-                if not isinstance(val, list):
-                    val = [val]
-                cands_table[self.column_names_list[ind]] = val
-
-            if self.calculate_image_coordinates:
-                x, y = get_xy_from_wcs(
-                    ra_deg=list(cands_table[self.ra_header_key]),
-                    dec_deg=list(cands_table[self.dec_header_key]),
-                    header=header,
-                )
-                cands_table[XPOS_KEY] = x
-                cands_table[YPOS_KEY] = y
-
-            metadata = {}
-            for key in core_fields:
-                metadata[key] = image[key]
-
-            all_cands.append(SourceTable(cands_table, metadata=metadata))
-
-        return all_cands
-
-
-class ForcedPhotometryCandidateTable(SourceTablefromHeader):
-    """
-    Class to create a candidates table for performing forced photometry
-    """
-
-    base_key = "forcedphot_table_exporter"
-
-    def __init__(self, ra_header_key: str, dec_header_key: str, name_header_key: str):
-        header_keys_list = [
-            ra_header_key,
-            dec_header_key,
-            name_header_key,
-            LATEST_SAVE_KEY,
-            UNC_IMG_KEY,
-            NORM_PSFEX_KEY,
-            ZP_KEY,
-            ZP_STD_KEY,
-        ]
-        column_names_list = [
-            CAND_RA_KEY,
-            CAND_DEC_KEY,
-            CAND_NAME_KEY,
-            LATEST_SAVE_KEY,
-            UNC_IMG_KEY,
-            NORM_PSFEX_KEY,
-            ZP_KEY,
-            ZP_STD_KEY,
-        ]
-        super().__init__(
-            header_keys_list=header_keys_list,
-            calculate_image_coordinates=True,
-            column_names_list=column_names_list,
-        )
-
-    def append_keys_to_header(self, header: Header):
-        if UNC_IMG_KEY not in header.keys():
-            header[UNC_IMG_KEY] = None
-        if NORM_PSFEX_KEY not in header.keys():
-            header[NORM_PSFEX_KEY] = None
-        return header
-
-
-class SourceTablefromCoordinates(ForcedPhotometryCandidateTable):
-    """
-    Class to create a source table to run forced photometry from user-specified
-    coordinates
-    """
-
-    base_key = "coords_table_exporter"
-
-    def __init__(
-        self,
-        ra_deg: float | list[float],
-        dec_deg: float | list[float],
-        ra_header_key: str = "USERRA",
-        dec_header_key: str = "USERDEC",
-        name_header_key: str = "USERNAME",
-    ):
-        self.ra_deg = ra_deg
-        self.dec_deg = dec_deg
-        super().__init__(
-            ra_header_key=ra_header_key,
-            dec_header_key=dec_header_key,
-            name_header_key=name_header_key,
-        )
-
-    def append_keys_to_header(self, header: Header):
-        header["USERRA"] = self.ra_deg
-        header["USERDEC"] = self.dec_deg
-        if UNC_IMG_KEY not in header.keys():
-            header[UNC_IMG_KEY] = None
-        if NORM_PSFEX_KEY not in header.keys():
-            header[NORM_PSFEX_KEY] = None
-        return header
diff --git a/mirar/processors/sources/source_table_modifier.py b/mirar/processors/sources/source_table_modifier.py
index 0a16089ec..09912a6fc 100644
--- a/mirar/processors/sources/source_table_modifier.py
+++ b/mirar/processors/sources/source_table_modifier.py
@@ -4,22 +4,20 @@
 import logging
 from typing import Callable
 
-import pandas as pd
-
 from mirar.data import SourceBatch
 from mirar.processors.base_processor import BaseSourceProcessor
 
 logger = logging.getLogger(__name__)
 
 
-class CustomSourceModifier(BaseSourceProcessor):
+class CustomSourceTableModifier(BaseSourceProcessor):
     """
     Class to modify a source table based on a function
     """
 
     base_key = "custom_source_modifier"
 
-    def __init__(self, modifier_function: Callable[[pd.DataFrame], pd.DataFrame]):
+    def __init__(self, modifier_function: Callable[[SourceBatch], SourceBatch]):
         super().__init__()
         self.modifier_function = modifier_function
 
@@ -30,13 +28,5 @@ def __str__(self) -> str:
         )
 
     def _apply_to_sources(self, batch: SourceBatch) -> SourceBatch:
-        modified_batch = SourceBatch()
-        for source_list in batch:
-            candidate_table = source_list.get_data()
-            modified_table = self.modifier_function(candidate_table)
-            if len(modified_table) == 0:
-                logger.warning("Modified source table is empty")
-            else:
-                source_list.set_data(modified_table)
-                modified_batch.append(source_list)
+        modified_batch = self.modifier_function(batch)
         return modified_batch
diff --git a/mirar/processors/sources/utils/regions_writer.py b/mirar/processors/sources/utils/regions_writer.py
index 43e79daab..841d550fe 100644
--- a/mirar/processors/sources/utils/regions_writer.py
+++ b/mirar/processors/sources/utils/regions_writer.py
@@ -7,7 +7,7 @@
 from typing import Optional
 
 from mirar.data import SourceBatch
-from mirar.paths import DIFF_IMG_KEY, base_output_dir, get_output_path
+from mirar.paths import BASE_NAME_KEY, base_output_dir, get_output_path
 from mirar.processors.base_processor import BaseSourceProcessor
 
 logger = logging.getLogger(__name__)
@@ -44,28 +44,21 @@ def _apply_to_sources(
         for source_list in batch:
             candidate_table = source_list.get_data()
 
-            started_regions_paths = []
-            for ind in range(len(candidate_table)):
-                row = candidate_table.iloc[ind]
-                regions_basepath = os.path.basename(row[DIFF_IMG_KEY]).replace(
-                    ".fits", ".reg"
-                )
-                regions_path = get_output_path(
-                    regions_basepath,
-                    dir_root=self.output_dir_name,
-                    sub_dir=self.night_sub_dir,
-                    output_dir=self.output_dir,
-                )
+            regions_basepath = os.path.basename(source_list[BASE_NAME_KEY]).replace(
+                ".fits", ".reg"
+            )
+            regions_path = get_output_path(
+                regions_basepath,
+                dir_root=self.output_dir_name,
+                sub_dir=self.night_sub_dir,
+                output_dir=self.output_dir,
+            )
 
-                regions_path.parent.mkdir(parents=True, exist_ok=True)
+            regions_path.parent.mkdir(parents=True, exist_ok=True)
 
-                if regions_path not in started_regions_paths:
-                    logger.debug(f"Writing regions path to {regions_path}")
-                    with open(f"{regions_path}", "w", encoding="utf8") as regions_f:
-                        regions_f.write("image\n")
-                    started_regions_paths.append(regions_path)
-
-                with open(f"{regions_path}", "w", encoding="utf8") as regions_f:
+            with open(f"{regions_path}", "w", encoding="utf8") as regions_f:
+                regions_f.write("image\n")
+                for _, row in candidate_table.iterrows():
                     regions_f.write(
                         f"CIRCLE({row['X_IMAGE']},{row['Y_IMAGE']},"
                         f"{self.region_pix_radius})\n"
diff --git a/mirar/references/base_reference_generator.py b/mirar/references/base_reference_generator.py
index 746aea245..7d1266c12 100644
--- a/mirar/references/base_reference_generator.py
+++ b/mirar/references/base_reference_generator.py
@@ -279,6 +279,11 @@ def _get_reference(self, image: Image) -> (fits.PrimaryHDU, fits.PrimaryHDU):
         :return: ref image HDU, ref weight HDU
         """
         component_image_batch = self.get_component_images(image)
+
+        if not len(component_image_batch) > 0:
+            raise ReferenceGenerationError(
+                "No component images found for reference image"
+            )
         resampler = self.image_resampler_generator()
 
         resampler.set_night(night_sub_dir=self.references_base_subdir_name)
diff --git a/mirar/references/wfcam/wfcam_stack.py b/mirar/references/wfcam/wfcam_stack.py
index bc8f92730..fa2173f7b 100644
--- a/mirar/references/wfcam/wfcam_stack.py
+++ b/mirar/references/wfcam/wfcam_stack.py
@@ -26,7 +26,10 @@
 from mirar.processors.astromatic.swarp import Swarp
 from mirar.processors.base_processor import ImageHandler
 from mirar.processors.photcal import PhotCalibrator
-from mirar.references.base_reference_generator import BaseStackReferenceGenerator
+from mirar.references.base_reference_generator import (
+    BaseStackReferenceGenerator,
+    ReferenceGenerationError,
+)
 from mirar.references.wfcam.wfcam_query import BaseWFCAMQuery
 
 logger = logging.getLogger(__name__)
@@ -123,7 +126,13 @@ def get_component_images(self, image: Image) -> ImageBatch:
         """
         wfau_images = self.wfcam_query.run_query(image)
 
-        wfau_images = self.filter_images(wfau_images)
+        if len(wfau_images) > 0:
+            wfau_images = self.filter_images(wfau_images)
+
+        if len(wfau_images) == 0:
+            raise ReferenceGenerationError(
+                f"No good WFAU images found for {image[BASE_NAME_KEY]}"
+            )
 
         # change BASENAME to gel well with parallel processing
         for ind, ref_img in enumerate(wfau_images):
diff --git a/mirar/utils/pipeline_visualisation.py b/mirar/utils/pipeline_visualisation.py
index fb681ee71..a2997be7c 100644
--- a/mirar/utils/pipeline_visualisation.py
+++ b/mirar/utils/pipeline_visualisation.py
@@ -16,6 +16,7 @@
     BaseProcessor,
     BaseSourceGenerator,
     BaseSourceProcessor,
+    PrerequisiteError,
 )
 
 logger = logging.getLogger(__name__)
@@ -68,7 +69,7 @@ def flowify(processor_list: list[BaseProcessor], output_path: Path):
         elif isinstance(processor, BaseSourceProcessor):
             class_kwargs = {"color": "red"}
         else:
-            raise Exception(f"processor type ({type(processor)} not recognised")
+            raise ValueError(f"processor type ({type(processor)} not recognised")
 
         if i < len(processor_list) - 1:
             arrowprops = {
@@ -144,13 +145,24 @@ def iterate_flowify(
             night=str(datetime.now()).split(" ", maxsplit=1)[0].replace("-", ""),
         )
 
+        logger.info(f"Visualising {pipeline} pipeline")
+
         if config is None:
             config_list = pipe.all_pipeline_configurations.keys()
         else:
             config_list = pipe.selected_configurations
 
         for single_config in config_list:
-            flowify(
-                pipe.set_configuration(single_config),
-                get_save_path(pipeline, single_config),
-            )
+            logger.info(f"Visualising {single_config} configuration")
+            try:
+                flowify(
+                    pipe.set_configuration(single_config),
+                    get_save_path(pipeline, single_config),
+                )
+            except PrerequisiteError as exc:
+                err = (
+                    f"Error for '{pipeline}' pipeline, "
+                    f"'{single_config}' configuration"
+                )
+                logger.error(err)
+                raise PrerequisiteError(err) from exc


Test patch:
diff --git a/mirar/downloader/get_test_data.py b/mirar/downloader/get_test_data.py
index d40008993..963092cd2 100644
--- a/mirar/downloader/get_test_data.py
+++ b/mirar/downloader/get_test_data.py
@@ -16,8 +16,7 @@
     os.path.basename(TEST_DATA_URL.replace(".git", ""))
 )
 
-
-TEST_DATA_TAG = "v0.1.14b"
+TEST_DATA_TAG = "v0.1.16"
 
 COMPLETED_CHECK_BOOL = f"{PACKAGE_NAME}_testdata_check"
 NEED_TEST_DATA = "TESTDATA_CHECK"
diff --git a/tests/test_forced_photometry.py b/tests/test_forced_photometry.py
index 6c0a29a7b..a5e89e0b3 100644
--- a/tests/test_forced_photometry.py
+++ b/tests/test_forced_photometry.py
@@ -5,8 +5,7 @@
 
 from mirar.data import Dataset, ImageBatch
 from mirar.downloader.get_test_data import get_test_data_dir
-from mirar.io import open_fits
-from mirar.paths import LATEST_SAVE_KEY, ZP_KEY, ZP_STD_KEY, get_output_path
+from mirar.paths import ZP_KEY, ZP_STD_KEY
 from mirar.pipelines.wirc.blocks import (
     candidate_photometry,
     export_candidates_from_header,
@@ -26,10 +25,7 @@
 NIGHT_NAME = "20210330"
 
 EXPECTED_HEADER_VALUES = {
-    "MAGAP": 17.01042620967162,
-    "MAGPSF": 16.957491654080247,
-    "MAGERRAP": 0.05580721809850005,
-    "SIGMAPSF": 0.04793450626938275,
+    "TARGNAME": "ZTF21aagppzg",
 }
 
 EXPECTED_DATAFRAME_VALUES = {
@@ -37,7 +33,6 @@
     "magap": [17.014130826817812],
     "sigmapsf": [0.04793450628321317],
     "sigmagap": [0.055862250242806104],
-    "objectid": ["ZTF21aagppzg"],
 }
 
 test_fp_configuration = (
@@ -87,57 +82,33 @@ def test_pipeline(self):
         )
         self.assertEqual(len(res), 1)
 
-        candidates_table = res[0][0].get_data()
-        diff_imgpath = get_output_path(
-            base_name=candidates_table.iloc[0][LATEST_SAVE_KEY],
-            dir_root="photom",
-            sub_dir=NIGHT_NAME,
-        )
+        new = res[0][0]
+
+        source_table = new.get_data()
+        metadata = new.get_metadata()
+
+        print("Got the following values:")
+
+        new_header = "expected_header_values = { \n"
+        for key in EXPECTED_HEADER_VALUES:
+            new_header += f'    "{key}": {metadata[key]}, \n'
+        new_header += "}"
+        print(new_header)
+
+        new_src_table = "expected_dataframe_values = { \n"
+        for key in EXPECTED_DATAFRAME_VALUES:
+            new_src_table += f'    "{key}": {list(source_table[key])}, \n'
+        new_src_table += "}"
+        print(new_src_table)
 
-        _, header = open_fits(diff_imgpath)
         for key, value in EXPECTED_HEADER_VALUES.items():
-            if isinstance(value, float):
-                self.assertAlmostEqual(value, header[key], places=2)
-            elif isinstance(value, int):
-                self.assertEqual(value, header[key])
+            if isinstance(value, str):
+                self.assertEqual(value, metadata[key])
             else:
-                raise TypeError(
-                    f"Type for value ({type(value)} is neither float not int."
-                )
+                self.assertAlmostEqual(value, metadata[key], delta=0.05)
+
+        self.assertEqual(len(source_table), 1)
 
-        self.assertEqual(len(candidates_table), 1)
         for key, value in EXPECTED_DATAFRAME_VALUES.items():
-            if isinstance(value, list):
-                for ind, val in enumerate(value):
-                    self.assertAlmostEqual(
-                        candidates_table.iloc[ind][key], val, delta=0.05
-                    )
-
-
-if __name__ == "__main__":
-    print("Calculating latest forced phot dictionary")
-
-    # Code to generate updated ZP dict of the results change
-
-    new_res, new_errorstack = pipeline.reduce_images(
-        dataset=Dataset(ImageBatch()), catch_all_errors=False
-    )
-    new_candidates_table = new_res[0][0].get_data()
-    new_diff_imgpath = get_output_path(
-        base_name=new_candidates_table.iloc[0][LATEST_SAVE_KEY],
-        dir_root="photom",
-        sub_dir="20210330",
-    )
-    _, new_header = open_fits(new_diff_imgpath)
-
-    NEW_EXP_HEADER = "expected_header_values = { \n"
-    for key in EXPECTED_HEADER_VALUES:
-        NEW_EXP_HEADER += f'    "{key}": {new_header[key]}, \n'
-    NEW_EXP_HEADER += "}"
-    print(NEW_EXP_HEADER)
-
-    NEW_EXP_DATAFRAME = "expected_dataframe_values = { \n"
-    for key in EXPECTED_DATAFRAME_VALUES:
-        NEW_EXP_DATAFRAME += f'    "{key}": {list(new_candidates_table[key])}, \n'
-    NEW_EXP_DATAFRAME += "}"
-    print(NEW_EXP_DATAFRAME)
+            for i, row in source_table.iterrows():
+                self.assertAlmostEqual(row[key], value[i], delta=0.05)
diff --git a/tests/test_pipeline_visualisation.py b/tests/test_pipeline_visualisation.py
index 8e9d6c9e9..8245dcc1c 100644
--- a/tests/test_pipeline_visualisation.py
+++ b/tests/test_pipeline_visualisation.py
@@ -1,5 +1,7 @@
+"""
+Test pipeline visualisation
+"""
 import logging
-import unittest
 
 from mirar.testing import BaseTestCase
 from mirar.utils.pipeline_visualisation import iterate_flowify
@@ -7,7 +9,7 @@
 logger = logging.getLogger(__name__)
 
 
-class TestErrors(BaseTestCase):
+class TestPipelineVisualisation(BaseTestCase):
     def setUp(self):
         self.logger = logging.getLogger(__name__)
         self.logger.setLevel(logging.INFO)
diff --git a/tests/test_sedmv2_stellar_pipeline.py b/tests/test_sedmv2_stellar_pipeline.py
index 8a722ce4d..2334a2932 100644
--- a/tests/test_sedmv2_stellar_pipeline.py
+++ b/tests/test_sedmv2_stellar_pipeline.py
@@ -33,26 +33,25 @@
     "ZP_AUTO_nstars": 39,
 }
 
-
 expected_ap_phot = {
-    "FLUXAP2": -10.440357798826348,
-    "FLUXUNC2": 205.54719177533997,
-    "FLUXAP3": 185.81936010251917,
-    "FLUXUNC3": 310.73759182222466,
-    "MAGAP3": 21.336715504439553,
-    "magerrap3": 1.8965330652977908,
-    "FLUXAP4": 288.53853734268324,
-    "FLUXUNC4": 412.89523151752974,
-    "MAGAP4": 20.85893334004581,
-    "magerrap4": 1.6473637846943183,
-    "FLUXAP5": 164.94609445952742,
-    "FLUXUNC5": 513.6055760247314,
-    "MAGAP5": 21.466087812602073,
-    "magerrap5": 3.425448357780633,
-    "FLUXAP10": 804.3668543646124,
-    "fluxunc10": 1027.5301316831465,
-    "MAGAP10": 19.745807489218134,
-    "magerrap10": 1.4910821986784057,
+    "fluxap2": -13.975273319503032,
+    "fluxuncap2": 205.5743073795116,
+    "fluxap3": 161.0970116411306,
+    "fluxuncap3": 310.76716950607755,
+    "magap3": 21.49172419413229,
+    "sigmagap3": 2.1650909426204,
+    "fluxap4": 302.1728556831727,
+    "fluxuncap4": 412.97618099512533,
+    "magap4": 20.80880428265741,
+    "sigmagap4": 1.5816593932948184,
+    "fluxap5": 102.1520544303209,
+    "fluxuncap5": 513.7579900889148,
+    "magap5": 21.986325141481082,
+    "sigmagap5": 5.48914777382836,
+    "fluxap10": 1184.1255370749232,
+    "fluxuncap10": 1027.4733552630178,
+    "magap10": 19.325948536665393,
+    "sigmagap10": 1.0893624615816915,
 }
 
 test_configuration = (
@@ -99,21 +98,16 @@ def test_pipeline(self):
 
         self.assertEqual(len(res), 29)
 
-        header = res[0][0].get_header()
+        new = res[0][0]
+
+        header = new.get_metadata()
 
         new_exp = "expected_zp = { \n"  # pylint: disable=C0103
-        for header_key in header.keys():
-            if "ZP_" in header_key:
-                new_exp += f'    "{header_key}": {header[header_key]}, \n'
+        for header_key in expected_zp.keys():
+            new_exp += f'    "{header_key}": {header[header_key]}, \n'
         new_exp += "}"
         print(new_exp)
 
-        full_header = "\n FULL HEADER: \n \n"
-        for header_key in header.keys():
-            full_header += f'    "{header_key}": {header[header_key]}, \n'
-        full_header += "\n \n \n"
-        print(full_header)
-
         for key, value in expected_zp.items():
             if isinstance(value, float):
                 self.assertAlmostEqual(value, header[key], places=2)
@@ -124,15 +118,26 @@ def test_pipeline(self):
                     f"Type for value ({type(value)} is neither float not int."
                 )
 
-        for key, value in expected_ap_phot.items():
-            if isinstance(value, float):
-                self.assertAlmostEqual(value, header[key], places=2)
-            elif isinstance(value, int):
-                self.assertEqual(value, header[key])
-            else:
-                raise TypeError(
-                    f"Type for value ({type(value)} is neither float not int."
-                )
+        source_table = new.get_data()
+
+        assert len(source_table) == 1
+
+        new_row = "expected_ap_phot = { \n"  # pylint: disable=C0103
+        for header_key in expected_ap_phot.keys():
+            new_row += f'    "{header_key}": {source_table.iloc[0][header_key]}, \n'
+        new_row += "}"
+        print(new_row)
+
+        for _, row in source_table.iterrows():
+            for key, value in expected_ap_phot.items():
+                if isinstance(value, float):
+                    self.assertAlmostEqual(value, row[key], places=2)
+                elif isinstance(value, int):
+                    self.assertEqual(value, row[key])
+                else:
+                    raise TypeError(
+                        f"Type for value ({type(value)} is neither float not int."
+                    )
 
 
 if __name__ == "__main__":
diff --git a/tests/test_winter_imsub_pipeline.py b/tests/test_winter_imsub_pipeline.py
deleted file mode 100644
index 79dcb61ed..000000000
--- a/tests/test_winter_imsub_pipeline.py
+++ /dev/null
@@ -1,116 +0,0 @@
-"""
-Tests for WINTER image subtraction
-"""
-import logging
-import os
-
-from mirar.data import Dataset, ImageBatch
-from mirar.io import open_raw_image
-from mirar.paths import DIFF_IMG_KEY, get_output_path
-from mirar.pipelines import get_pipeline
-from mirar.testing import BaseTestCase
-
-logger = logging.getLogger(__name__)
-
-NIGHT_NAME = "20230726"
-
-EXPECTED_HEADER_VALUES = {
-    "SCORMEAN": -0.13625905938349483,
-    "SCORMED": -0.13512726465303718,
-    "SCORSTD": 1.3021071256299737,
-}
-EXPECTED_DATAFRAME_VALUES = {
-    "magpsf": [12.173105476828795],
-    "magap": [12.062389584233465],
-}
-
-
-pipeline = get_pipeline(
-    instrument="winter", selected_configurations=["test_imsub"], night="20230726"
-)
-
-logging.basicConfig(level=logging.DEBUG)
-
-
-class TestWinterImsubPipeline(BaseTestCase):
-    """
-    Module for testing winter pipeline
-    """
-
-    def setUp(self):
-        """
-        Function to set up test
-        Returns:
-
-        """
-        self.logger = logging.getLogger(__name__)
-        self.logger.setLevel(logging.INFO)
-        self.check_tokens()
-
-    def check_tokens(self):
-        """If required tokens do not exist raise an error.
-        :raises RuntimeError: If missing token.
-        """
-        if os.environ.get("FRITZ_TOKEN", default="") == "":
-            raise RuntimeError(
-                "No Fritz token. Set environment variable FRITZ_TOKEN to test."
-            )
-        if os.environ.get("KOWALSKI_TOKEN", default="") == "":
-            raise RuntimeError(
-                "No Kowalski token. Set environment variable KOWALSKI_TOKEN to test."
-            )
-
-    def test_pipeline(self):
-        """
-        Test winter pipeline
-        Returns:
-
-        """
-        self.logger.info("\n\n Testing winter imsub pipeline \n\n")
-
-        res, _ = pipeline.reduce_images(Dataset([ImageBatch()]), catch_all_errors=False)
-
-        # Expect two datasets, for two different sub-boards
-        self.assertEqual(len(res), 1)
-
-        candidates_table = res[0][0].get_data()
-        diff_imgpath = get_output_path(
-            base_name=candidates_table.iloc[0][DIFF_IMG_KEY],
-            dir_root="diffs",
-            sub_dir=NIGHT_NAME,
-        )
-
-        diff_image = open_raw_image(diff_imgpath)
-
-        print("New Results WINTER IMSUB:")
-        new_exp = "EXPECTED_HEADER_VALUES = { \n"
-        for header_key in diff_image.header.keys():
-            if header_key in EXPECTED_HEADER_VALUES:
-                new_exp += f'    "{header_key}": {diff_image[header_key]}, \n'
-        new_exp += "}"
-        print(new_exp)
-
-        new_df = "EXPECTED_DATAFRAME_VALUES = { \n"
-        for key in candidates_table.keys():
-            if key in EXPECTED_DATAFRAME_VALUES:
-                new_df += f'    "{key}": {candidates_table[key].tolist()}, \n'
-        new_df += "}"
-        print(new_df)
-
-        for key, value in EXPECTED_HEADER_VALUES.items():
-            if isinstance(value, float):
-                self.assertAlmostEqual(value, diff_image[key], places=2)
-            elif isinstance(value, int):
-                self.assertEqual(value, diff_image[key])
-            else:
-                raise TypeError(
-                    f"Type for value ({type(value)} is neither float not int."
-                )
-
-        self.assertEqual(len(candidates_table), 1)
-        for key, value in EXPECTED_DATAFRAME_VALUES.items():
-            if isinstance(value, list):
-                for ind, val in enumerate(value):
-                    self.assertAlmostEqual(
-                        candidates_table.iloc[ind][key], val, delta=0.05
-                    )
diff --git a/tests/test_winter_pipeline.py b/tests/test_winter_pipeline.py
index 74e584b2a..0d6d725b2 100644
--- a/tests/test_winter_pipeline.py
+++ b/tests/test_winter_pipeline.py
@@ -9,63 +9,43 @@
 
 logger = logging.getLogger(__name__)
 
-expected_zp_list = [
-    {
-        "ZP_2.0": 23.90024757385254,
-        "ZP_2.0_std": 0.06780964881181717,
-        "ZP_2.0_nstars": 411,
-        "ZP_3.0": 24.386281967163086,
-        "ZP_3.0_std": 0.0703275129199028,
-        "ZP_3.0_nstars": 415,
-        "ZP_4.0": 24.547704696655273,
-        "ZP_4.0_std": 0.07585957646369934,
-        "ZP_4.0_nstars": 405,
-        "ZP_5.0": 24.589977264404297,
-        "ZP_5.0_std": 0.0833786204457283,
-        "ZP_5.0_nstars": 411,
-        "ZP_6.0": 24.60162353515625,
-        "ZP_6.0_std": 0.08639319241046906,
-        "ZP_6.0_nstars": 404,
-        "ZP_7.0": 24.604106903076172,
-        "ZP_7.0_std": 0.09461381286382675,
-        "ZP_7.0_nstars": 421,
-        "ZP_8.0": 24.594432830810547,
-        "ZP_8.0_std": 0.09780757129192352,
-        "ZP_8.0_nstars": 419,
-        "ZP_AUTO": 24.599721908569336,
-        "ZP_AUTO_std": 0.08646132051944733,
-        "ZP_AUTO_nstars": 398,
-    },
-    {
-        "ZP_2.0": 24.1755313873291,
-        "ZP_2.0_std": 0.03563622757792473,
-        "ZP_2.0_nstars": 422,
-        "ZP_3.0": 24.661762237548828,
-        "ZP_3.0_std": 0.03949178382754326,
-        "ZP_3.0_nstars": 436,
-        "ZP_4.0": 24.835601806640625,
-        "ZP_4.0_std": 0.04035663977265358,
-        "ZP_4.0_nstars": 414,
-        "ZP_5.0": 24.880395889282227,
-        "ZP_5.0_std": 0.04603949934244156,
-        "ZP_5.0_nstars": 427,
-        "ZP_6.0": 24.89640235900879,
-        "ZP_6.0_std": 0.04970197379589081,
-        "ZP_6.0_nstars": 431,
-        "ZP_7.0": 24.910377502441406,
-        "ZP_7.0_std": 0.05080369487404823,
-        "ZP_7.0_nstars": 432,
-        "ZP_8.0": 24.91377067565918,
-        "ZP_8.0_std": 0.04984834045171738,
-        "ZP_8.0_nstars": 419,
-        "ZP_AUTO": 24.909290313720703,
-        "ZP_AUTO_std": 0.05263081565499306,
-        "ZP_AUTO_nstars": 433,
-    },
-]
+expected_zp = {
+    "ZP_2.0": 23.98996925354004,
+    "ZP_2.0_std": 0.03564070165157318,
+    "ZP_2.0_nstars": 30,
+    "ZP_3.0": 24.477495193481445,
+    "ZP_3.0_std": 0.04281936213374138,
+    "ZP_3.0_nstars": 31,
+    "ZP_4.0": 24.63871955871582,
+    "ZP_4.0_std": 0.04711835831403732,
+    "ZP_4.0_nstars": 32,
+    "ZP_5.0": 24.724422454833984,
+    "ZP_5.0_std": 0.0430118553340435,
+    "ZP_5.0_nstars": 30,
+    "ZP_6.0": 24.78891944885254,
+    "ZP_6.0_std": 0.03930766507983208,
+    "ZP_6.0_nstars": 27,
+    "ZP_7.0": 24.79821014404297,
+    "ZP_7.0_std": 0.04727752134203911,
+    "ZP_7.0_nstars": 30,
+    "ZP_8.0": 24.81728744506836,
+    "ZP_8.0_std": 0.05584121495485306,
+    "ZP_8.0_nstars": 32,
+    "ZP_AUTO": 24.794212341308594,
+    "ZP_AUTO_std": 0.043879956007003784,
+    "ZP_AUTO_nstars": 29,
+    "SCORMEAN": -0.1324857697402311,
+    "SCORMED": -0.1305649672434809,
+    "SCORSTD": 1.3039875768136526,
+}
+expected_dataframe_values = {
+    "magpsf": [12.110608868454948],
+    "magap": [12.020240804368768],
+}
+
 
 pipeline = get_pipeline(
-    instrument="winter", selected_configurations=["test"], night="20230710"
+    instrument="winter", selected_configurations=["test"], night="20230726"
 )
 
 logging.basicConfig(level=logging.DEBUG)
@@ -95,35 +75,45 @@ def test_pipeline(self):
 
         res, _ = pipeline.reduce_images(Dataset([ImageBatch()]), catch_all_errors=False)
 
-        # Expect two datasets, for two different sub-boards
+        # Expect one dataset, for one different sub-boards
         self.assertEqual(len(res[0]), 1)
-        self.assertEqual(len(res[1]), 1)
 
-        headers = [x[0].get_header() for x in res]
-        subdetids = [x["SUBDETID"] for x in headers]
-        headers = [x for _, x in sorted(zip(subdetids, headers))]
+        source_table = res[0][0]
 
         # # Uncomment to print new expected ZP dict
         print("New Results WINTER:")
-        new_exp = "expected_zp_list = [ \n"
-        for ind, header in enumerate(headers):
-            expected_zp = expected_zp_list[ind]
-            new_exp += "{ \n"
-            for header_key in header.keys():
-                if header_key in expected_zp.keys():
-                    new_exp += f'    "{header_key}": {header[header_key]}, \n'
-            new_exp += "}, \n"
-        new_exp += "]"
+        new_exp = "expected_zp = { \n"
+        for header_key in source_table.get_metadata():
+            if header_key in expected_zp:
+                new_exp += f'    "{header_key}": {source_table[header_key]}, \n'
+        new_exp += "}"
         print(new_exp)
 
-        for ind, header in enumerate(headers):
-            expected_zp = expected_zp_list[ind]
-            for key, value in expected_zp.items():
-                if isinstance(value, float):
-                    self.assertAlmostEqual(value, header[key], places=2)
-                elif isinstance(value, int):
-                    self.assertEqual(value, header[key])
-                else:
-                    raise TypeError(
-                        f"Type for value ({type(value)} is neither float not int."
+        new_candidates_table = source_table.get_data()
+
+        new_exp_dataframe = "expected_dataframe_values = { \n"
+        for key in expected_dataframe_values:
+            new_exp_dataframe += f'    "{key}": {list(new_candidates_table[key])}, \n'
+        new_exp_dataframe += "}"
+
+        print(new_exp_dataframe)
+
+        for key, value in expected_zp.items():
+            if isinstance(value, float):
+                self.assertAlmostEqual(value, source_table[key], places=2)
+            elif isinstance(value, int):
+                self.assertEqual(value, source_table[key])
+            else:
+                raise TypeError(
+                    f"Type for value ({type(value)} is neither float not int."
+                )
+
+        candidates_table = source_table.get_data()
+
+        self.assertEqual(len(candidates_table), 1)
+        for key, value in expected_dataframe_values.items():
+            if isinstance(value, list):
+                for ind, val in enumerate(value):
+                    self.assertAlmostEqual(
+                        candidates_table.iloc[ind][key], val, delta=0.05
                     )
diff --git a/tests/test_wirc_imsub_pipeline.py b/tests/test_wirc_imsub_pipeline.py
index 642111676..cd73ef542 100644
--- a/tests/test_wirc_imsub_pipeline.py
+++ b/tests/test_wirc_imsub_pipeline.py
@@ -8,8 +8,6 @@
 
 from mirar.data import Dataset, ImageBatch
 from mirar.downloader.get_test_data import get_test_data_dir
-from mirar.io import open_fits
-from mirar.paths import get_output_path
 from mirar.pipelines.wirc.blocks import candidates, subtract
 from mirar.pipelines.wirc.generator import (
     wirc_reference_image_resampler,
@@ -139,24 +137,20 @@ def test_pipeline(self):
 
         self.assertEqual(len(res), 1)
 
-        candidates_table = res[0][0].get_data()
-        diff_imgpath = get_output_path(
-            base_name=candidates_table.iloc[0]["diffimgname"],
-            dir_root="subtract",
-            sub_dir=NIGHT_NAME,
-        )
+        source_table = res[0][0]
 
-        _, header = open_fits(diff_imgpath)
         for key, value in EXPECTED_HEADER_VALUES.items():
             if isinstance(value, float):
-                self.assertAlmostEqual(value, header[key], places=2)
+                self.assertAlmostEqual(value, source_table[key], places=2)
             elif isinstance(value, int):
-                self.assertEqual(value, header[key])
+                self.assertEqual(value, source_table[key])
             else:
                 raise TypeError(
                     f"Type for value ({type(value)} is neither float not int."
                 )
 
+        candidates_table = source_table.get_data()
+
         self.assertEqual(len(candidates_table), 5)
         for key, value in EXPECTED_DATAFRAME_VALUES.items():
             if isinstance(value, list):
@@ -174,21 +168,17 @@ def test_pipeline(self):
     new_res, new_errorstack = pipeline.reduce_images(
         dataset=Dataset(ImageBatch()), catch_all_errors=False
     )
-    new_candidates_table = new_res[0][0].get_data()
-    new_diff_imgpath = get_output_path(
-        base_name=new_candidates_table.iloc[0]["diffimname"],
-        dir_root="subtract",
-        sub_dir="20210330",
-    )
-    _, new_header = open_fits(new_diff_imgpath)
+    new_source_table = new_res[0][0]
 
     NEW_EXP_HEADER = "expected_header_values = { \n"
-    for header_key in new_header.keys():
+    for header_key in new_source_table.keys():
         if "SCOR" in header_key:
-            NEW_EXP_HEADER += f'    "{header_key}": {new_header[header_key]}, \n'
+            NEW_EXP_HEADER += f'    "{header_key}": {new_source_table[header_key]}, \n'
     NEW_EXP_HEADER += "}"
     print(NEW_EXP_HEADER)
 
+    new_candidates_table = new_source_table.get_data()
+
     NEW_EXP_DATAFRAME = "expected_dataframe_values = { \n"
     for key in EXPECTED_DATAFRAME_VALUES:
         NEW_EXP_DATAFRAME += f'    "{key}": {list(new_candidates_table[key])}, \n'