media

camtrapdp/__init__.py

@@ -1,3 +1,4 @@
 from .deployment import Deployment
+from .media import Media
 
-__all__ = ["Deployment"]
+__all__ = ["Deployment", "Media"]

camtrapdp/deployment.py

@@ -1,10 +1,8 @@
 """
-Camtrap Data Package
+Camtrap Data Package deployment module
 """
 
-# flake8: N815
-
-from dataclasses import dataclass
+from dataclasses import dataclass, asdict
 from typing import Optional, List
 from enum import Enum
 from csv import DictReader, DictWriter
@@ -186,32 +184,57 @@ class FeatureType(str, Enum):
 
     @staticmethod
     def from_csv(file_path: str) -> List["Deployment"]:
-        deployments = []
+        """
+        Read deployment objects from a CSV file.
+
+        Args:
+            file_path: Path to the CSV file.
+
+        Returns:
+            List of deployment objects.
+        """
+
         with open(file_path, "r") as file:
             reader = DictReader(file)
-            for row in reader:
-                deployment = Deployment(**row)
-                deployments.append(deployment)
-
-        return deployments
+            return [Deployment(**row) for row in reader]
 
     @staticmethod
-    def to_csv(observations: List["Deployment"], file_path: str):
+    def to_csv(deployments: List["Deployment"], file_path: str):
+        """
+        Write deployment objects to a CSV file.
+
+        Args:
+            deployments: List of deployment objects.
+            file_path: Path to the CSV file.
+        """
         with open(file_path, "w") as file:
             writer = DictWriter(file, fieldnames=Deployment.__dataclass_fields__.keys())
             writer.writeheader()
-            for deployment in observations:
-                writer.writerow(deployment.__dict__)
+            writer.writerows([asdict(m) for m in deployments])
 
     @staticmethod
-    def to_pandas(observations: List["Deployment"]):
-        return DataFrame([deployment.__dict__ for deployment in observations])
+    def to_pandas(deployments: List["Deployment"]):
+        """
+        Convert deployment objects to a pandas DataFrame.
+
+        Args:
+            deployments: List of deployment objects.
+
+        Returns:
+            DataFrame of deployment objects.
+        """
+        return DataFrame([deployment.__dict__ for deployment in deployments])
 
     @staticmethod
     def from_pandas(dataframe: DataFrame) -> List["Deployment"]:
-        deployments = []
-        for index, row in dataframe.iterrows():
-            deployment = Deployment(**row)
-            deployments.append(deployment)
+        """
+        Convert deployment objects from a pandas DataFrame.
+
+        Args:
+            dataframe: DataFrame of deployment objects.
+
+        Returns:
+            List of deployment objects.
+        """
 
-        return deployments
+        return [Deployment(**row) for index, row in dataframe.iterrows()]

camtrapdp/media.py

