Epoch

You can import icepy4d core module by

from icepy4d import core as icecore

and directly access to the Epoch and Epoches class by

icecore.Epoch

icepy4d.core.epoch.EpochDataMap

Bases: dict

A class for managing epoch data mapping, including timestamps and associated images.

Parameters:
  • image_dir (Union[str, Path]) –

    The directory containing image data.

  • master_camera (str, default: None ) –

    The name of the master camera (optional).

  • time_tolerance_sec (int, default: 180 ) –

    The maximum time difference allowed to consider two images taken by different cameras as simultaneous (this allows for considering a non-perfect time synchronization between different cameras). Default is 180 seconds (3 minutes).

  • min_images (int, default: 2 ) –

    The minimum number of images required for an epoch to be included (optional).

Attributes:
  • _image_dir (Path) –

    The path to the image directory.

  • _master_camera (str) –

    The name of the master camera.

  • _timetolerance (timedelta) –

    The time tolerance for timestamp matching.

  • _cams (List[str]) –

    The list of camera names.

  • _map (dict) –

    The mapping of epoch data.

Methods:

Name Description
__init__

Union[str, Path], master_camera=None, time_tolerance_sec: timedelta = 180, min_images: int = 2): Initialize the EpochDataMap with image directory, master camera, and time tolerance.

__getitem__

Retrieve epoch data by key.

__repr__

Return a string representation of the EpochDataMap.

__len__

Return the number of epochs in the EpochDataMap.

__iter__

Initialize an iterator for the EpochDataMap.

__next__

Get the next element in the EpochDataMap.

__contains__

Union[str, dt]) -> bool: Check if a timestamp is present in the EpochDataMap.

cams

Return the list of camera names.

get_timestamp

int) -> dt: Get the timestamp of a specific epoch.

get_timestamp_str

int) -> str: Get the timestamp of a specific epoch as a formatted string.

get_images

int) -> List[Path]: Get the images associated with a specific epoch.

get_images_by_timestamp

Union[str, dt]) -> List[Path]: Get the images associated with an epoch by timestamp.

_get_timestamps

Union[str, Path]) -> Tuple[List[dt], List[Path]]: Get timestamps and image paths from a specified folder.

_build_map

int = None): Build the mapping of epoch data.

_write_map

str, sep: str = ",", header: bool = True) -> None: Write the mapping data to a CSV file.

