Project-MONAI / Project-MONAI/MONAILabel
Create new datastore "localJson" to read from json file the location of images and labels
Open
Nobody has claimed this yet.
- Dominant language
- Python
- Stars
- 891
- Forks
- 269
- Avg merge
- 15h 41m
- Merged PRs (30d)
- 1
Description
Is your feature request related to a problem? Please describe.
Monai core can easily train using a json config with images and labels. Monai label on the other hand need to move files in a certain structure.
It is very hard even impossible to move data on servers managed by admins.
Describe the solution you'd like
It would be great to be able to load from a json which points to images and labels. Moreover you should be able to give it a directory to create the normal datastoer sturcture for any labels to be added to unlabeled images, allow to add images to this directory.
In short keep all existing functionality but be able to load from json as a start. This will allow to easily modify the json (add or remove images) so user can experiment with different data sizes
Describe alternatives you've considered
I have an initial solution here. I copied the local.py / inherited from it so it would be easier to diff with local. I broke some software engineering principals by overwriting some private functions.
I am sure you can come up with an easier way to do it or restructure the code to a parent class and 2 children.
to get this to work, I start the server with flags below to point to the dataset.json and the data root
--studies /data/Challenges/Totalsegmentator_dataset/monailabel \
--conf dataJsonPath /data/Challenges/Totalsegmentator_dataset/monailabel2/dataset.json \
--conf dataJsonRoot /data/Challenges/Totalsegmentator_dataset/monailabel/
then a minor change to radiology/main.py to check on these flags to load the localjson datastore instead of the local datastore
def init_datastore(self) -> Datastore:
dataJsonPath = self.conf.get("dataJsonPath", "")
if dataJsonPath:
from monailabel.datastore.localJson import LocalJsonDatastore
from monailabel.config import settings
dataJsonRoot = self.conf.get("dataJsonRoot", "")
if not dataJsonRoot:
dataJsonRoot = os.path.dirname(dataJsonPath)
print(f" ------------------------------ got dataJsonPath as {dataJsonPath}")
datastore = LocalJsonDatastore(
self.studies,
extensions=settings.MONAI_LABEL_DATASTORE_FILE_EXT,
auto_reload=settings.MONAI_LABEL_DATASTORE_AUTO_RELOAD,
dataJsonRoot=dataJsonRoot, dataJsonPath=dataJsonPath
)
# datastore = self.init_datastore_json(dataJsonPath)
else:
datastore = super().init_datastore()
if self.heuristic_planner:
self.planner.run(datastore)
return datastore
below is my initial implementation to LocalJsonDatastore
# Copyright (c) MONAI Consortium
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# http://www.apache.org/licenses/LICENSE-2.0
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import copy
import fnmatch
import io
import json
import logging
import os
import pathlib
import shutil
import time
from typing import Any, Dict, List, Tuple
from filelock import FileLock
from pydantic import BaseModel
from watchdog.events import PatternMatchingEventHandler
from watchdog.observers import Observer
from monailabel.interfaces.datastore import Datastore, DefaultLabelTag
from monailabel.interfaces.exception import ImageNotFoundException, LabelNotFoundException
from monailabel.utils.others.generic import file_checksum, file_ext, remove_file
from monai.data import load_decathlon_properties, load_decathlon_datalist
logger = logging.getLogger(__name__)
class DataModel2(BaseModel):
ext: str = ""
path: str = ""
info: Dict[str, Any] = {}
class ImageLabelModel2(BaseModel):
image: DataModel2
labels: Dict[str, DataModel2] = {} # tag => label
def tags(self):
return self.labels.keys()
def label_path(self,tag): # aeh added
# return path from json if tag == final
if tag==DefaultLabelTag.FINAL:
lbPath = self.labels[tag].path
elif tag==DefaultLabelTag.ORIGINAL:
lbPath = self.labels[tag].path
return lbPath
class LocalDatastoreModel2(BaseModel):
name: str
description: str
images_dir: str = ""
labels_dir: str = "labels"
objects: Dict[str, ImageLabelModel2] = {}
# will be ignored while saving...
base_path: str = ""
def tags(self):
tags = set()
for v in self.objects.values():
tags.update(v.tags())
return tags
def filter_by_tag(self, tag: str):
return {k: v for k, v in self.objects.items() if v.labels.get(tag)}
def label(self, id: str, tag: str):
obj = self.objects.get(id)
return obj.labels.get(tag) if obj else None
def image_path(self):
return os.path.join(self.base_path, self.images_dir) if self.base_path else self.images_dir
def label_path(self, tag):
path = os.path.join(self.labels_dir, tag) if tag else self.labels_dir
return os.path.join(self.base_path, path) if self.base_path else path
def labels_path(self):
path = self.labels_dir
return {tag: os.path.join(path, tag) if self.base_path else path for tag in self.tags()}
from monailabel.datastore.local import LocalDatastore
class LocalJsonDatastore(LocalDatastore): #Datastore LocalDatastore
"""
Class to represent a datastore local to the MONAI-Label Server
Attributes
----------
`name: str`
The name of the datastore
`description: str`
The description of the datastore
"""
def __init__(
self,
datastore_path: str,
dataJsonRoot, dataJsonPath, dataJsonSection: str = "training",
images_dir: str = ".",
labels_dir: str = "labels",
datastore_config: str = "datastore_v2.json",
extensions=("*.nii.gz", "*.nii"),
auto_reload=False,
):
"""
Creates a `LocalDataset` object
Parameters:
`datastore_path: str`
a string to the directory tree of the desired dataset
`datastore_config: str`
optional file name of the dataset configuration file (by default `dataset.json`)
"""
#self._dataJsonPath = "/data/Challenges/medicaldecathlon/rawUnzipped/Task09_Spleen/dataset_0.json"
#self._dataRoot = "/data/Challenges/medicaldecathlon/rawUnzipped/Task09_Spleen/"
# self._dataJsonSection = "training"
#self._dataJsonSection = "test"
self._dataJsonPath = dataJsonPath
self._dataRoot =dataJsonRoot
self._dataJsonSection = dataJsonSection
self._datastore_path = datastore_path
print(f" --- {datastore_path=}")
self._datastore_config_path = os.path.join(datastore_path, datastore_config)
self._extensions = [extensions] if isinstance(extensions, str) else extensions
self._ignore_event_count = 0
self._ignore_event_config = False
self._config_ts = 0
self._auto_reload = auto_reload
logging.getLogger("filelock").setLevel(logging.ERROR)
logger.info(f"Auto Reload: {auto_reload}; Extensions: {self._extensions}")
os.makedirs(self._datastore_path, exist_ok=True)
self._lock_file = os.path.join(datastore_path, ".lock")
self._datastore: LocalDatastoreModel2 = LocalDatastoreModel2(
name="new-dataset", description="New Dataset", images_dir=images_dir, labels_dir=labels_dir
)
self._datastore.base_path = self._datastore_path
self._init_from_datastore_file(throw_exception=True)
os.makedirs(self._datastore.image_path(), exist_ok=True)
os.makedirs(self._datastore.label_path(None), exist_ok=True)
os.makedirs(self._datastore.label_path(DefaultLabelTag.FINAL), exist_ok=True)
os.makedirs(self._datastore.label_path(DefaultLabelTag.ORIGINAL), exist_ok=True)
# reconcile the loaded datastore file with any existing files in the path
self._reconcile_datastore()
if auto_reload:
logger.info("Start observing external modifications on datastore (AUTO RELOAD)")
# Image Dir
include_patterns = [f"{self._datastore.image_path()}{os.path.sep}{ext}" for ext in [*extensions]]
# Label Dir(s)
label_dirs = self._datastore.labels_path()
label_dirs[DefaultLabelTag.FINAL] = self._datastore.label_path(DefaultLabelTag.FINAL)
label_dirs[DefaultLabelTag.ORIGINAL] = self._datastore.label_path(DefaultLabelTag.ORIGINAL)
for label_dir in label_dirs.values():
include_patterns.extend(f"{label_dir}{os.path.sep}{ext}" for ext in [*extensions])
# Config
include_patterns.append(self._datastore_config_path)
self._handler = PatternMatchingEventHandler(patterns=include_patterns)
self._handler.on_created = self._on_any_event
self._handler.on_deleted = self._on_any_event
self._handler.on_modified = self._on_modify_event
try:
self._ignore_event_count = 0
self._ignore_event_config = False
self._observer = Observer()
self._observer.schedule(self._handler, recursive=True, path=self._datastore_path)
self._observer.start()
except OSError as e:
logger.error(
"Failed to start File watcher. "
"Local datastore will not update if images and labels are moved from datastore location."
)
logger.error(str(e))
# def name(self) -> str:
# """
# Dataset name (if one is assigned)
#
# Returns:
# name (str): Dataset name as string
# """
# return self._datastore.name
#
# def set_name(self, name: str):
# """
# Sets the dataset name in a standardized format (lowercase, no spaces).
#
# Parameters:
# name (str): Desired dataset name
# """
# self._datastore.name = name
# self._update_datastore_file()
#
# def description(self) -> str:
# """
# Gets the description field for the dataset
#
# :return description: str
# """
# return self._datastore.description
#
# def set_description(self, description: str):
# """
# Set a description for the dataset
#
# :param description: str
# """
# self._datastore.description = description
# self._update_datastore_file()
#
# def _to_id(self, file: str) -> Tuple[str, str]:
# ext = file_ext(file)
# extensions = [e.replace("*", "") for e in self._extensions]
# for e in extensions:
# if file.endswith(e):
# ext = e
# id = file.replace(ext, "")
# return id, ext
#
# def _filename(self, id: str, ext: str) -> str:
# return id + ext
#
# def _to_bytes(self, file):
# return io.BytesIO(pathlib.Path(file).read_bytes())
#
def datalist(self, full_path=True) -> List[Dict[str, Any]]:
"""
Return a dictionary of image and label pairs corresponding to the 'image' and 'label'
keys respectively
:return: the {'label': image, 'label': label} pairs for training
"""
tag = DefaultLabelTag.FINAL
image_path = self._datastore.image_path()
label_path = self._datastore.label_path(tag)
ds = []
for k, v in self._datastore.filter_by_tag(tag).items():
ds.append(
{
# "image": os.path.realpath(os.path.join(image_path, self._filename(k, v.image.ext))),
"image": os.path.realpath(v.image.path),
# "label": os.path.realpath(os.path.join(label_path, self._filename(k, v.labels[tag].ext))),
"label": os.path.realpath(v.labels[tag].path),
"meta": {"image": v.image.info, "label": v.labels[tag].info},
}
)
if not full_path:
ds = json.loads(json.dumps(ds).replace(f"{self._datastore_path.rstrip(os.pathsep)}{os.pathsep}", ""))
return ds
#
# def get_image(self, image_id: str) -> Any:
# """
# Retrieve image object based on image id
#
# :param image_id: the desired image's id
# :return: return the "image"
# """
# uri = self.get_image_uri(image_id)
# return self._to_bytes(uri) if uri else None
def get_image_uri(self, image_id: str) -> str:
"""
Retrieve image uri based on image id
:param image_id: the desired image's id
:return: return the image uri
"""
obj = self._datastore.objects.get(image_id)
return str(os.path.realpath(obj.image.path)) if obj else ""
def get_image_info(self, image_id: str) -> Dict[str, Any]:
"""
Get the image information for the given image id
:param image_id: the desired image id
:return: image info as a list of dictionaries Dict[str, Any]
"""
obj = self._datastore.objects.get(image_id)
info = copy.deepcopy(obj.image.info) if obj else {}
if obj:
name = self._filename(image_id, obj.image.ext)
# path = os.path.realpath(os.path.join(self._datastore.image_path(), name))
path = os.path.realpath(obj.image.path)
info["path"] = path
logger.info(f"----------------------------aeh org label --- in get_image_info ------ {info}")
return info
############ TODO Need to add this code / function on load from json file
# def __call__(self, request, datastore: Datastore):
# loader = LoadImage(image_only=True)
# result = {}
# for image_id in datastore.list_images():
# for tag in self.tags:
# label_id: str = datastore.get_label_by_image_id(image_id, tag)
# if label_id:
# label = loader(datastore.get_label_uri(label_id, tag))
# if isinstance(label, torch.Tensor):
# label = label.numpy()
# slices = [sid for sid in range(label.shape[0]) if np.sum(label[sid] > 0)]
# info = {"sum": int(np.sum(label)), "slices": len(slices)}
# logger.debug(f"{label_id} => {info}")
#
# datastore.update_label_info(label_id, tag, info)
# result[label_id] = info
# return result
#
def get_label(self, label_id: str, label_tag: str) -> Any:
"""
Retrieve image object based on label id
:param label_id: the desired label's id
:param label_tag: the matching label's tag
:return: return the "label"
"""
uri = self.get_label_uri(label_id, label_tag)
return self._to_bytes(uri) if uri else None
# def get_label_uri(self, label_id: str, label_tag: str) -> str:
# """
# Retrieve label uri based on image id
#
# :param label_id: the desired label's id
# :param label_tag: the matching label's tag
# :return: return the label uri
# """
# label = self._datastore.label(label_id, label_tag)
# name = self._filename(label_id, label.ext) if label else ""
# return str(os.path.realpath(os.path.join(self._datastore.label_path(label_tag), name))) if label else ""
#
# def get_labels_by_image_id(self, image_id: str) -> Dict[str, str]:
# """
# Retrieve all label ids for the given image id
#
# :param image_id: the desired image's id
# :return: label ids mapped to the appropriate `LabelTag` as Dict[LabelTag, str]
# """
# obj = self._datastore.objects.get(image_id)
# return {tag: image_id for tag in obj.labels} if obj else {}
#
# def get_label_by_image_id(self, image_id: str, tag: str) -> str:
# """
# Retrieve label id for the given image id and tag
#
# :param image_id: the desired image's id
# :param tag: matching tag name
# :return: label id
# """
# return self.get_labels_by_image_id(image_id).get(tag, "")
#
# def get_label_info(self, label_id: str, label_tag: str) -> Dict[str, Any]:
# """
# Get the label information for the given label id
#
# :param label_id: the desired label id
# :param label_tag: the matching label tag
# :return: label info as a list of dictionaries Dict[str, Any]
# """
# label = self._datastore.label(label_id, label_tag)
# info: Dict[str, Any] = label.info if label else {}
# return info
#
# def get_labeled_images(self) -> List[str]:
# """
# Get all images that have a corresponding label
#
# :return: list of image ids List[str]
# """
# return [k for k, v in self._datastore.objects.items() if v.labels.get(DefaultLabelTag.FINAL)]
#
# def get_unlabeled_images(self) -> List[str]:
# """
# Get all images that have no corresponding label
#
# :return: list of image ids List[str]
# """
# return [k for k, v in self._datastore.objects.items() if not v.labels.get(DefaultLabelTag.FINAL)]
#
# def list_images(self) -> List[str]:
# """
# Return list of image ids available in the datastore
#
# :return: list of image ids List[str]
# """
# return list(self._datastore.objects.keys())
#
# def _on_any_event(self, event):
# if self._ignore_event_count:
# logger.debug(f"Ignoring event by count: {self._ignore_event_count} => {event}")
# self._ignore_event_count = max(self._ignore_event_count - 1, 0)
# return
#
# logger.debug(f"Event: {event}")
# self.refresh()
#
# def _on_modify_event(self, event):
# # handle modify events only for config path; rest ignored
# if event.src_path != self._datastore_config_path:
# return
#
# if self._ignore_event_config:
# self._ignore_event_config = False
# return
#
# self._init_from_datastore_file()
#
# def refresh(self):
# """
# Refresh the datastore based on the state of the files on disk
# """
# print(f"----- in refresh")
# self._reconcile_datastore()
#
def add_image(self, image_id: str, image_filename: str, image_info: Dict[str, Any]) -> str:
raise NotImplemented
# id, image_ext = self._to_id(os.path.basename(image_filename))
# if not image_id:
# image_id = id
#
# logger.info(f"Adding Image: {image_id} => {image_filename}")
# name = self._filename(image_id, image_ext)
# dest = os.path.realpath(os.path.join(self._datastore.image_path(), name))
#
# with FileLock(self._lock_file):
# logger.debug("Acquired the lock!")
# shutil.copy(image_filename, dest)
#
# image_info = image_info if image_info else {}
# image_info["ts"] = int(time.time())
# # image_info["checksum"] = file_checksum(dest)
# image_info["name"] = name
#
# self._datastore.objects[image_id] = ImageLabelModel2(image=DataModel2(info=image_info, ext=image_ext,path=dest))
# self._update_datastore_file(lock=False)
# logger.debug("Released the lock!")
# return image_id
#
# def remove_image(self, image_id: str) -> None:
# logger.info(f"Removing Image: {image_id}")
#
# obj = self._datastore.objects.get(image_id)
# if not obj:
# raise ImageNotFoundException(f"Image {image_id} not found")
#
# # Remove all labels
# tags = list(obj.labels.keys())
# for tag in tags:
# self.remove_label(image_id, tag)
#
# # Remove Image
# name = self._filename(image_id, obj.image.ext)
# remove_file(os.path.realpath(os.path.join(self._datastore.image_path(), name)))
#
# if not self._auto_reload:
# self.refresh()
#
def save_label(self, image_id: str, label_filename: str, label_tag: str, label_info: Dict[str, Any]) -> str:
"""
Save a label for the given image id and return the newly saved label's id
:param image_id: the image id for the label
:param label_filename: the path to the label file
:param label_tag: the tag for the label
:param label_info: additional info for the label
:return: the label id for the given label filename
"""
logger.info(f"Saving Label for Image: {image_id}; Tag: {label_tag}; Info: {label_info}")
obj = self._datastore.objects.get(image_id)
if not obj:
raise ImageNotFoundException(f"Image {image_id} not found")
_, label_ext = self._to_id(os.path.basename(label_filename))
label_id = image_id
logger.info(f"Adding Label: {image_id} => {label_tag} => {label_filename}")
label_path = self._datastore.label_path(label_tag)
name = self._filename(image_id, label_ext)
dest = os.path.join(label_path, name)
with FileLock(self._lock_file):
logger.debug("Acquired the lock!")
os.makedirs(label_path, exist_ok=True)
shutil.copy(label_filename, dest)
label_info = label_info if label_info else {}
label_info["ts"] = int(time.time())
label_info["checksum"] = file_checksum(dest)
label_info["name"] = name
obj.labels[label_tag] = DataModel2(info=label_info, ext=label_ext, path=dest)
logger.info(f"Label Info: {label_info}")
self._update_datastore_file(lock=False)
logger.debug("Release the lock!")
return label_id
# def remove_label(self, label_id: str, label_tag: str) -> None:
# logger.info(f"Removing label: {label_id} => {label_tag}")
# remove_file(self.get_label_uri(label_id, label_tag))
#
# if not self._auto_reload:
# self.refresh()
#
# def update_image_info(self, image_id: str, info: Dict[str, Any]) -> None:
# """
# Update (or create a new) info tag for the desired image
#
# :param image_id: the id of the image we want to add/update info
# :param info: a dictionary of custom image information Dict[str, Any]
# """
# obj = self._datastore.objects.get(image_id)
# if not obj:
# raise ImageNotFoundException(f"Image {image_id} not found")
#
# obj.image.info.update(info)
# self._update_datastore_file()
#
# def update_label_info(self, label_id: str, label_tag: str, info: Dict[str, Any]) -> None:
# """
# Update (or create a new) info tag for the desired label
#
# :param label_id: the id of the label we want to add/update info
# :param label_tag: the matching label tag
# :param info: a dictionary of custom label information Dict[str, Any]
# """
# label = self._datastore.label(label_id, label_tag)
# if not label:
# raise LabelNotFoundException(f"Label: {label_id} Tag: {label_tag} not found")
#
# label.info.update(info)
# self._update_datastore_file()
#
def _list_files(self, path, patterns):
## don't want to call this should call _list_json_files instead
raise NotImplemented
# files = os.listdir(path)
#
# filtered = dict()
# for pattern in patterns:
# matching = fnmatch.filter(files, pattern)
# for file in matching:
# filtered[os.path.basename(file)] = file
# return filtered
def _list_json_files(self, jsonKey, LabelTag=""):
logger.info(f" ------------------------ in _list_json_files {jsonKey=}")
filtered = dict()
logger.info(f"------------loading {self._dataJsonPath=} -------- {self._dataJsonSection=} ------- {self._dataRoot=}")
dsLst =load_decathlon_datalist(self._dataJsonPath , True, self._dataJsonSection,base_dir= self._dataRoot )
for imgNLabel in dsLst:
if jsonKey in imgNLabel:
imgPath=imgNLabel[jsonKey]
# labelPath=imgNLabel["label"]
filtered[os.path.basename(imgPath)] = imgPath
else:
print(f"-->>>>>>>>>>>>>>>>>>>>> tag {jsonKey} is not is json dict")
return filtered
# def _reconcile_datastore(self):
# logger.debug("reconcile datastore...")
# invalidate = 0
# invalidate += self._remove_non_existing()
# invalidate += self._add_non_existing_images()
#
# labels_dir = self._datastore.label_path(None)
# logger.debug(f"Labels Dir {labels_dir}")
#
# tags = [f for f in os.listdir(labels_dir) if os.path.isdir(os.path.join(labels_dir, f))]
# logger.debug(f"Label Tags: {tags}")
# for tag in tags:
# invalidate += self._add_non_existing_labels(tag)
#
# invalidate += self._remove_non_existing()
#
# logger.info(f"Invalidate count: {invalidate}")
# if invalidate:
# logger.debug("Save datastore file to disk")
# self._update_datastore_file()
# else:
# logger.debug("No changes needed to flush to disk")
#
def _add_non_existing_images(self) -> int:
invalidate = 0
self._init_from_datastore_file()
# local_images = self._list_files(self._datastore.image_path(), self._extensions,jsonKey="image")
local_images = self._list_json_files(jsonKey="image")
image_ids = list(self._datastore.objects.keys())
# for image_file in local_images:
for image_file, image_path in local_images.items():
image_id, image_ext = self._to_id(image_file)
if image_id not in image_ids:
logger.info(f"Adding New Image: {image_id} => {image_file} ==>{image_path}")
name = self._filename(image_id, image_ext)
image_info = {
"ts": int(time.time()),
# "checksum": file_checksum(os.path.join(self._datastore.image_path(), name)),
"name": name,
}
invalidate += 1
self._datastore.objects[image_id] = ImageLabelModel2(image=DataModel2(info=image_info, ext=image_ext,path=image_path))
return invalidate
def _add_non_existing_labels(self, tag) -> int:
invalidate = 0
self._init_from_datastore_file()
if tag == DefaultLabelTag.ORIGINAL:
### TODO AEH hack can do 1 or the other
#### either main behavoure: orignal loads the inference
#### or local the orignal from the json file
local_labels = self._list_json_files(jsonKey="orignal", LabelTag=tag)
logger.info(f"--------------- aeh_org {local_labels=}")
# local_labels = self._list_files(self._datastore.label_path(tag), self._extensions)
else:
local_labels = self._list_json_files(jsonKey="label", LabelTag=tag)
image_ids = list(self._datastore.objects.keys())
# for label_file in local_labels:
for label_file, label_path in local_labels.items():
label_id, label_ext = self._to_id(label_file)
obj = self._datastore.objects.get(label_id)
if not obj or label_id not in image_ids:
logger.warning(f"IGNORE:: No matching image exists for '{label_id}' to add [{label_file}]")
continue
if not obj.labels.get(tag):
logger.info(f"Adding New Label: {tag} => {label_id} => {label_file}")
name = self._filename(label_id, label_ext)
label_info = {
"ts": int(time.time()),
# "checksum": file_checksum(os.path.join(self._datastore.label_path(tag), name)),
"checksum": file_checksum(label_path),
"name": name,
}
self._datastore.objects[label_id].labels[tag] = DataModel2(info=label_info, ext=label_ext,path=label_path)
invalidate += 1
return invalidate
def _remove_non_existing(self) -> int:
invalidate = 0
self._init_from_datastore_file()
objects: Dict[str, ImageLabelModel2] = {}
for image_id, obj in self._datastore.objects.items():
name = self._filename(image_id, obj.image.ext)
# print(f" aeh {image_id=} , {type(obj)=} details {obj=}")
# if not os.path.exists(os.path.realpath(os.path.join(self._datastore.image_path(), name))):
if not os.path.exists(os.path.realpath(obj.image.path)):
logger.info(f"Removing non existing Image Id: {image_id} with path {obj.image.path}")
invalidate += 1
else:
labels: Dict[str, DataModel2] = {}
for tag, label in obj.labels.items():
name = self._filename(image_id, label.ext)
if tag == DefaultLabelTag.ORIGINAL:
lbPath=os.path.join(self._datastore.label_path(tag), name)
logger.info(f"-----------aeh_org {lbPath=}")
else:
lbPath= label.path
if not os.path.exists(os.path.realpath(lbPath)):
logger.info(f"Removing non existing Label Id: '{image_id}' for '{tag}'")
invalidate += 1
else:
labels[tag] = label
obj.labels.clear()
obj.labels.update(labels)
objects[image_id] = obj
self._datastore.objects.clear()
self._datastore.objects.update(objects)
return invalidate
def _init_from_datastore_file(self, throw_exception=False):
try:
with FileLock(self._lock_file):
logger.debug("Acquired the lock!")
if os.path.exists(self._datastore_config_path):
ts = os.stat(self._datastore_config_path).st_mtime
if self._config_ts != ts:
logger.debug(f"Reload Datastore; old ts: {self._config_ts}; new ts: {ts}")
self._datastore = LocalDatastoreModel2.parse_file(self._datastore_config_path)
self._datastore.base_path = self._datastore_path
self._config_ts = ts
logger.debug("Release the Lock...")
except ValueError as e:
logger.error(f"+++ Failed to load datastore => {e}")
if throw_exception:
raise e
# def _update_datastore_file(self, lock=True):
# def _write_to_file():
# logger.debug("+++ Datastore is updated...")
# self._ignore_event_config = True
# with open(self._datastore_config_path, "w") as f:
# f.write(json.dumps(self._datastore.dict(exclude={"base_path"}), indent=2, default=str))
# self._config_ts = os.stat(self._datastore_config_path).st_mtime
#
# if lock:
# with FileLock(self._lock_file):
# logger.debug("Acquired the Lock...")
# _write_to_file()
# logger.debug("Released the Lock...")
# else:
# _write_to_file()
#
# def status(self) -> Dict[str, Any]:
# tags: dict = {}
# for obj in self._datastore.objects.values():
# for tag, _ in obj.labels.items():
# tags[tag] = tags.get(tag, 0) + 1
#
# return {
# "total": len(self.list_images()),
# "completed": len(self.get_labeled_images()),
# "label_tags": tags,
# }
#
# def json(self):
# return self._datastore.dict(exclude={"base_path"})
Contributor guide
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
Research direction
Start by comparing the proposed LocalJsonDatastore with monailabel/datastore/local.py and the datastore initialization in radiology/main.py. Review the dataset.json format and the dataJsonPath, dataJsonRoot, and dataJsonSection settings. Done means JSON-referenced images and labels load while the existing datastore behavior for adding and managing labels remains available.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python
- Domain
- backend, data
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 30/100