@@ -0,0 +1,142 @@
+"""
+Camtrap Data Package media module
+"""
+
+from dataclasses import dataclass, asdict
+from typing import Optional, List
+from enum import Enum
+from csv import DictReader, DictWriter
+from pandas import DataFrame
+
+
+@dataclass
+class Media:
+    """
+    A media object represents a single image or video file captured by a camera
+    trap.
+    """
+
+    mediaID: str
+    """
+    Unique identifier of the media object.
+    """
+
+    deploymentID: str
+    """
+    Identifier of the deployment the media file belongs to. Foreign key to
+    deployments.deploymentID.
+    """
+
+    class CaptureMethod(Enum):
+        ACTIVITY_DETECTION = "activityDetection"
+        TIME_LAPSE = "timeLapse"
+
+    captureMethod: Optional[CaptureMethod]
+    """
+    Method used to capture the media file.
+    """
+
+    timestamp: str
+    """
+    Date and time at which the media file was recorded. Formatted as an ISO
+    8601 string with timezone designator (YYYY-MM-DDThh:mm:ssZ or
+    YYYY-MM-DDThh:mm:ss±hh:mm).
+    """
+
+    filePath: str
+    """
+    URL or relative path to the media file, respectively for externally hosted
+    files or files that are part of the package.
+    """
+
+    filePublic: bool
+    """
+    false if the media file is not publicly accessible (e.g. to protect the
+    privacy of people).
+    """
+
+    fileName: Optional[str]
+    """
+    Name of the media file. If provided, one should be able to sort media files
+    chronologically within a deployment on timestamp (first) and fileName
+    (second).
+    """
+
+    fileMediatype: str
+    """
+    Mediatype of the media file. Expressed as an IANA Media Type.
+    """
+
+    exifData: Optional[dict]
+    """
+    EXIF data of the media file. Formatted as a valid JSON object.
+    """
+
+    favorite: Optional[bool]
+    """
+    true if the media file is deemed of interest (e.g. an exemplar image of an individual).
+    """
+
+    mediaComments: Optional[str]
+    """
+    Comments or notes about the media file.
+    """
+
+    @staticmethod
+    def from_csv(file_path: str) -> List["Media"]:
+        """
+        Read media objects from a CSV file.
+
+        Args:
+            file_path: Path to the CSV file.
+
+        Returns:
+            List of media objects.
+        """
+
+        with open(file_path, "r") as file:
+            reader = DictReader(file)
+            return [Media(**row) for row in reader]
+
+    @staticmethod
+    def to_csv(media: List["Media"], file_path: str):
+        """
+        Write media objects to a CSV file.
+
+        Args:
+            media: List of media objects.
+            file_path: Path to the CSV file.
+        """
+
+        with open(file_path, "w") as file:
+            writer = DictWriter(file, fieldnames=Media.__dataclass_fields__.keys())
+            writer.writeheader()
+            writer.writerows([asdict(m) for m in media])
+
+    @staticmethod
+    def to_pandas(media: List["Media"]) -> DataFrame:
+        """
+        Convert media objects to a pandas DataFrame.
+
+        Args:
+            media: List of media objects.
+
+        Returns:
+            DataFrame of media objects.
+        """
+
+        return DataFrame([asdict(m) for m in media])
+
+    @staticmethod
+    def from_pandas(dataframe: DataFrame) -> List["Media"]:
+        """
+        Convert media objects from a pandas DataFrame.
+
+        Args:
+            dataframe: DataFrame of media objects.
+
+        Returns:
+            List of media objects.
+        """
+
+        return [Media(**row) for index, row in dataframe.iterrows()]

camtrapdp/test_deployment.py

@@ -16,10 +16,11 @@ def test_write_to_csv():
             with open(file.name, "r") as new:
                 assert original.read() == new.read()
 
+
 def test_to_from_pandas():
     deployments = Deployment.from_csv("fixtures/deployments.csv")
     dataframe = Deployment.to_pandas(deployments)
     new_deployments = Deployment.from_pandas(dataframe)
     assert len(deployments) == len(new_deployments)
     for old, new in zip(deployments, new_deployments):
-        assert old == new
+        assert old == new

camtrapdp/test_media.py

@@ -0,0 +1,26 @@
+from . import Media
+from tempfile import NamedTemporaryFile
+
+
+def test_read_from_csv():
+    media = Media.from_csv("fixtures/media.csv")
+    print(media)
+    assert len(media) == 423
+
+
+def test_write_to_csv():
+    media = Media.from_csv("fixtures/media.csv")
+    with NamedTemporaryFile(mode="w", delete=False) as file:
+        Media.to_csv(media, file.name)
+        with open("fixtures/media.csv", "r") as original:
+            with open(file.name, "r") as new:
+                assert original.read() == new.read()
+
+
+def test_to_from_pandas():
+    media = Media.from_csv("fixtures/media.csv")
+    dataframe = Media.to_pandas(media)
+    new_media = Media.from_pandas(dataframe)
+    assert len(media) == len(new_media)
+    for old, new in zip(media, new_media):
+        assert old == new