Source code in src/icepy4d/core/epoch.py
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
class EpochDataMap(dict):
    """
    A class for managing epoch data mapping, including timestamps and associated images.

    Args:
        image_dir (Union[str, Path]): The directory containing image data.
        master_camera (str): The name of the master camera (optional).
        time_tolerance_sec (int): The maximum time difference allowed to consider two images taken by different cameras as simultaneous (this allows for considering a non-perfect time synchronization between different cameras). Default is 180 seconds (3 minutes).
        min_images (int): The minimum number of images required for an epoch to be included (optional).

    Attributes:
        _image_dir (Path): The path to the image directory.
        _master_camera (str): The name of the master camera.
        _timetolerance (timedelta): The time tolerance for timestamp matching.
        _cams (List[str]): The list of camera names.
        _map (dict): The mapping of epoch data.

    Methods:
        __init__(self, image_dir: Union[str, Path], master_camera=None, time_tolerance_sec: timedelta = 180, min_images: int = 2):
            Initialize the EpochDataMap with image directory, master camera, and time tolerance.
        __getitem__(self, key):
            Retrieve epoch data by key.
        __repr__(self) -> str:
            Return a string representation of the EpochDataMap.
        __len__(self) -> int:
            Return the number of epochs in the EpochDataMap.
        __iter__(self):
            Initialize an iterator for the EpochDataMap.
        __next__(self):
            Get the next element in the EpochDataMap.
        __contains__(self, timestamp: Union[str, dt]) -> bool:
            Check if a timestamp is present in the EpochDataMap.
        cams(self) -> List[str]:
            Return the list of camera names.
        get_timestamp(self, epoch_id: int) -> dt:
            Get the timestamp of a specific epoch.
        get_timestamp_str(self, epoch_id: int) -> str:
            Get the timestamp of a specific epoch as a formatted string.
        get_images(self, epoch_id: int) -> List[Path]:
            Get the images associated with a specific epoch.
        get_images_by_timestamp(self, timestamp: Union[str, dt]) -> List[Path]:
            Get the images associated with an epoch by timestamp.
        _get_timestamps(self, folder: Union[str, Path]) -> Tuple[List[dt], List[Path]]:
            Get timestamps and image paths from a specified folder.
        _build_map(self, min_images: int = None):
            Build the mapping of epoch data.
        _write_map(self, filename: str, sep: str = ",", header: bool = True) -> None:
            Write the mapping data to a CSV file.
    """

    def __init__(
        self,
        image_dir: Union[str, Path],
        master_camera=None,
        time_tolerance_sec: timedelta = 180,
        min_images: int = 2,
    ):
        """
        Initialize the EpochDataMap with image directory, master camera, and time tolerance.
        """
        self._image_dir = Path(image_dir)
        assert self._image_dir.exists(), f"{self._image_dir} does not exist"

        # if a master camera is specified, check if it exists in the image dir
        if master_camera:
            assert self._master_camera in os.scandir(
                self._image_dir
            ), f"Master camera not found in image directory {self._image_dir}"
        # if master camera is not specified, use the first camera found
        else:
            master_camera = sorted(
                [f.name for f in os.scandir(self._image_dir) if f.is_dir()]
            )[0]
        self._master_camera = master_camera

        self._timetolerance = timedelta(seconds=time_tolerance_sec)

        # Get camera names
        self._cams = sorted([f.name for f in os.scandir(self._image_dir) if f.is_dir()])

        # Build dict
        self._map = {}
        self._build_map(min_images)

        # Write dict to file
        self._write_map(self._image_dir / "epoch_map.csv")

    def __getitem__(self, key):
        return self._map[key]

    def __repr__(self) -> str:
        return f"{self.__class__.__name__} with {len(self._map)} epochs"

    def __len__(self):
        return len(self._map)

    def __iter__(self):
        self._elem = 0
        return self

    def __next__(self):
        while self._elem < len(self._map):
            file = self._map[self._elem]
            self._elem += 1
            return file
        else:
            self._elem = 0
            raise StopIteration

    def __contains__(self, timestamp: Union[str, dt]) -> bool:
        timestamp = parse_str_to_datetime(timestamp)
        timestamps = [x["timestamp"] for x in self._map.values()]
        return timestamp in timestamps

    @property
    def cams(self):
        return self._cams

    def get_timestamp(self, epoch_id: int) -> dt:
        return self._map[epoch_id]["timestamp"]

    def get_timestamp_str(self, epoch_id: int) -> str:
        return self._map[epoch_id]["timestamp"].strftime(DATETIME_FMT)

    def get_images(self, epoch_id: int) -> List[Path]:
        return self._map[epoch_id]["images"]

    def get_images_by_timestamp(self, timestamp: Union[str, dt]) -> List[Path]:
        timestamp = parse_str_to_datetime(timestamp)
        timestamps = [x["timestamp"] for x in self._map.values()]
        idx = timestamps.index(timestamp)
        return self._map[idx]["images"]

    def _get_timestamps(self, folder: Union[str, Path]) -> Tuple[List[dt], List[Path]]:
        imageDS = ImageDS(folder)
        paths = list(imageDS.files)
        timestamps = list(imageDS.timestamps.values())
        return timestamps, paths

    def _build_map(self, min_images: int = None):
        # Re-initialize map
        self._map = {}

        # Build imageDS for master camera and get timestamps
        timestamps, paths = self._get_timestamps(self._image_dir / self._master_camera)

        # build mapping dict for master camera
        for i, (ts, path) in enumerate(zip(timestamps, paths)):
            img = Image(path)
            self._map[i] = DotDict(
                {
                    "timestamp": ts,
                    "images": {self._master_camera: img},
                    "dt": {self._master_camera: timedelta(seconds=0)},
                }
            )

        # Find closest timestamp for each camera
        slave_cameras = set(self._cams)
        slave_cameras.discard(self._master_camera)

        for cam in slave_cameras:
            timestamps1, paths1 = self._get_timestamps(self._image_dir / cam)
            for key, value in self._map.items():
                ref_ts = value["timestamp"]
                _, closest_idx, dt = find_closest_timestamp(
                    ref_ts, timestamps1, self._timetolerance
                )
                if closest_idx is not None:
                    self._map[key]["images"][cam] = Image(paths1[closest_idx])
                    self._map[key]["dt"][cam] = dt
        logger.info(f"Building EpochDataMap: found {len(self._map)} epochs")

        # Get max dt for each epoch
        self._max_dt_sec = {
            ep: max(list(x["dt"].values())).seconds for ep, x in self._map.items()
        }
        dtmax = np.array([x for x in self._max_dt_sec.values()])
        logger.info(
            f"Mean max dt: {np.mean(dtmax):.2f} seconds (max: {np.max(dtmax):.2f} seconds))"
        )

        # Remove epochs with less than min_images images
        removed = 0
        if min_images is not None:
            for key, value in self._map.copy().items():
                if len(value["images"]) < min_images:
                    del self._map[key]
                    removed += 1
            logger.info(f"Removed {removed} epochs with less than {min_images} images")

    def _write_map(self, filename: str, sep: str = ",", header: bool = True) -> None:
        file = open(filename, "w")
        if header:
            columns = ["epoch", "date", "time"]
            for cam in self._cams:
                columns.append(cam)
                columns.append(f"{cam}_timestamp")
            file.write(f"{sep}".join(columns) + "\n")
        for key, value in self._map.items():
            value = self._map[key]
            date = value["timestamp"].strftime("%Y-%m-%d")
            time = value["timestamp"].strftime("%H:%M:%S")
            str_2_add = []
            for cam in self._cams:
                str_2_add.append(value["images"][cam].name)
                str_2_add.append(
                    f"{value['images'][cam].date}_{value['images'][cam].time}"
                )
            line = [str(key), date, time] + str_2_add
            file.write(f"{sep}".join(line) + "\n")
        file.close()

