485
-486
+ | def assign_chirps(
- assign_data: Dict[str, np.ndarray],
- chirp_df: pd.DataFrame,
- data: Dataset,
-) -> None:
- """Assign chirps to wavetracker tracks.
-
- This function uses the extracted envelope troughs to assign chirps to
- tracks. It computes a cost function that is high when the trough prominence
- is high and the distance to the chirp center is low. For each chirp, the
- track with the highest cost function value is chosen.
-
- Parameters
- ----------
- - `assign_data`: `dict`
- Dictionary containing the data needed for assignment
- - `chirp_df`: `pd.dataframe`
- Dataframe containing the chirp bboxes
- - `data`: `gridtools.datasets.Dataset`
- Dataset object containing the data
- """
- # extract data from assign_data
- peak_prominences = assign_data["proms"]
- peak_distances = assign_data["peaks"]
- peak_times = assign_data["ptimes"]
- chirp_indices = assign_data["cindices"]
- track_ids = assign_data["track_ids"]
-
- # compute cost function.
- # this function is high when the trough prominence is high
- # (-> chirp with high contrast)
- # and when the trough is close to the chirp center as detected by the
- # r-cnn (-> detected chirp is close to the actual chirp)
- cost = peak_prominences / peak_distances**2
-
- # set cost to zero for cases where no peak was found
- cost[np.isnan(cost)] = 0
-
- # for each chirp, choose the track where the cost is highest
- # TODO: to avoid confusion make a cost function where high is good and low
- # is bad. this is more like a "gain function"
- chosen_tracks = []
- chosen_track_times = []
- for idx in np.unique(chirp_indices):
- candidate_tracks = track_ids[chirp_indices == idx]
- candidate_costs = cost[chirp_indices == idx]
- candidate_times = peak_times[chirp_indices == idx]
- chosen_tracks.append(candidate_tracks[np.argmax(candidate_costs)])
- chosen_track_times.append(candidate_times[np.argmax(candidate_costs)])
-
- # store chosen tracks in chirp_df
- chirp_df["assigned_track"] = chosen_tracks
-
- # store chirp time estimated from envelope trough in chirp_df
- chirp_df["envelope_trough_time"] = chosen_track_times
-
- # save chirp_df
- chirp_df.to_csv(data.path / "chirpdetector_bboxes.csv", index=False)
-
- # save old format:
- np.save(data.path / "chirp_ids_rcnn.npy", chosen_tracks)
- np.save(data.path / "chirp_times_rcnn.npy", chosen_track_times)
+546
+547
| def assign_chirps(
+ assign_data: Dict[str, np.ndarray],
+ chirp_df: pd.DataFrame,
+ data: Dataset,
+) -> None:
+ """Assign chirps to wavetracker tracks.
+
+ This function uses the extracted envelope troughs to assign chirps to
+ tracks. It computes a cost function that is high when the trough prominence
+ is high and the distance to the chirp center is low. For each chirp, the
+ track with the highest cost function value is chosen.
+
+ Parameters
+ ----------
+ - `assign_data`: `dict`
+ Dictionary containing the data needed for assignment
+ - `chirp_df`: `pd.dataframe`
+ Dataframe containing the chirp bboxes
+ - `data`: `gridtools.datasets.Dataset`
+ Dataset object containing the data
+ """
+ # extract data from assign_data
+ peak_prominences = assign_data["proms"]
+ peak_distances = assign_data["peaks"]
+ peak_times = assign_data["ptimes"]
+ chirp_indices = assign_data["cindices"]
+ track_ids = assign_data["track_ids"]
+
+ # compute cost function.
+ # this function is high when the trough prominence is high
+ # (-> chirp with high contrast)
+ # and when the trough is close to the chirp center as detected by the
+ # r-cnn (-> detected chirp is close to the actual chirp)
+ cost = peak_prominences / peak_distances**2
+
+ # set cost to zero for cases where no peak was found
+ cost[np.isnan(cost)] = 0
+
+ # for each chirp, choose the track where the cost is highest
+ # TODO: to avoid confusion make a cost function where high is good and low
+ # is bad. this is more like a "gain function"
+ chosen_tracks = []
+ chosen_track_times = []
+ for idx in np.unique(chirp_indices):
+ candidate_tracks = track_ids[chirp_indices == idx]
+ candidate_costs = cost[chirp_indices == idx]
+ candidate_times = peak_times[chirp_indices == idx]
+ chosen_tracks.append(candidate_tracks[np.argmax(candidate_costs)])
+ chosen_track_times.append(candidate_times[np.argmax(candidate_costs)])
+
+ # store chosen tracks in chirp_df
+ chirp_df["assigned_track"] = chosen_tracks
+
+ # store chirp time estimated from envelope trough in chirp_df
+ chirp_df["envelope_trough_time"] = chosen_track_times
+
+ # save chirp_df
+ chirp_df.to_csv(data.path / "chirpdetector_bboxes.csv", index=False)
+
+ # save old format:
+ np.save(data.path / "chirp_ids_rcnn.npy", chosen_tracks)
+ np.save(data.path / "chirp_times_rcnn.npy", chosen_track_times)
|
Parameters
Source code in chirpdetector/assign_chirps.py
- 549
-550
+ | def assign_cli(path: pathlib.Path) -> None:
- """Assign chirps to wavetracker tracks.
-
- this is the command line interface for the assign_chirps function.
-
- Parameters
- ----------
- - `path`: `pathlib.path`
- path to the directory containing the chirpdetector.toml file
- """
- if not path.is_dir():
- msg = f"{path} is not a directory"
- raise ValueError(msg)
-
- if not (path / "chirpdetector.toml").is_file():
- msg = f"{path} does not contain a chirpdetector.toml file"
- raise ValueError(msg)
-
- logger = make_logger(__name__, path / "chirpdetector.log")
- # config = load_config(path / "chirpdetector.toml")
- recs = list(path.iterdir())
- recs = [r for r in recs if r.is_dir()]
- # recs = [path / "subset_2020-03-18-10_34_t0_9320.0_t1_9920.0"]
-
- msg = f"found {len(recs)} recordings in {path}, starting assignment"
- prog.console.log(msg)
- logger.info(msg)
-
- prog.console.rule("starting assignment")
- with prog:
- task = prog.add_task("assigning chirps", total=len(recs))
- for rec in recs:
- msg = f"assigning chirps in {rec}"
- logger.info(msg)
- prog.console.log(msg)
-
- data = load(rec)
- chirp_df = pd.read_csv(rec / "chirpdetector_bboxes.csv")
- assign_data, chirp_df, data = extract_assignment_data(
- data, chirp_df
- )
- assign_chirps(assign_data, chirp_df, data)
- prog.update(task, advance=1)
+591
+592
| def assign_cli(path: pathlib.Path) -> None:
+ """Assign chirps to wavetracker tracks.
+
+ this is the command line interface for the assign_chirps function.
+
+ Parameters
+ ----------
+ - `path`: `pathlib.path`
+ path to the directory containing the chirpdetector.toml file
+ """
+ if not path.is_dir():
+ msg = f"{path} is not a directory"
+ raise ValueError(msg)
+
+ if not (path / "chirpdetector.toml").is_file():
+ msg = f"{path} does not contain a chirpdetector.toml file"
+ raise ValueError(msg)
+
+ logger = make_logger(__name__, path / "chirpdetector.log")
+ # config = load_config(path / "chirpdetector.toml")
+ recs = list(path.iterdir())
+ recs = [r for r in recs if r.is_dir()]
+ # recs = [path / "subset_2020-03-18-10_34_t0_9320.0_t1_9920.0"]
+
+ msg = f"found {len(recs)} recordings in {path}, starting assignment"
+ prog.console.log(msg)
+ logger.info(msg)
+
+ prog.console.rule("starting assignment")
+ with prog:
+ task = prog.add_task("assigning chirps", total=len(recs))
+ for rec in recs:
+ msg = f"assigning chirps in {rec}"
+ logger.info(msg)
+ prog.console.log(msg)
+
+ data = load(rec)
+ chirp_df = pd.read_csv(rec / "chirpdetector_bboxes.csv")
+ assign_data, chirp_df, data = extract_assignment_data(
+ data, chirp_df
+ )
+ assign_chirps(assign_data, chirp_df, data)
+ prog.update(task, advance=1)
|
Returns
Source code in chirpdetector/assign_chirps.py
- 208
+ | def bbox_to_chirptimes(chirp_df: pd.DataFrame) -> pd.DataFrame:
- """Convert chirp bboxes to chirp times.
-
- Parameters
- ----------
- - `chirp_df`: `pd.dataframe`
- dataframe containing the chirp bboxes
-
- Returns
- -------
- - `chirp_df`: `pd.dataframe`
- dataframe containing the chirp bboxes with chirp times.
- """
- chirp_df["chirp_times"] = np.mean(chirp_df[["t1", "t2"]], axis=1)
-
- return chirp_df
+221
| def bbox_to_chirptimes(chirp_df: pd.DataFrame) -> pd.DataFrame:
+ """Convert chirp bboxes to chirp times.
+
+ Parameters
+ ----------
+ - `chirp_df`: `pd.dataframe`
+ dataframe containing the chirp bboxes
+
+ Returns
+ -------
+ - `chirp_df`: `pd.dataframe`
+ dataframe containing the chirp bboxes with chirp times.
+ """
+ chirp_df["chirp_times"] = np.mean(chirp_df[["t1", "t2"]], axis=1)
+
+ return chirp_df
|
Returns
Source code in chirpdetector/assign_chirps.py
- 150
+ | def clean_bboxes(data: Dataset, chirp_df: pd.DataFrame) -> pd.DataFrame:
- """Clean the chirp bboxes.
-
- This is a collection of filters that remove bboxes that
- either overlap, are out of range or otherwise do not make sense.
-
- Parameters
- ----------
- - `data`: `gridtools.datasets.Dataset`
- Dataset object containing the data
- - `chirp_df`: `pd.dataframe`
- Dataframe containing the chirp bboxes
-
- Returns
- -------
- - `chirp_df_tf`: `pd.dataframe`
- Dataframe containing the chirp bboxes that overlap with the range
- """
- # non-max suppression: remove all chirp bboxes that overlap with
- # another more than threshold
- pick_indices = non_max_suppression_fast(chirp_df, 0.5)
- chirp_df_nms = chirp_df.loc[pick_indices, :]
-
- # track filter: remove all chirp bboxes that do not overlap with
- # the range spanned by the min and max of the wavetracker frequency tracks
- minf = np.min(data.track.freqs).astype(float)
- maxf = np.max(data.track.freqs).astype(float)
- # maybe add some more cleaning here, such
- # as removing chirps that are too short or too long
- return track_filter(chirp_df_nms, minf, maxf)
+177
| def clean_bboxes(data: Dataset, chirp_df: pd.DataFrame) -> pd.DataFrame:
+ """Clean the chirp bboxes.
+
+ This is a collection of filters that remove bboxes that
+ either overlap, are out of range or otherwise do not make sense.
+
+ Parameters
+ ----------
+ - `data`: `gridtools.datasets.Dataset`
+ Dataset object containing the data
+ - `chirp_df`: `pd.dataframe`
+ Dataframe containing the chirp bboxes
+
+ Returns
+ -------
+ - `chirp_df_tf`: `pd.dataframe`
+ Dataframe containing the chirp bboxes that overlap with the range
+ """
+ # non-max suppression: remove all chirp bboxes that overlap with
+ # another more than threshold
+ pick_indices = non_max_suppression_fast(chirp_df, 0.5)
+ chirp_df_nms = chirp_df.loc[pick_indices, :]
+
+ # track filter: remove all chirp bboxes that do not overlap with
+ # the range spanned by the min and max of the wavetracker frequency tracks
+ minf = np.min(data.track.freqs).astype(float)
+ maxf = np.max(data.track.freqs).astype(float)
+ # maybe add some more cleaning here, such
+ # as removing chirps that are too short or too long
+ return track_filter(chirp_df_nms, minf, maxf)
|
Returns
Source code in chirpdetector/assign_chirps.py
- 182
+ | def cleanup(chirp_df: pd.DataFrame, data: Dataset) -> pd.DataFrame:
- """Clean the chirp bboxes.
-
- This is a collection of filters that remove bboxes that
- either overlap, are out of range or otherwise do not make sense.
-
- Parameters
- ----------
- - `chirp_df`: `pd.dataframe`
- Dataframe containing the chirp bboxes
- - `data`: `gridtools.datasets.Dataset`
- Dataset object containing the data
-
- Returns
- -------
- - `chirp_df`: `pd.dataframe`
- Dataframe containing the chirp bboxes that overlap with the range
- """
- # first clean the bboxes
- chirp_df = clean_bboxes(data, chirp_df)
- # sort chirps in df by time, i.e. t1
- chirp_df = chirp_df.sort_values(by="t1", ascending=True)
- # compute chirp times, i.e. center of the bbox x axis
- return bbox_to_chirptimes(chirp_df)
+203
| def cleanup(chirp_df: pd.DataFrame, data: Dataset) -> pd.DataFrame:
+ """Clean the chirp bboxes.
+
+ This is a collection of filters that remove bboxes that
+ either overlap, are out of range or otherwise do not make sense.
+
+ Parameters
+ ----------
+ - `chirp_df`: `pd.dataframe`
+ Dataframe containing the chirp bboxes
+ - `data`: `gridtools.datasets.Dataset`
+ Dataset object containing the data
+
+ Returns
+ -------
+ - `chirp_df`: `pd.dataframe`
+ Dataframe containing the chirp bboxes that overlap with the range
+ """
+ # first clean the bboxes
+ chirp_df = clean_bboxes(data, chirp_df)
+ # sort chirps in df by time, i.e. t1
+ chirp_df = chirp_df.sort_values(by="t1", ascending=True)
+ # compute chirp times, i.e. center of the bbox x axis
+ return bbox_to_chirptimes(chirp_df)
|
|
|
|
|
|