model_generator.py

def association_models_generator(
    assoc_df: pd.DataFrame
) -> Iterable[Generator[Association, None, None]]:
    """
    Creates a generator object containing yielded Association objects from
    an input pipeline association dataframe.

    Args:
        assoc_df:
            The dataframe from the pipeline containing the associations between
            measurements and sources.

    Returns:
        An iterable generator object containing the yielded Association objects.
    """
    for i, row in assoc_df.iterrows():
        yield Association(
            meas_id=row['id'],
            source_id=row['source_id'],
            d2d=row['d2d'],
            dr=row['dr'],
        )

def measurement_models_generator(
    meas_df: pd.DataFrame
) -> Iterable[Generator[Measurement, None, None]]:
    """
    Creates a generator object containing yielded Measurement objects from
    an input pipeline measurement dataframe.

    Args:
        meas_df:
            The dataframe from the pipeline containing the measurements of an
            image.

    Returns:
        An iterable generator object containing the yielded Measurement
        objects.
    """
    for i, row in meas_df.iterrows():
        one_m = Measurement()
        for fld in one_m._meta.get_fields():
            if getattr(fld, 'attname', None) and fld.attname in row.index:
                setattr(one_m, fld.attname, row[fld.attname])
        yield one_m

def measurement_pair_models_generator(
    measurement_pairs_df: pd.DataFrame,
) -> Iterable[Generator[MeasurementPair, None, None]]:
    """
    Creates a generator of MeasurementPair objects from an input pipeline
    measurement pair dataframe.

    Args:
        measurement_pairs_df:
            The DataFrame of measurement pairs.

    Returns:
        An iterable of MeasurementPair objects, one for each row of
        `measurement_pairs_df`.
    """
    for i, row in measurement_pairs_df.iterrows():
        yield MeasurementPair(
            source_id=row["source_id"],
            measurement_a_id=row["id_a"],
            measurement_b_id=row["id_b"],
            vs_peak=row["vs_peak"],
            vs_int=row["vs_int"],
            m_peak=row["m_peak"],
            m_int=row["m_int"],
        )

def related_models_generator(
    related_df: pd.DataFrame
) -> Iterable[Generator[RelatedSource, None, None]]:
    """
    Creates a generator object containing yielded Association objects from
    an input pipeline association dataframe.

    Args:
        related_df:
            The dataframe from the pipeline containing the relations between
            sources.

    Returns:
        An iterable generator object containing the yielded Association objects.
    """
    for i, row in related_df.iterrows():
        yield RelatedSource(**row.to_dict())

def source_models_generator(
    src_df: pd.DataFrame, pipeline_run: Run
) -> Iterable[Generator[Source, None, None]]:
    """
    Creates a generator object containing yielded Source objects from
    an input pipeline sources dataframe.

    Args:
        src_df:
            The dataframe from the pipeline containing the measurements of
            an image.
        pipeline_run:
            The pipeline Run object of which the sources are associated with.

    Returns:
        An iterable generator object containing the yielded Source objects.
    """
    for i, row in src_df.iterrows():
        name = (
            f"J{deg2hms(row['wavg_ra'], precision=1)}"
            f"{deg2dms(row['wavg_dec'], precision=0)}"
        ).replace(":", "")
        src = Source()
        src.run_id = pipeline_run.id
        src.name = name
        for fld in src._meta.get_fields():
            if getattr(fld, 'attname', None) and fld.attname in row.index:
                setattr(src, fld.attname, row[fld.attname])

        yield src