__init__(image_dir, master_camera=None, time_tolerance_sec=180, min_images=2)

Initialize the EpochDataMap with image directory, master camera, and time tolerance.

Source code in src/icepy4d/core/epoch.py
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
def __init__(
    self,
    image_dir: Union[str, Path],
    master_camera=None,
    time_tolerance_sec: timedelta = 180,
    min_images: int = 2,
):
    """
    Initialize the EpochDataMap with image directory, master camera, and time tolerance.
    """
    self._image_dir = Path(image_dir)
    assert self._image_dir.exists(), f"{self._image_dir} does not exist"

    # if a master camera is specified, check if it exists in the image dir
    if master_camera:
        assert self._master_camera in os.scandir(
            self._image_dir
        ), f"Master camera not found in image directory {self._image_dir}"
    # if master camera is not specified, use the first camera found
    else:
        master_camera = sorted(
            [f.name for f in os.scandir(self._image_dir) if f.is_dir()]
        )[0]
    self._master_camera = master_camera

    self._timetolerance = timedelta(seconds=time_tolerance_sec)

    # Get camera names
    self._cams = sorted([f.name for f in os.scandir(self._image_dir) if f.is_dir()])

    # Build dict
    self._map = {}
    self._build_map(min_images)

    # Write dict to file
    self._write_map(self._image_dir / "epoch_map.csv")

icepy4d.core.epoch.Epoch

Class for storing, saving, and reading ICEpy4D solution at one epoch

Attributes:
  • cameras (Camera) –

    The dictionary of camera parameters

  • images (ImagesDict) –

    The dictionary of images and their metadata

  • features (Features) –

    The dictionary of feature points

  • points (Points) –

    The dictionary of 3D points

