Produce images read from a list of zip files.
| 8 | |
| 9 | |
| 10 | class ImageDataFromZIPFile(RNGDataFlow): |
| 11 | """ Produce images read from a list of zip files. """ |
| 12 | def __init__(self, zip_file, shuffle=False): |
| 13 | """ |
| 14 | Args: |
| 15 | zip_file (list): list of zip file paths. |
| 16 | """ |
| 17 | assert os.path.isfile(zip_file) |
| 18 | self._file = zip_file |
| 19 | self.shuffle = shuffle |
| 20 | self.open() |
| 21 | |
| 22 | def open(self): |
| 23 | self.archivefiles = [] |
| 24 | archive = zipfile.ZipFile(self._file) |
| 25 | imagesInArchive = archive.namelist() |
| 26 | for img_name in imagesInArchive: |
| 27 | if img_name.endswith('.jpg'): |
| 28 | self.archivefiles.append((archive, img_name)) |
| 29 | |
| 30 | def reset_state(self): |
| 31 | super(ImageDataFromZIPFile, self).reset_state() |
| 32 | # Seems necessary to reopen the zip file in forked processes. |
| 33 | self.open() |
| 34 | |
| 35 | def size(self): |
| 36 | return len(self.archivefiles) |
| 37 | |
| 38 | def __iter__(self): |
| 39 | if self.shuffle: |
| 40 | self.rng.shuffle(self.archivefiles) |
| 41 | for archive in self.archivefiles: |
| 42 | im_data = archive[0].read(archive[1]) |
| 43 | im_data = np.asarray(bytearray(im_data), dtype='uint8') |
| 44 | yield [im_data] |
| 45 | |
| 46 | |
| 47 | class ImageEncode(MapDataComponent): |
no outgoing calls
no test coverage detected