Custom import
To integrate custom import module, developer has to import our Python library and create simple class with few methods. You can find described "Video Import" example below.
Hint
Look for up-to-date sources on GitHub.
In order to make import module, developer need to code few stages:
-
Read paths and settings
-
Get list of input files / datasets
-
Generate output project folders and meta-json.
-
Process and write images and annotations to output project according to datasets
Installing environment¶
Assuming you have Linux based OS and Docker (nvidia-docker) already
- Create project folder
$ mkdir import_test $ cd import_test
- Clone Supervisely library from github:
$ git clone https://github.com/supervisely/supervisely.git
- Create
src
folder and createmain.py
script test file:
$ mkdir src $ cd src $ echo "print('Hello')" > main.py $ cd ..
- Create
Dockerfile
:
$ nano Dockerfile
and paste next text:
FROM supervisely/base-py:latest ENV PYTHONPATH /workdir:/workdir/supervisely_lib/worker_proto:$PYTHONPATH WORKDIR /workdir/src COPY ./supervisely/supervisely_lib /workdir/supervisely_lib COPY src /workdir/src ENTRYPOINT ["sh", "-c", "python -u main.py"]
- Try to build Docker image:
$ docker build -t "supervisely/custom_video_import:ver_01" .
- Run Docker container:
$ nvidia-docker run --rm -t -i "supervisely/custom_video_import:ver_01"
After run you can see Hello
text in your shell.
Import configuration¶
Before coding see default configuration for this import module:
{ "options": { "skip_frame": 25 }, "res_names": { "project": "test_project" } }
Code¶
All code we will write in main.py
file, created above.
At first - import all required libs:
import os import cv2 import supervisely_lib as sly from supervisely_lib import logger
Then define class VideoImporter
class and write constructor:
class VideoImporter: def __init__(self): # Gets all path's we need task_paths = sly.DtlPaths() self.in_dir = task_paths.data_dir self.out_dir = task_paths.results_dir # Load task settings into variable self.settings = sly.json_load(task_paths.settings_path) # Extract task options that we need self.skip_frame = int(self.settings['options'].get('skip_frame', 25))
Main process
method getting video files names and put into output project dataset structure for each video file.
def process(self): # Getting video files list, filtered by allowed file extensions video_files_list = sly.ImportVideoLister.list_videos(self.in_dir) logger.info('Read videos from source:', extra={'count': len(video_files_list)}) # Generate output project items info (list if items) out_pr = sly.ProjectStructure(self.settings['res_names']['project']) for video_file in video_files_list: file_name, file_ext = os.path.splitext(video_file) src_video_file_path = os.path.join(self.in_dir, video_file) meta_data = { 'src_video_file_path': src_video_file_path, 'image_ext': file_ext # - Required field } out_pr.add_item(file_name, file_name, meta_data) # Generate output project paths and save project-meta information out_pr_fs = sly.ProjectFS(self.out_dir, out_pr) out_pr_fs.make_dirs() res_meta = sly.ProjectMeta() res_meta.to_dir(out_pr_fs.project_path) # Process samples (video files) for sample_info in out_pr_fs: self.process_video_file(sample_info, out_pr)
Method process_video_file
makes next steps:
-
Open video capture
-
Read current frame
-
Save frame as image file
-
Save empty annotation related to image
-
Repeat while frames is not ended
See in code:
def process_video_file(self, sample_info, out_pr): try: video_capture = cv2.VideoCapture(sample_info.ia_data["src_video_file_path"]) # Get total frames count in current video file length = int(video_capture.get(cv2.CAP_PROP_FRAME_COUNT)) # We are to report progress of task over sly.ProgressCounter if we want to observe the progress in web panel. progress = sly.progress_counter_import(out_pr.name, length) # Read frames from video video capture while video_capture.isOpened(): ret, image = video_capture.read() # If frames is out, we need stop process if ret is not True: break # Get current frame number frame_id = int(video_capture.get(cv2.CAP_PROP_POS_FRAMES)) # Skip few frames if need if frame_id % self.skip_frame == 0: # Generate names for files frame_name = "{0}_frame_{1}".format(sample_info.image_name, str(frame_id).zfill(7)) img_frame_path = os.path.join(os.path.dirname(sample_info.img_path), frame_name + '.jpg') json_frame_path = os.path.join(os.path.dirname(sample_info.ann_path), frame_name + '.json') # Write image sly.write_image(img_frame_path, image) # Write empty annotation imsize_wh = image.shape[:2] ann = sly.Annotation.new_with_objects(imsize_wh, []) sly.json_dump(ann.pack(), json_frame_path) progress.iter_done_report() video_capture.release() except Exception as e: exc_str = str(e) logger.warn('Input sample skipped due to error: {}'.format(exc_str), exc_info=True, extra={ 'exc_str': exc_str, 'dataset_name': sample_info.ds_name, 'image_name': sample_info.image_name, })
Finally, you may wrap your main
function as shown below:
def main(): importer = VideoImporter() importer.process() if __name__ == '__main__': sly.main_wrapper('VIDEO_ONLY_IMPORT', main)
Supervisely main_wrapper
function can handle root exceptions and log about it to web UI.
Deploy¶
Build Docker image again with updated src/main.py
script:
$ docker build -t "supervisely/custom_video_import:ver_01" .
Push Docker image:
$ docker push -t "supervisely/custom_video_import:ver_01"
In Supervisely create new import plugin and choose Docker image which you pushed above. Don't forget insert default config also showed above.