Source code in src/icepy4d/core/epoch.py
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
class Epoch:
    """
    Class for storing, saving, and reading ICEpy4D solution at one epoch

    Attributes:
        cameras (classes.Camera): The dictionary of camera parameters
        images (classes.ImagesDict): The dictionary of images and their metadata
        features (classes.Features): The dictionary of feature points
        points (classes.Points): The dictionary of 3D points
    """

    def __init__(
        self,
        timestamp: Union[str, dt],
        epoch_dir: Union[str, Path] = None,
        images: Dict[str, Image] = None,
        cameras: Dict[str, Camera] = None,
        features: Dict[str, Features] = None,
        points: Points = None,
        targets: Targets = None,
        point_cloud: PointCloud = None,
        datetime_format: str = DATETIME_FMT,
    ) -> None:
        """
        Initializes a Epcoh object with the provided data

        Args:
            cameras (classes.Camera): The dictionary of camera parameters
            images (classes.ImagesDict): The dictionary of images and their metadata
            features (classes.Features): The dictionary of feature points
            points (classes.Points): The dictionary of 3D points

        TODO: Simplify the initialization of the Epoch object if no parameters are provided.
        """

        self._timestamp = parse_str_to_datetime(timestamp, datetime_format)

        if images is not None:
            assert (
                isinstance(images, dict) and isinstance(x, Image)
                for x in images.values()
            ), "Input images must be a dictionary mapping each camera name to the image (of type Image)"
            self.images = images
        self.cameras = cameras
        self.features = features
        self.targets = targets
        self.point_cloud = point_cloud

        if points is not None:
            assert isinstance(points, Points), "Input points must be of type Points"
            self.points = points
        else:
            self.points = Points()

        if epoch_dir is not None:
            self.epoch_dir = Path(epoch_dir)
        else:
            logger.info("Epoch directory not provided. Using epoch timestamp.")
            self.epoch_dir = Path(str(self._timestamp).replace(" ", "_"))
        self.epoch_dir.mkdir(parents=True, exist_ok=True)

    @property
    def timestamp(self):
        return self._timestamp

    @property
    def date_str(self) -> str:
        """
        Returns the date and time of the epoch in a string.

        Returns:
            str: The date and time of the epoch in the format "YYYY:MM:DD HH:MM:SS".
        """
        return self._timestamp.strftime("%Y:%m:%d")

    @property
    def time_str(self) -> str:
        """
        Returns the time of the epoch as a string.

        Returns:
            str: The time of the epoch in the format "HH:MM:SS".

        """
        return self._timestamp.strftime("%H:%M:%S")

    def __str__(self) -> str:
        """
        Returns a string representation of the Epoch object

        Returns:
            str: The string representation of the Epoch object
        """
        return f"{self._timestamp.strftime(DATETIME_FMT).replace(' ', '_')}"

    def __repr__(self):
        """
        Returns a string representation of the Epoch object

        Returns:
            str: The string representation of the Epoch object
        """
        return f"Epoch {self._timestamp}"

    def __iter__(self):
        """
        Returns an iterator over the four dictionaries of Epoch object

        Yields:
            dict: The dictionary of camera parameters
            dict: The dictionary of images and their metadata
            dict: The dictionary of feature points
            dict: The dictionary of 3D points
        """
        yield self.cameras
        yield self.images
        yield self.features
        yield self.points

    def __hash__(self):
        """
        Computes the hash value of the Epoch object

        Returns:
            int: The hash value of the Epoch object
        """
        return hash((self.cameras, self.images, self.features, self.points))

    def save_pickle(self, path: Union[str, Path]) -> bool:
        """
        Saves the Epoch object to a binary file

        Args:
            path (Union[str, Path]): The path to the binary file

        Returns:
            bool: True if the object was successfully saved to file, False otherwise
        """
        path = Path(path)
        path.parent.mkdir(parents=True, exist_ok=True)
        try:
            with open(path, "wb") as f:
                pickle.dump(self, f, pickle.HIGHEST_PROTOCOL)
            return True
        except:
            logger.error("Unable to save the Solution as Pickle object")
            return False

    @staticmethod
    def read_pickle(path: Union[str, Path]):
        """
        Load a Epoch object from a binary file

        Args:
            path (Union[str, Path]): The path to the binary file

        Returns:
            Epoch: An Epoh object
        """
        path = Path(path)
        assert path.exists(), f"Input path {path} does not exists"

        try:
            logger.info(f"Loading epoch from {path}")
            with open(path, "rb") as inp:
                epoch = pickle.load(inp)
            if epoch is not None:
                logger.info(f"Epoch loaded from {path}")
                return epoch
            else:
                logger.error(f"Unable to load epoch from {path}")
                return None
        except Exception as e:
            raise e(f"Unable to read Epoch from file {path}")

date_str: str property

Returns the date and time of the epoch in a string.

Returns:
  • str( str ) –

    The date and time of the epoch in the format "YYYY:MM:DD HH:MM:SS".

time_str: str property

Returns the time of the epoch as a string.

Returns:
  • str( str ) –

    The time of the epoch in the format "HH:MM:SS".

__hash__()

Computes the hash value of the Epoch object

Returns:
  • int

    The hash value of the Epoch object

