Bugfixes from 870

[CBroz1]

Description

This fixes two bugs introduced in 870 that had the potential to generate bad data

• dlc_utils.py:red_led_bisector was failing in case where LEDs 1 and 2 were present, but 3 was NaN.
• dlc_utils.py:Centroid: one-point centroid had issues with partial NaNs, where NaNs were passed as 0.

Using the script below, I verified that no production data was impacted.

Verification Script

import os
import time
from datetime import datetime, timedelta
from pathlib import Path

from spyglass.position.v1.position_dlc_centroid import DLCCentroid as DC
from spyglass.position.v1.position_dlc_centroid import DLCCentroidParams as DCP
from spyglass.position.v1.position_dlc_orient import DLCOrientation as DO
from spyglass.position.v1.position_dlc_orient import DLCOrientationParams as DOP

ANALYSIS_DIR = Path("/stelmo/nwb/analysis")
CHECK_DATE = datetime(2023, 6, 25)


def recent_edit(file_names, directory=ANALYSIS_DIR, check_date=CHECK_DATE):
    """
    Takes a list of file names and checks if they were edited in the last 2 weeks.

    Parameters:
    - file_names: List of file names to check.
    - directory: The common directory where these files are located.

    Returns:
    - A dictionary with file names as keys and boolean values indicating if they were edited in the last 2 weeks.
    """
    directory_path = Path(directory)

    recent = set()
    for file_name in file_names:
        file_path = directory_path / file_name
        if not file_path.exists():
            file_path = directory_path / file_name.split("_")[0] / file_name
            if not file_path.exists():
                print(f"{file_name} not found.")
                continue
        last_modified_time = datetime.fromtimestamp(file_path.stat().st_mtime)
        if last_modified_time > check_date:
            print(f"{file_name} was edited recently.")
            recent.add(file_name)
        else:
            print("")
    return recent


def get_potential_files():
    ret_o = list(
        (DO & 'dlc_orientation_params_name = "bisector_default"').fetch(
            "analysis_file_name"
        )
    )
    centroid_params = [
        {"dlc_centroid_params_name": param["dlc_centroid_params_name"]}
        for param in DCP.fetch(as_dict=True)
        if param["params"]["centroid_method"] == "one_pt_centroid"
    ]
    ret_c = list((DC & centroid_params).fetch("analysis_file_name"))

    return ret_o, ret_c


def get_impacted_centroid_files(file_names):
    impacted_files = []
    for file_name in file_names:
        ck = (DC & {"analysis_file_name": file_name}).fetch1("KEY")
        pre_df = DC()._fetch_pos_df(ck, bodyparts_to_use=["WhiteLED"])
        if not pre_df["WhiteLED"]["y"].isna().any():
            continue
        post_df = (DC & ck).fetch1_dataframe()
        if (post_df["position_y"].eq(0)).any():
            impacted_files.append(file_name)
            print(f"Centroid: {file_name} impacted.")
    return impacted_files


def orient_impact_condition(df):
    one_two_good = df["redLED_L"]["y"].notna() & df["redLED_R"]["y"].notna()
    three_bad = df["redLED_C"]["y"].isna()
    return (one_two_good & three_bad).any()


def get_impacted_orientation_files(file_names):
    impacted_files = []
    for file_name in file_names:
        ok = (DO & {"analysis_file_name": file_name}).fetch1("KEY")
        df = DO()._get_pos_df(ok)
        if orient_impact_condition(df):
            impacted_files.append(file_name)
            print(f"Orientation: {file_name} has NaN")
    return impacted_files


if __name__ == "__main__":
    fi_o, fi_c = get_potential_files()
    recent_o = recent_edit(fi_o)
    recent_c = recent_edit(fi_c)
    impact_o = get_impacted_orientation_files(recent_o)
    impact_c = get_impacted_centroid_files(recent_c)  

Resulting impact vars were empty

Checklist:

• No. This PR should be accompanied by a release: (yes/no/unsure)
• N/a. If release, I have updated the CITATION.cff
• No. This PR makes edits to table definitions: (yes/no)
• N/a. If table edits, I have included an alter snippet for release notes.
• If this PR makes changes to position, I ran the relevant tests locally.
• I have updated the CHANGELOG.md with PR number and description.
• N/a. I have added/edited docs/notebooks to reflect the changes

[edeno]

Not a big deal but instead of computing equality to zero several times, it would probably be more efficient / clear to save it a variable. Also generally floating point number comparison should be computed using np.isclose

Are we assuming that if y_vec is zero in all the same places x_vec is?

[CBroz1]

Sure, I can make those changes

I don't think that assumption is here - I think we're just handling the cases of y_vec being 0 differently - It there's enough info to determine up/down with X, do so. If all ys are 0 (because of bad data, misaligned leds, etc), raise the error.

[edeno]

Just double checking that norm_y should use x_vec? I didn't look at the calculation too carefully.

[CBroz1]

So, this iterrows version has x_vec norm in the first arg of arctan, and -y_vec in the second arg ...

norm = np.array([-y_vec / length, x_vec / length])
orientation.append(np.arctan2(norm[1], norm[0]))

That became this with version, with the same order of args. By maybe my norm vars are mislabeled?

norm_x = (-y_vec / length)[~y_vec.eq(0)]
norm_y = (x_vec / length)[~y_vec.eq(0)]
orient[~y_vec.eq(0)] = np.arctan2(norm_y, norm_x)