Optimise missing sources

CHANGELOG.md

@@ -29,6 +29,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 
 #### Changed
 
+- V2: Optimise missing source calculation to use vectorised numpy operations [#899](https://github.com/askap-vast/vast-pipeline/pull/899)
 - V2: Pin package repos for compatibility with latest Python 3.11 & 3.12 [#879](https://github.com/askap-vast/vast-pipeline/pull/879)
 - V2: Revert back to parallel association upload [#852](https://github.com/askap-vast/vast-pipeline/pull/852)
 - V2: Change worker throttling to use Dask Semaphore [#852](https://github.com/askap-vast/vast-pipeline/pull/852)
@@ -95,6 +96,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Removed Aladin lite display containing RACS and other ASKAP HIPS images [#874](https://github.com/askap-vast/vast-pipeline/pull/874)
 
 #### List of PRs
+- [#899](https://github.com/askap-vast/vast-pipeline/pull/899): feat: V2: Optimise missing source calculation
 - [#893](https://github.com/askap-vast/vast-pipeline/pull/893): feat: V2: Parallelise ingest of measurements
 - [#880](https://github.com/askap-vast/vast-pipeline/pull/880): fix: V2: Ensure external dependencies pass when etxternal websites are down
 - [#879](https://github.com/askap-vast/vast-pipeline/pull/879): fix: V2: Update dependencies to work with latest python 3.11&3.12 and pin them

vast_pipeline/pipeline/forced_extraction.py

@@ -652,8 +652,7 @@ def forced_extraction(
     # | VAST_2118-06A.EPOCH03x.I.fits |  0.165395 | 2019-10-29 10:01:20.500000+00:00 |             319.652 |              -6.2989 |               6.7401 |
     # | VAST_2118-06A.EPOCH02.I.fits  |  0.16323  | 2019-10-30 08:31:20.200000+00:00 |             319.652 |              -6.2989 |               6.7401 |
 
-    # Explode out the img_diff column.
-    extr_df = extr_df.explode("img_diff").reset_index()
+    extr_df = extr_df.reset_index()
     total_to_extract = extr_df.shape[0].compute()
     logger.info("%d forced measurements to extract", total_to_extract)
 

vast_pipeline/pipeline/main.py

@@ -278,22 +278,11 @@ def process_pipeline(self, p_run: Run) -> None:
             unforced_df,
             images_df,
             skyregs_df,
+            self.dm.num_workers,
         )
         del images_df
         del unforced_df
 
-        # Make missing sources into Dask dataframe
-        # NOTE: This would not be necessary if the get_src_skyregion_merged_df
-        # function was improved to use Dask. (See NOTE in parallel_groupby function.)
-        npartitions = calculate_n_partitions(missing_sources_df,
-                                             n_cpu=self.dm.num_workers,
-                                             partition_size_mb=self.config['processing']['max_partition_mb'])
-        missing_sources_df = dd.from_pandas(
-            missing_sources_df,
-            npartitions=npartitions
-        )
-        wait(missing_sources_df)
-
         # STEP #4 New source analysis
         logger.info("Running step #4: new source analysis...")
         new_sources_df = new_sources(

vast_pipeline/pipeline/new_sources.py

@@ -357,9 +357,6 @@ def new_sources(
     # save the index before exploding
     new_sources_df = new_sources_df.reset_index()
 
-    # Explode now to avoid two loops below
-    new_sources_df = new_sources_df.explode('img_diff')
-
     # Merge the respective image information to the df
     new_sources_df = new_sources_df.merge(
         images_df[['datetime']],