Source code in src/icepy4d/core/epoch.py
446
447
448
449
450
451
452
453
def __hash__(self):
    """
    Computes the hash value of the Epoch object

    Returns:
        int: The hash value of the Epoch object
    """
    return hash((self.cameras, self.images, self.features, self.points))

__init__(timestamp, epoch_dir=None, images=None, cameras=None, features=None, points=None, targets=None, point_cloud=None, datetime_format=DATETIME_FMT)

Initializes a Epcoh object with the provided data

Parameters:
  • cameras (Camera, default: None ) –

    The dictionary of camera parameters

  • images (ImagesDict, default: None ) –

    The dictionary of images and their metadata

  • features (Features, default: None ) –

    The dictionary of feature points

  • points (Points, default: None ) –

    The dictionary of 3D points

TODO: Simplify the initialization of the Epoch object if no parameters are provided.

Source code in src/icepy4d/core/epoch.py
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
def __init__(
    self,
    timestamp: Union[str, dt],
    epoch_dir: Union[str, Path] = None,
    images: Dict[str, Image] = None,
    cameras: Dict[str, Camera] = None,
    features: Dict[str, Features] = None,
    points: Points = None,
    targets: Targets = None,
    point_cloud: PointCloud = None,
    datetime_format: str = DATETIME_FMT,
) -> None:
    """
    Initializes a Epcoh object with the provided data

    Args:
        cameras (classes.Camera): The dictionary of camera parameters
        images (classes.ImagesDict): The dictionary of images and their metadata
        features (classes.Features): The dictionary of feature points
        points (classes.Points): The dictionary of 3D points

    TODO: Simplify the initialization of the Epoch object if no parameters are provided.
    """

    self._timestamp = parse_str_to_datetime(timestamp, datetime_format)

    if images is not None:
        assert (
            isinstance(images, dict) and isinstance(x, Image)
            for x in images.values()
        ), "Input images must be a dictionary mapping each camera name to the image (of type Image)"
        self.images = images
    self.cameras = cameras
    self.features = features
    self.targets = targets
    self.point_cloud = point_cloud

    if points is not None:
        assert isinstance(points, Points), "Input points must be of type Points"
        self.points = points
    else:
        self.points = Points()

    if epoch_dir is not None:
        self.epoch_dir = Path(epoch_dir)
    else:
        logger.info("Epoch directory not provided. Using epoch timestamp.")
        self.epoch_dir = Path(str(self._timestamp).replace(" ", "_"))
    self.epoch_dir.mkdir(parents=True, exist_ok=True)

__iter__()

Returns an iterator over the four dictionaries of Epoch object

Yields:
  • dict

    The dictionary of camera parameters

  • dict

    The dictionary of images and their metadata

  • dict

    The dictionary of feature points

  • dict

    The dictionary of 3D points

Source code in src/icepy4d/core/epoch.py
431
432
433
434
435
436
437
438
439
440
441
442
443
444
def __iter__(self):
    """
    Returns an iterator over the four dictionaries of Epoch object

    Yields:
        dict: The dictionary of camera parameters
        dict: The dictionary of images and their metadata
        dict: The dictionary of feature points
        dict: The dictionary of 3D points
    """
    yield self.cameras
    yield self.images
    yield self.features
    yield self.points

__repr__()

Returns a string representation of the Epoch object

Returns:
  • str

    The string representation of the Epoch object

Source code in src/icepy4d/core/epoch.py
422
423
424
425
426
427
428
429
def __repr__(self):
    """
    Returns a string representation of the Epoch object

    Returns:
        str: The string representation of the Epoch object
    """
    return f"Epoch {self._timestamp}"

__str__()

Returns a string representation of the Epoch object

Returns:
  • str( str ) –

    The string representation of the Epoch object

Source code in src/icepy4d/core/epoch.py
413
414
415
416
417
418
419
420
def __str__(self) -> str:
    """
    Returns a string representation of the Epoch object

    Returns:
        str: The string representation of the Epoch object
    """
    return f"{self._timestamp.strftime(DATETIME_FMT).replace(' ', '_')}"

read_pickle(path) staticmethod

Load a Epoch object from a binary file

Parameters:
  • path (Union[str, Path]) –

    The path to the binary file

Returns:
  • Epoch

    An Epoh object

