diff --git a/client/platform/web-girder/api/waitForFolderDatasetReady.spec.ts b/client/platform/web-girder/api/waitForFolderDatasetReady.spec.ts new file mode 100644 index 000000000..daea0e606 --- /dev/null +++ b/client/platform/web-girder/api/waitForFolderDatasetReady.spec.ts @@ -0,0 +1,69 @@ +import { + beforeEach, describe, expect, it, vi, +} from 'vitest'; + +import girderRest from 'platform/web-girder/plugins/girder'; +import { getFolder } from './girder.service'; +import { waitForFolderDatasetReady } from './waitForFolderDatasetReady'; + +vi.mock('platform/web-girder/plugins/girder', () => ({ + default: { get: vi.fn() }, +})); +vi.mock('./girder.service', () => ({ + getFolder: vi.fn(), + getItemsInFolder: vi.fn(), +})); + +const RUNNING = 2; +const SUCCESS = 3; + +/** A job that reports one more unit of progress on every poll, reaching `total`. */ +function progressingJob(total: number) { + let current = 0; + return () => { + current = Math.min(current + 1, total); + return { + data: { + status: current < total ? RUNNING : SUCCESS, + progress: { current, total }, + }, + }; + }; +} + +describe('waitForFolderDatasetReady', () => { + beforeEach(() => { + vi.mocked(girderRest.get).mockReset(); + vi.mocked(getFolder).mockReset(); + }); + + it('keeps waiting past the timeout while a job is still making progress', async () => { + const job = progressingJob(40); + let polls = 0; + vi.mocked(girderRest.get).mockImplementation(async () => job()); + // Not ready until the job has run for far longer than timeoutMs. + vi.mocked(getFolder).mockImplementation(async () => { + polls += 1; + return { data: { meta: { annotate: polls > 40 } } } as never; + }); + + await expect(waitForFolderDatasetReady( + 'folder', + { pollIntervalMs: 5, timeoutMs: 50 }, + ['job'], + )).resolves.toBeUndefined(); + }); + + it('times out when the job stops making progress', async () => { + vi.mocked(girderRest.get).mockResolvedValue({ + data: { status: RUNNING, progress: { current: 3, total: 10 } }, + }); + vi.mocked(getFolder).mockResolvedValue({ data: { meta: {} } } as never); + + await expect(waitForFolderDatasetReady( + 'folder', + { pollIntervalMs: 5, timeoutMs: 50 }, + ['job'], + )).rejects.toThrow(/Timed out/); + }); +}); diff --git a/client/platform/web-girder/api/waitForFolderDatasetReady.ts b/client/platform/web-girder/api/waitForFolderDatasetReady.ts index 45843d8db..025565dc9 100644 --- a/client/platform/web-girder/api/waitForFolderDatasetReady.ts +++ b/client/platform/web-girder/api/waitForFolderDatasetReady.ts @@ -61,6 +61,7 @@ export async function waitForFolderDatasetReady( folderId: string, options?: { pollIntervalMs?: number; + /** Give up after this long without job progress. */ timeoutMs?: number; /** Called with average job completion fraction in [0, 1] when jobs report progress. */ onProgress?: (fraction: number) => void; @@ -73,7 +74,8 @@ export async function waitForFolderDatasetReady( ): Promise { const pollIntervalMs = options?.pollIntervalMs ?? 1000; const timeoutMs = options?.timeoutMs ?? 10 * 60 * 1000; - const deadline = Date.now() + timeoutMs; + let deadline = Date.now() + timeoutMs; + let lastProgress = ''; async function folderReady(): Promise { const { data: folder } = await getFolder(folderId); @@ -108,6 +110,11 @@ export async function waitForFolderDatasetReady( return job; }), ); + const progress = JSON.stringify(jobs.map((job) => [job.status, job.progress?.current])); + if (progress !== lastProgress) { + lastProgress = progress; + deadline = Date.now() + timeoutMs; + } if (options?.onProgress) { const jobsWithProgress = jobs.filter((job) => job.progress?.total); if (jobsWithProgress.length) { diff --git a/server/dive_server/__init__.py b/server/dive_server/__init__.py index 0cda82a9c..7f635fcf4 100644 --- a/server/dive_server/__init__.py +++ b/server/dive_server/__init__.py @@ -6,9 +6,11 @@ from girder.constants import AccessType from girder.models.user import User from girder.plugin import getPlugin +from girder.settings import SettingDefault from girder.utility import mail_utils from girder.utility.model_importer import ModelImporter from girder_jobs.models.job import Job +from girder_large_image.constants import PluginSettings as LargeImageSettings from dive_utils import constants @@ -42,6 +44,8 @@ def load(self, info): info["apiRoot"].dive_scoring = ScoringResource("dive_scoring") # required because girder doesn't load plugins in order so we need to manually load first. getPlugin('jobs').load(info) + # Tile in a postprocess job, not per file on upload; an admin setting still wins. + SettingDefault.defaults[LargeImageSettings.LARGE_IMAGE_AUTO_SET] = False # Setup route additions for exsting resources info['apiRoot'].job.route("GET", ("queued",), countJobs) info["apiRoot"].user.route("PUT", (":id", "use_private_queue"), use_private_queue) diff --git a/server/dive_server/crud_rpc.py b/server/dive_server/crud_rpc.py index fede3918c..1734ea816 100644 --- a/server/dive_server/crud_rpc.py +++ b/server/dive_server/crud_rpc.py @@ -1497,7 +1497,7 @@ def process_items( user: types.GirderUserModel, additive=False, additivePrepend='', - set='', + annotation_set='', configuration_plan=None, ): """ @@ -1611,7 +1611,7 @@ def process_items( upsert_groups=results['annotations']['groups'].values(), overwrite=True, description=f'Import {results["type"].name} from {file["name"]}', - set=set, + set=annotation_set, ) if results['attributes']: crud.saveImportAttributes(folder, results['attributes'], user) @@ -1646,7 +1646,7 @@ def postprocess( skipTranscoding=False, additive=False, additivePrepend='', - set='', + annotation_set='', stitchedSide='', jobDatasetId='', ) -> dict: @@ -1657,7 +1657,7 @@ def postprocess( skipTranscoding, additive, additivePrepend, - set, + annotation_set, stitchedSide, jobDatasetId, ) @@ -1670,7 +1670,7 @@ def _postprocess( skipTranscoding=False, additive=False, additivePrepend='', - set='', + annotation_set='', stitchedSide='', jobDatasetId='', ) -> dict: @@ -1680,6 +1680,7 @@ def _postprocess( When skipJobs=False, the following may run as jobs: Transcoding of Video Transcoding of Images + Tile creation for large images (TIFF, NITF, ...) Splitting of stitched stereo media (stitchedSide), in place of transcoding Conversion of KPF annotations into track JSON Extraction and upload of zip files @@ -1864,6 +1865,13 @@ def _postprocess( largeImageItems = Folder().childItems( dsFolder, filters={"lowerName": {"$regex": constants.largeImageRegEx}} ) + untiledLargeImageItems = Folder().childItems( + dsFolder, + filters={ + "lowerName": {"$regex": constants.largeImageRegEx}, + "largeImage": {"$exists": False}, + }, + ) if stitchedSide: pass @@ -1895,6 +1903,35 @@ def _postprocess( ) created_job_ids.append(job['_id']) + elif untiledLargeImageItems.count() > 0: + # auto_set is off, so large images get their tiles here. + tiles_params = { + 'user_id': str(user["_id"]), + 'user_login': str(user["login"]), + 'input_folder': str(dsFolder["_id"]), + } + newjob = tasks.create_large_image_tiles.apply_async( + queue=_get_queue_name(user), + kwargs=dict( + folderId=str(dsFolder["_id"]), + user_id=str(user["_id"]), + user_login=str(user["login"]), + girder_client_token=str(token["_id"]), + girder_job_title=f"Preparing {dsFolder['name']} large images for viewing", + girder_job_type="private" if job_is_private else "convert", + ), + ) + job = _persist_async_job_metadata( + newjob, + **{ + constants.JOBCONST_PRIVATE_QUEUE: job_is_private, + constants.JOBCONST_DATASET_ID: job_dataset_id, + constants.JOBCONST_PARAMS: tiles_params, + constants.JOBCONST_CREATOR: str(user['_id']), + }, + ) + created_job_ids.append(job['_id']) + elif imageItems.count() > 0 or largeImageItems.count() > 0: # Safe/web images need annotate now; convert_images sets it when it runs. crud.refresh_folder_document(dsFolder) @@ -1935,7 +1972,7 @@ def _postprocess( user, additive, additivePrepend, - set, + annotation_set, configuration_plan=configuration_plan, ) # Image sequences start at fps=-1 (auto). CSV import may have set a value; @@ -1965,7 +2002,7 @@ def _postprocess( user_login=str(user["login"]), additive=additive, additivePrepend=additivePrepend, - set=set, + set=annotation_set, girder_client_token=str(rle_job_token["_id"]), girder_job_title=( f"Importing COCO RLE masks for {dsFolder['name']}" diff --git a/server/dive_server/views_rpc.py b/server/dive_server/views_rpc.py index 92cef0339..c6e1a86a7 100644 --- a/server/dive_server/views_rpc.py +++ b/server/dive_server/views_rpc.py @@ -271,6 +271,8 @@ def postprocess( stitchedSide, jobDatasetId, ): + # Keep Girder form param name ``set``; CRUD uses annotation_set to avoid + # shadowing the builtin. return crud_rpc.postprocess( self.getCurrentUser(), folder, diff --git a/server/dive_tasks/convert_images.py b/server/dive_tasks/convert_images.py index e52619c5c..dafb488f3 100644 --- a/server/dive_tasks/convert_images.py +++ b/server/dive_tasks/convert_images.py @@ -215,6 +215,28 @@ def convert_images(self: Task, folderId, user_id: str, user_login: str): ) +def _ensure_large_image(gc: GirderClient, manager: JobManager, item: GirderModel) -> None: + """Create the item's large-image tile metadata unless it already has it.""" + try: + gc.get(f'item/{item["_id"]}/tiles') + manager.write(f'Skipping {item["name"]}, already a large image\n') + return + except HttpError as e: + # Safely parse JSON if possible + message = "" + try: + message = e.response.json().get("message", "") + except Exception: + pass # non-JSON response, leave message empty + # This is the Girder message when no large image exists + if e.status == 400 and message == "No large image file in this item.": + manager.write(f'Converting {item["name"]} to large image\n') + gc.post(f'item/{item["_id"]}/tiles') + else: + # Re-raise unexpected errors to fail the job + raise + + @app.task(bind=True, acks_late=True) def convert_large_images(self: Task, folderId, user_id: str, user_login: str): """ @@ -236,31 +258,37 @@ def convert_large_images(self: Task, folderId, user_id: str, user_login: str): ] for item in items_to_convert: # Assumes 1 file per item - try: - # Does it already have tiles? - gc.get(f'item/{item["_id"]}/tiles') - manager.write(f'Skipping {item["name"]}, already a large image\n') - continue - except HttpError as e: - # Safely parse JSON if possible - message = "" - try: - message = e.response.json().get("message", "") - except Exception: - pass # non-JSON response, leave message empty - # This is the Girder message when no large image exists - if e.status == 400 and message == "No large image file in this item.": - manager.write(f'Converting {item["name"]} to large image\n') - gc.post(f'item/{item["_id"]}/tiles') - else: - # Re-raise unexpected errors to fail the job - raise + _ensure_large_image(gc, manager, item) gc.addMetadataToFolder( str(folderId), {"type": constants.LargeImageType}, # mark the parent folder as able to annotate. ) +@app.task(bind=True, acks_late=True, ignore_result=True) +def create_large_image_tiles(self: Task, folderId, user_id: str, user_login: str): + """Create tiles for every large-image file in the folder, then mark it a dataset.""" + context: dict = {} + gc: GirderClient = self.girder_client + manager: JobManager = patch_manager(self.job_manager) + if utils.check_canceled(self, context): + manager.updateStatus(JobStatus.CANCELED) + return + + items = [ + item for item in gc.listItem(folderId) if constants.largeImageRegEx.search(item["name"]) + ] + for index, item in enumerate(items): + if utils.check_canceled(self, context, force=False): + manager.updateStatus(JobStatus.CANCELED) + return + manager.updateProgress(total=len(items), current=index) + # Assumes 1 file per item + _ensure_large_image(gc, manager, item) + manager.updateProgress(total=len(items), current=len(items), forceFlush=True) + gc.addMetadataToFolder(str(folderId), {constants.DatasetMarker: True}) + + @app.task(bind=True, acks_late=True, ignore_result=True) def extract_zip( self: Task, diff --git a/server/dive_tasks/tasks.py b/server/dive_tasks/tasks.py index d35bdd6af..97d68547a 100644 --- a/server/dive_tasks/tasks.py +++ b/server/dive_tasks/tasks.py @@ -9,6 +9,7 @@ convert_calibration, convert_images, convert_large_images, + create_large_image_tiles, extract_zip, ) from dive_tasks.convert_video import convert_video, resolve_annotation_fps @@ -42,6 +43,7 @@ 'convert_images', 'convert_large_images', 'convert_video', + 'create_large_image_tiles', 'download_google_drive_zip', 'export_trained_pipeline', 'extract_zip',