Source code in src/icepy4d/core/epoch.py
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
@staticmethod
def read_pickle(path: Union[str, Path]):
    """
    Load a Epoch object from a binary file

    Args:
        path (Union[str, Path]): The path to the binary file

    Returns:
        Epoch: An Epoh object
    """
    path = Path(path)
    assert path.exists(), f"Input path {path} does not exists"

    try:
        logger.info(f"Loading epoch from {path}")
        with open(path, "rb") as inp:
            epoch = pickle.load(inp)
        if epoch is not None:
            logger.info(f"Epoch loaded from {path}")
            return epoch
        else:
            logger.error(f"Unable to load epoch from {path}")
            return None
    except Exception as e:
        raise e(f"Unable to read Epoch from file {path}")

save_pickle(path)

Saves the Epoch object to a binary file

Parameters:
  • path (Union[str, Path]) –

    The path to the binary file

Returns:
  • bool( bool ) –

    True if the object was successfully saved to file, False otherwise

Source code in src/icepy4d/core/epoch.py
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
def save_pickle(self, path: Union[str, Path]) -> bool:
    """
    Saves the Epoch object to a binary file

    Args:
        path (Union[str, Path]): The path to the binary file

    Returns:
        bool: True if the object was successfully saved to file, False otherwise
    """
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    try:
        with open(path, "wb") as f:
            pickle.dump(self, f, pickle.HIGHEST_PROTOCOL)
        return True
    except:
        logger.error("Unable to save the Solution as Pickle object")
        return False

icepy4d.core.epoch.Epoches

Class for storing all the epochs in ICEpy4D processing

Source code in src/icepy4d/core/epoch.py
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
class Epoches:
    """Class for storing all the epochs in ICEpy4D processing"""

    def __init__(self, starting_epoch: int = 0) -> None:
        self._starting_epoch = starting_epoch
        self._last_epoch: int = -1
        self._epochs: Dict[int, Epoch] = {}
        self._epoches_map: Dict[int, dt] = {}
        self._elem = 0

    def __repr__(self):
        """
        Returns a string representation of the Epoches object

        Returns:
            str: The string representation of the Epoches object
        """

        return f"Epoches with {len(self._epochs)} epochs"

    def __len__(self) -> int:
        """Get number of epoches in the Epoches object"""
        return len(self._epochs)

    def __iter__(self):
        self._elem = self._starting_epoch
        return self

    def __next__(self):
        while self._elem <= self._last_epoch:
            file = self._epochs[self._elem]
            self._elem += 1
            return file
        else:
            self._elem
            raise StopIteration

    def __getitem__(self, epoch_id):
        """
        Returns the epoch object with the provided epoch_id

        Args:
            epoch_id (int): The numeric key of the epoch

        Returns:
            Epoch: The epoch object
        """
        return self._epochs[epoch_id]

    def __contains__(
        self, epoch_date: Union[str, dt], datetime_format: str = DATETIME_FMT
    ) -> bool:
        """Check if an epoch is in the Epoch objet"""
        timestamp = parse_str_to_datetime(epoch_date, datetime_format)
        timestamps = [x for x in self._epoches_map.values()]
        return timestamp in timestamps

    def add_epoch(self, epoch: Epoch):
        """
        Adds an epoch to the Epoches object

        Args:
            epoch_id (int): The numeric key of the epoch
            epoch_date (str): The corresponding date of the epoch
            epoch (Epoch): The epoch object to be added
        """
        assert isinstance(epoch, Epoch), "Input epoch must be of type Epoch"
        assert hasattr(epoch, "timestamp"), "Epoch must have a timestamp attribute"
        assert isinstance(
            epoch.timestamp, dt
        ), "Epoch timestamp must be of type datetime"
        assert (
            epoch.timestamp not in self._epoches_map.values()
        ), "Epoch with the same timestamp already exists"
        if self._last_epoch == -1:
            epoch_id = self._starting_epoch
        else:
            epoch_id = self._last_epoch + 1
        self._epoches_map[epoch_id] = epoch.timestamp
        self._epochs[epoch_id] = epoch
        self._last_epoch = epoch_id

    def get_epoch_date(self, epoch_id: int) -> str:
        """
        Retrieves the date corresponding to an epoch from the Epoches object

        Args:
            epoch_id (int): The numeric key of the epoch

        Returns:
            str: The date corresponding to the epoch_id
        """
        return self._epochs.get(epoch_id).timestamp

    def get_epoch_id(
        self, epoch_date: Union[str, dt], datetime_format: str = DATETIME_FMT
    ) -> int:
        timestamp = parse_str_to_datetime(epoch_date, datetime_format)
        for i, ep in self._epochs.items():
            if ep.timestamp == timestamp:
                return i

    def get_epoch_by_date(
        self, timestamp: Union[str, dt], datetime_format: str = DATETIME_FMT
    ) -> Epoch:
        timestamp = parse_str_to_datetime(timestamp, datetime_format)
        for ep in self._epochs.values():
            if ep.timestamp == timestamp:
                return ep
        logger.warning(f"Epoch with timestamp {timestamp} not found")
        return None

__contains__(epoch_date, datetime_format=DATETIME_FMT)

Check if an epoch is in the Epoch objet

Source code in src/icepy4d/core/epoch.py
552
553
554
555
556
557
558
def __contains__(
    self, epoch_date: Union[str, dt], datetime_format: str = DATETIME_FMT
) -> bool:
    """Check if an epoch is in the Epoch objet"""
    timestamp = parse_str_to_datetime(epoch_date, datetime_format)
    timestamps = [x for x in self._epoches_map.values()]
    return timestamp in timestamps

__getitem__(epoch_id)

Returns the epoch object with the provided epoch_id

Parameters:
  • epoch_id (int) –

    The numeric key of the epoch

Returns:
  • Epoch

    The epoch object

Source code in src/icepy4d/core/epoch.py
540
541
542
543
544
545
546
547
548
549
550
def __getitem__(self, epoch_id):
    """
    Returns the epoch object with the provided epoch_id

    Args:
        epoch_id (int): The numeric key of the epoch

    Returns:
        Epoch: The epoch object
    """
    return self._epochs[epoch_id]

__len__()

Get number of epoches in the Epoches object

Source code in src/icepy4d/core/epoch.py
523
524
525
def __len__(self) -> int:
    """Get number of epoches in the Epoches object"""
    return len(self._epochs)

__repr__()

Returns a string representation of the Epoches object

Returns:
  • str

    The string representation of the Epoches object

Source code in src/icepy4d/core/epoch.py
513
514
515
516
517
518
519
520
521
def __repr__(self):
    """
    Returns a string representation of the Epoches object

    Returns:
        str: The string representation of the Epoches object
    """

    return f"Epoches with {len(self._epochs)} epochs"

add_epoch(epoch)

Adds an epoch to the Epoches object

Parameters:
  • epoch_id (int) –

    The numeric key of the epoch

  • epoch_date (str) –

    The corresponding date of the epoch

  • epoch (Epoch) –

    The epoch object to be added

Source code in src/icepy4d/core/epoch.py
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
def add_epoch(self, epoch: Epoch):
    """
    Adds an epoch to the Epoches object

    Args:
        epoch_id (int): The numeric key of the epoch
        epoch_date (str): The corresponding date of the epoch
        epoch (Epoch): The epoch object to be added
    """
    assert isinstance(epoch, Epoch), "Input epoch must be of type Epoch"
    assert hasattr(epoch, "timestamp"), "Epoch must have a timestamp attribute"
    assert isinstance(
        epoch.timestamp, dt
    ), "Epoch timestamp must be of type datetime"
    assert (
        epoch.timestamp not in self._epoches_map.values()
    ), "Epoch with the same timestamp already exists"
    if self._last_epoch == -1:
        epoch_id = self._starting_epoch
    else:
        epoch_id = self._last_epoch + 1
    self._epoches_map[epoch_id] = epoch.timestamp
    self._epochs[epoch_id] = epoch
    self._last_epoch = epoch_id

get_epoch_date(epoch_id)

Retrieves the date corresponding to an epoch from the Epoches object

Parameters:
  • epoch_id (int) –

    The numeric key of the epoch

Returns:
  • str( str ) –

    The date corresponding to the epoch_id

Source code in src/icepy4d/core/epoch.py
585
586
587
588
589
590
591
592
593
594
595
def get_epoch_date(self, epoch_id: int) -> str:
    """
    Retrieves the date corresponding to an epoch from the Epoches object

    Args:
        epoch_id (int): The numeric key of the epoch

    Returns:
        str: The date corresponding to the epoch_id
    """
    return self._epochs.get(epoch_id).timestamp