API Reference

The API reference documentation below is generated using mkdocstrings. The API reference documentation is written in docstrings and is located in the dfg_tool/ directory.

cli

Digital frame grab tool.

callback(verbose=False, log=None)

Handle the CLI callback.

Source code in dfg_tool/cli/__init__.py

def callback(
    verbose: Annotated[
        bool,
        typer.Option("-v", "--verbose", help="Enable verbose logging to the console"),
    ] = False,
    log: Annotated[
        Optional[Path],
        typer.Option("-l", "--log", help="Log to the given file", writable=True),
    ] = None,
):
    """
    Handle the CLI callback.
    """
    if verbose:
        RICH_HANDLER.setLevel(logging.DEBUG)

    if log:
        file_handler = logging.FileHandler(log)
        file_handler.setLevel(logging.DEBUG)
        set_file_handler(file_handler)

create_app()

Create the main CLI app.

Returns:

Type	Description
Typer	typer.Typer: The main CLI app

Source code in dfg_tool/cli/__init__.py

def create_app() -> typer.Typer:
    """
    Create the main CLI app.

    Returns:
        typer.Typer: The main CLI app
    """
    app = typer.Typer(
        name="dfg",
        help=__doc__,
        context_settings={"help_option_names": ["-h", "--help"]},
        pretty_exceptions_show_locals=False,
    )

    return app

link_app(app)

Link commands and callbacks to the app.

Parameters:

Name	Type	Description	Default
app	Typer	The main CLI app	required

Source code in dfg_tool/cli/__init__.py

def link_app(app: typer.Typer) -> None:
    """
    Link commands and callbacks to the app.

    Args:
        app (typer.Typer): The main CLI app
    """
    # Link callbacks
    app.callback()(callback)

    # Link commands
    app.command(name="extract")(extract)
    app.command(name="upload")(upload)

commands

Command

Bases: LogMixin

Base class for CLI commands.

run()

Run the command.

Source code in dfg_tool/cli/commands/__init__.py

def run(self):
    """
    Run the command.
    """
    self.logger.debug(f"Running command: {self.__class__.__name__}")
    try:
        self._run()
        self.logger.info(f"{self.__class__.__name__} command completed")
        print(
            Panel(
                f"The [bold]{self.__class__.__name__}[/bold] command completed successfully.",
                border_style="green",
                title="Command completed",
                title_align="left",
            )
        )
    except KeyboardInterrupt:
        self.logger.warning(f"{self.__class__.__name__} command interrupted")
        print(
            Panel(
                f"The [bold]{self.__class__.__name__}[/bold] command was interrupted.",
                border_style="yellow",
                title="Command interrupted",
                title_align="left",
            )
        )
    except Exception as e:
        self.logger.error(
            f"Command {self.__class__.__name__} failed: {e}", exc_info=True
        )
        print(
            Panel(
                f"The [bold]{self.__class__.__name__}[/bold] command failed. See the logs for more information.",
                border_style="red",
                title="Command failed",
                title_align="left",
            )
        )

extract

Extract(imaged_moment_uuids_path, output_dir_path, raziel_api_url, username, password)

Bases: Command

Extract frames and save them to disk.

Source code in dfg_tool/cli/commands/extract.py

def __init__(
    self,
    imaged_moment_uuids_path: Path,
    output_dir_path: Path,
    raziel_api_url: str,
    username: str,
    password: str,
) -> None:
    self._imaged_moment_uuids_path = imaged_moment_uuids_path
    self._output_dir_path = output_dir_path
    self._raziel_api_url = raziel_api_url
    self._username = username
    self._password = password

upload

Upload(input_dir_path, raziel_api_url, username, password)

Bases: Command

Upload captured frame grabs and register them in VARS.

Source code in dfg_tool/cli/commands/upload.py

def __init__(
    self,
    input_dir_path: Path,
    raziel_api_url: str,
    username: str,
    password: str,
) -> None:
    self._input_dir_path = input_dir_path
    self._raziel_api_url = raziel_api_url
    self._username = username
    self._password = password

log

Logging utilities

LogMixin

Mixin class to add a logger to a class.

logger: logging.Logger property

Get the logger for the class.

get_logger(name, level=logging.DEBUG, stream=False, rich=True)

Create a new logger with the given name and level.

Attaches a NullHandler to the logger to override default behavior. Optionally attaches a StreamHandler and/or RichHandler to the logger. If the global FILE_HANDLER is set, it will also be attached to the logger.

Parameters:

Name	Type	Description	Default
name	str	The name of the logger.	required
level	int	The logging level.	DEBUG
stream	bool	Whether to attach the module StreamHandler to the logger.	False
rich	bool	Whether to attach the module RichHandler to the logger.	True

Returns:

Type	Description
Logger	logging.Logger: The created logger.

Source code in dfg_tool/cli/log.py

def get_logger(
    name: str, level: int = logging.DEBUG, stream: bool = False, rich: bool = True
) -> logging.Logger:
    """
    Create a new logger with the given name and level.

    Attaches a NullHandler to the logger to override default behavior.
    Optionally attaches a StreamHandler and/or RichHandler to the logger.
    If the global FILE_HANDLER is set, it will also be attached to the logger.

    Args:
        name (str): The name of the logger.
        level (int): The logging level.
        stream (bool): Whether to attach the module StreamHandler to the logger.
        rich (bool): Whether to attach the module RichHandler to the logger.

    Returns:
        logging.Logger: The created logger.
    """
    logger = logging.Logger(name, level=level)
    logger.addHandler(logging.NullHandler())
    if stream:
        logger.addHandler(STREAM_HANDLER)
    if rich:
        logger.addHandler(RICH_HANDLER)
    if FILE_HANDLER is not None:
        logger.addHandler(FILE_HANDLER)
    return logger

set_file_handler(handler)

Set the global file handler to log to the given file.

Sets the file handler formatter to the global formatter.

Parameters:

Name	Type	Description	Default
handler	FileHandler	The file handler to set.	required

Source code in dfg_tool/cli/log.py

def set_file_handler(handler: logging.FileHandler) -> None:
    """
    Set the global file handler to log to the given file.

    Sets the file handler formatter to the global formatter.

    Args:
        handler (logging.FileHandler): The file handler to set.
    """
    global FILE_HANDLER
    FILE_HANDLER = handler
    FILE_HANDLER.setFormatter(FORMATTER)

lib

constants

Application constants.

m3

MBARI Media Management (M3) REST API clients

APIKeyAuth(api_key)

Bases: Auth

API key authentication.

Source code in dfg_tool/lib/m3/__init__.py

def __init__(self, api_key: str):
    self.api_key = api_key

APIKeyAuthenticatedRESTClient(base_url)

Bases: RESTClient

REST client that supports API key authentication.

Source code in dfg_tool/lib/m3/__init__.py

def __init__(self, base_url: str):
    self._base_url = base_url.rstrip("/")
    self._client = httpx.Client()

BearerAuth(token)

Bases: Auth

Bearer token authentication.

Source code in dfg_tool/lib/m3/__init__.py

def __init__(self, token: str):
    self.token = token

M3AuthMixin(base_url)

Bases: RESTClient

M3 client authentication mixin class.

Adds: - an authenticate method to authenticate with an API key M3-style - a from_raziel factory method to create a client from a Raziel endpoint configuration

Source code in dfg_tool/lib/m3/__init__.py

def __init__(self, base_url: str):
    self._base_url = base_url.rstrip("/")
    self._client = httpx.Client()

authenticate(api_key)

Authenticate by posting an API key at a predefined route /auth.

Parameters:

Name	Type	Description	Default
api_key	str	The API key.	required

Raises:

Type	Description
HTTPStatusError	If the API returns an error.

Source code in dfg_tool/lib/m3/__init__.py

def authenticate(self, api_key: str) -> None:
    """
    Authenticate by posting an API key at a predefined route `/auth`.

    Args:
        api_key (str): The API key.

    Raises:
        httpx.HTTPStatusError: If the API returns an error.
    """
    response = self.post(
        "auth",
        headers={
            "Authorization": f"APIKEY {api_key}",
        },
    )
    response.raise_for_status()
    self.token = response.json()["access_token"]

from_raziel(endpoint) classmethod

Create an M3Client from a Raziel endpoint configuration.

Parameters:

Name	Type	Description	Default
endpoint	dict	The Raziel endpoint configuration.	required

Returns:

Name	Type	Description
M3Client	M3AuthMixin	The M3Client.

Raises:

Type	Description
KeyError	If the endpoint configuration is missing the url key.
HTTPStatusError	If the API returns an error during authentication.

Source code in dfg_tool/lib/m3/__init__.py

@classmethod
def from_raziel(cls, endpoint: dict) -> "M3AuthMixin":
    """
    Create an M3Client from a Raziel endpoint configuration.

    Args:
        endpoint (dict): The Raziel endpoint configuration.

    Returns:
        M3Client: The M3Client.

    Raises:
        KeyError: If the endpoint configuration is missing the `url` key.
        httpx.HTTPStatusError: If the API returns an error during authentication.
    """
    api_key = endpoint.get("secret", None)
    client = cls(endpoint["url"])
    client.authenticate(api_key)
    return client

RESTClient(base_url)

RESTful HTTP client. This class is a thin wrapper around httpx.Client that provides a convenient way to interact with a RESTful API.

Source code in dfg_tool/lib/m3/__init__.py

def __init__(self, base_url: str):
    self._base_url = base_url.rstrip("/")
    self._client = httpx.Client()

delete(path, **kwargs)

Send a DELETE request to the API.

Parameters:

Name	Type	Description	Default
path	str	The path to the resource.	required
**kwargs	dict	Additional keyword arguments to pass to httpx.Client.delete.	{}

Returns:

Type	Description
Response	httpx.Response: The response from the API.

Source code in dfg_tool/lib/m3/__init__.py

def delete(self, path: str, **kwargs) -> httpx.Response:
    """
    Send a DELETE request to the API.

    Args:
        path (str): The path to the resource.
        **kwargs (dict): Additional keyword arguments to pass to httpx.Client.delete.

    Returns:
        httpx.Response: The response from the API.
    """
    return self._client.delete(self.url_to(path), **kwargs)

get(path, **kwargs)

Send a GET request to the API.

Parameters:

Name	Type	Description	Default
path	str	The path to the resource.	required
**kwargs	dict	Additional keyword arguments to pass to httpx.Client.get.	{}

Returns:

Type	Description
Response	httpx.Response: The response from the API.

Source code in dfg_tool/lib/m3/__init__.py

def get(self, path: str, **kwargs) -> httpx.Response:
    """
    Send a GET request to the API.

    Args:
        path (str): The path to the resource.
        **kwargs (dict): Additional keyword arguments to pass to httpx.Client.get.

    Returns:
        httpx.Response: The response from the API.
    """
    return self._client.get(self.url_to(path), **kwargs)

post(path, **kwargs)

Send a POST request to the API.

Parameters:

Name	Type	Description	Default
path	str	The path to the resource.	required
**kwargs	dict	Additional keyword arguments to pass to httpx.Client.post.	{}

Returns:

Type	Description
Response	httpx.Response: The response from the API.

Source code in dfg_tool/lib/m3/__init__.py

def post(self, path: str, **kwargs) -> httpx.Response:
    """
    Send a POST request to the API.

    Args:
        path (str): The path to the resource.
        **kwargs (dict): Additional keyword arguments to pass to httpx.Client.post.

    Returns:
        httpx.Response: The response from the API.
    """
    return self._client.post(self.url_to(path), **kwargs)

put(path, **kwargs)

Send a PUT request to the API.

Parameters:

Name	Type	Description	Default
path	str	The path to the resource.	required
**kwargs	dict	Additional keyword arguments to pass to httpx.Client.put.	{}

Returns:

Type	Description
Response	httpx.Response: The response from the API.

Source code in dfg_tool/lib/m3/__init__.py

def put(self, path: str, **kwargs) -> httpx.Response:
    """
    Send a PUT request to the API.

    Args:
        path (str): The path to the resource.
        **kwargs (dict): Additional keyword arguments to pass to httpx.Client.put.

    Returns:
        httpx.Response: The response from the API.
    """
    return self._client.put(self.url_to(path), **kwargs)

url_to(path)

Get the URL to a path on the API.

Parameters:

Name	Type	Description	Default
path	str	The path to the resource.	required

Returns:

Name	Type	Description
str	str	The URL to the resource.

Source code in dfg_tool/lib/m3/__init__.py

def url_to(self, path: str) -> str:
    """
    Get the URL to a path on the API.

    Args:
        path (str): The path to the resource.

    Returns:
        str: The URL to the resource.
    """
    return f"{self._base_url}/{path.lstrip('/')}"

TokenAuthenticatedRESTClient(base_url)

Bases: RESTClient

REST client that supports bearer token authentication.

Source code in dfg_tool/lib/m3/__init__.py

def __init__(self, base_url: str):
    self._base_url = base_url.rstrip("/")
    self._client = httpx.Client()

annosaurus

imagedmoments

get(client, uuid)

Get an imaged moment by UUID.

Parameters:

Name	Type	Description	Default
client	RESTClient	The REST client.	required
uuid	UUID	The imaged moment UUID.	required

Returns:

Name	Type	Description
dict	dict	The imaged moment.

Raises:

Type	Description
HTTPStatusError	If the API returns an error.

Source code in dfg_tool/lib/m3/annosaurus/imagedmoments.py

def get(
    client: RESTClient,
    uuid: UUID,
) -> dict:
    """
    Get an imaged moment by UUID.

    Args:
        client (RESTClient): The REST client.
        uuid (UUID): The imaged moment UUID.

    Returns:
        dict: The imaged moment.

    Raises:
        httpx.HTTPStatusError: If the API returns an error.
    """
    response = client.get(f"imagedmoments/{uuid}")
    response.raise_for_status()
    return response.json()

images

create(client, video_reference_uuid, url, timecode=None, elapsed_time_millis=None, recorded_timestamp=None, format=None, width_pixels=None, height_pixels=None, description=None)

Create an image reference.

Note: one of timecode, elapsed_time_millis, or recorded_timestamp must be provided.

Parameters:

Name	Type	Description	Default
client	AnnosaurusClient	The Annosaurus client.	required
video_reference_uuid	UUID	The video reference UUID.	required
url	str	The image URL.	required
timecode	str	The image timecode.	None
elapsed_time_millis	int	The image elapsed time in milliseconds.	None
recorded_timestamp	str	The image recorded timestamp.	None
format	str	The image format.	None
width_pixels	int	The image width in pixels.	None
height_pixels	int	The image height in pixels.	None
description	str	The image description.	None

Source code in dfg_tool/lib/m3/annosaurus/images.py

def create(
    client: AnnosaurusClient,
    video_reference_uuid: UUID,
    url: str,
    timecode: Optional[str] = None,
    elapsed_time_millis: Optional[int] = None,
    recorded_timestamp: Optional[str] = None,
    format: Optional[str] = None,
    width_pixels: Optional[int] = None,
    height_pixels: Optional[int] = None,
    description: Optional[str] = None,
):
    """
    Create an image reference.

    Note: one of `timecode`, `elapsed_time_millis`, or `recorded_timestamp` must be provided.

    Args:
        client (AnnosaurusClient): The Annosaurus client.
        video_reference_uuid (UUID): The video reference UUID.
        url (str): The image URL.
        timecode (str, optional): The image timecode.
        elapsed_time_millis (int, optional): The image elapsed time in milliseconds.
        recorded_timestamp (str, optional): The image recorded timestamp.
        format (str, optional): The image format.
        width_pixels (int, optional): The image width in pixels.
        height_pixels (int, optional): The image height in pixels.
        description (str, optional): The image description.
    """
    data = {
        "video_reference_uuid": str(video_reference_uuid),
        "url": url,
    }
    if timecode is not None:
        data["timecode"] = timecode
    if elapsed_time_millis is not None:
        data["elapsed_time_millis"] = elapsed_time_millis
    if recorded_timestamp is not None:
        data["recorded_timestamp"] = recorded_timestamp
    if format is not None:
        data["format"] = format
    if width_pixels is not None:
        data["width_pixels"] = width_pixels
    if height_pixels is not None:
        data["height_pixels"] = height_pixels
    if description is not None:
        data["description"] = description

    response = client.post("images", data=data)
    response.raise_for_status()
    return response.json()

beholder

capture

capture(client, video_url, elapsed_time_millis)

Capture a frame.

Parameters:

Name	Type	Description	Default
client	RESTClient	The REST client.	required
video_url	str	The video URL.	required
elapsed_time_millis	int	The elapsed time in milliseconds.	required

Returns:

Name	Type	Description
bytes	bytes	The raw image data.

Raises:

Type	Description
HTTPStatusError	If the API returns an error.

Source code in dfg_tool/lib/m3/beholder/capture.py

def capture(client: RESTClient, video_url: str, elapsed_time_millis: int) -> bytes:
    """
    Capture a frame.

    Args:
        client (RESTClient): The REST client.
        video_url (str): The video URL.
        elapsed_time_millis (int): The elapsed time in milliseconds.

    Returns:
        bytes: The raw image data.

    Raises:
        httpx.HTTPStatusError: If the API returns an error.
    """
    body = {
        "videoUrl": video_url,
        "elapsedTimeMillis": elapsed_time_millis,
    }

    response = client.post(
        "capture",
        json=body,
    )
    response.raise_for_status()
    return response.content

panoptes

images

upload(client, camera_id, deployment_id, name, file)

Upload a framegrab.

Parameters:

Name	Type	Description	Default
client	PanoptesClient	The Panoptes client.	required
camera_id	str	The camera ID.	required
deployment_id	str	The deployment ID.	required
name	str	The framegrab name.	required
file	BytesIO	The framegrab file object.	required

Raises:

Type	Description
HTTPStatusError	If the API returns an error.

Source code in dfg_tool/lib/m3/panoptes/images.py

def upload(
    client: PanoptesClient,
    camera_id: str,
    deployment_id: str,
    name: str,
    file: BytesIO,
) -> dict:
    """
    Upload a framegrab.

    Args:
        client (PanoptesClient): The Panoptes client.
        camera_id (str): The camera ID.
        deployment_id (str): The deployment ID.
        name (str): The framegrab name.
        file (BytesIO): The framegrab file object.

    Raises:
        httpx.HTTPStatusError: If the API returns an error.
    """
    response = client.post(
        f"images/{camera_id}/{deployment_id}/{name}", files={"file": file}
    )
    response.raise_for_status()
    return response.json()

raziel

RazielClient(base_url)

Bases: TokenAuthenticatedRESTClient

Source code in dfg_tool/lib/m3/__init__.py

def __init__(self, base_url: str):
    self._base_url = base_url.rstrip("/")
    self._client = httpx.Client()

authenticate(username, password)

Authenticate the client with the Raziel API.

Parameters:

Name	Type	Description	Default
username	str	The username.	required
password	str	The password.	required

Raises:

Type	Description
HTTPStatusError	If the API returns an error.

Source code in dfg_tool/lib/m3/raziel/__init__.py

def authenticate(self, username: str, password: str) -> None:
    """
    Authenticate the client with the Raziel API.

    Args:
        username (str): The username.
        password (str): The password.

    Raises:
        httpx.HTTPStatusError: If the API returns an error.
    """
    self.token = authenticate(self, username, password)

auth

authenticate(client, username, password)

Authenticate with the Raziel API and return the JWT.

Parameters:

Name	Type	Description	Default
client	RESTClient	The REST client.	required
username	str	The username.	required
password	str	The password.	required

Returns:

Name	Type	Description
str	str	The JWT.

Raises:

Type	Description
HTTPStatusError	If the API returns an error.

Source code in dfg_tool/lib/m3/raziel/auth.py

def authenticate(client: RESTClient, username: str, password: str) -> str:
    """
    Authenticate with the Raziel API and return the JWT.

    Args:
        client (RESTClient): The REST client.
        username (str): The username.
        password (str): The password.

    Returns:
        str: The JWT.

    Raises:
        httpx.HTTPStatusError: If the API returns an error.
    """
    # Base64 encode the username and password
    credentials = f"{username}:{password}"
    encoded_credentials = b64encode(credentials.encode()).decode()

    # Authenticate with the API
    response = client.post(
        "/auth",
        headers={
            "Authorization": f"Basic {encoded_credentials}",
        },
    )
    response.raise_for_status()

    # Extract the JWT from the response
    token = response.json()["accessToken"]
    return token

endpoints

list(client)

List all endpoints.

Parameters:

Name	Type	Description	Default
client	AuthenticatedRESTClient	The authenticated REST client.	required

Returns:

Type	Description
List[dict]	List[dict]: The list of endpoints.

Raises:

Type	Description
HTTPStatusError	If the API returns an error.

Source code in dfg_tool/lib/m3/raziel/endpoints.py

def list(client: TokenAuthenticatedRESTClient) -> List[dict]:
    """
    List all endpoints.

    Args:
        client (AuthenticatedRESTClient): The authenticated REST client.

    Returns:
        List[dict]: The list of endpoints.

    Raises:
        httpx.HTTPStatusError: If the API returns an error.
    """
    response = client.get("/endpoints")
    response.raise_for_status()
    return response.json()

vampire_squid

media

get_by_video_reference_uuid(client, video_reference_uuid)

Get a media item by video reference UUID.

Parameters:

Name	Type	Description	Default
client	RESTClient	The REST client.	required
video_reference_uuid	UUID	The video reference UUID.	required

Returns:

Name	Type	Description
dict	dict	The media item.

Raises:

Type	Description
HTTPStatusError	If the API returns an error.

Source code in dfg_tool/lib/m3/vampire_squid/media.py

def get_by_video_reference_uuid(
    client: RESTClient,
    video_reference_uuid: UUID,
) -> dict:
    """
    Get a media item by video reference UUID.

    Args:
        client (RESTClient): The REST client.
        video_reference_uuid (UUID): The video reference UUID.

    Returns:
        dict: The media item.

    Raises:
        httpx.HTTPStatusError: If the API returns an error.
    """
    response = client.get(f"media/videoreference/{video_reference_uuid}")
    response.raise_for_status()
    return response.json()

videoreferences

get(client, uuid)

Get a video reference by UUID.

Parameters:

Name	Type	Description	Default
client	RESTClient	The REST client.	required
uuid	UUID	The video reference UUID.	required

Returns:

Name	Type	Description
dict	dict	The video reference.

Raises:

Type	Description
HTTPStatusError	If the API returns an error.

Source code in dfg_tool/lib/m3/vampire_squid/videoreferences.py

def get(
    client: RESTClient,
    uuid: UUID,
) -> dict:
    """
    Get a video reference by UUID.

    Args:
        client (RESTClient): The REST client.
        uuid (UUID): The video reference UUID.

    Returns:
        dict: The video reference.

    Raises:
        httpx.HTTPStatusError: If the API returns an error.
    """
    response = client.get(f"videoreferences/{uuid}")
    response.raise_for_status()
    return response.json()

videos

get(client, uuid)

Get a video by UUID.

Parameters:

Name	Type	Description	Default
client	RESTClient	The REST client.	required
uuid	UUID	The video UUID.	required

Returns:

Name	Type	Description
dict	dict	The video.

Raises:

Type	Description
HTTPStatusError	If the API returns an error.

Source code in dfg_tool/lib/m3/vampire_squid/videos.py

def get(
    client: RESTClient,
    uuid: UUID,
) -> dict:
    """
    Get a video by UUID.

    Args:
        client (RESTClient): The REST client.
        uuid (UUID): The video UUID.

    Returns:
        dict: The video.

    Raises:
        httpx.HTTPStatusError: If the API returns an error.
    """
    response = client.get(f"videos/{uuid}")
    response.raise_for_status()
    return response.json()

videosequences

get_by_name(client, name)

Get a video sequence by name.

Parameters:

Name	Type	Description	Default
client	RESTClient	The REST client.	required
name	str	The name of the video sequence.	required

Returns:

Name	Type	Description
dict	dict	The video sequence.

Raises:

Type	Description
HTTPStatusError	If the API returns an error.

Source code in dfg_tool/lib/m3/vampire_squid/videosequences.py

def get_by_name(
    client: RESTClient,
    name: str,
) -> dict:
    """
    Get a video sequence by name.

    Args:
        client (RESTClient): The REST client.
        name (str): The name of the video sequence.

    Returns:
        dict: The video sequence.

    Raises:
        httpx.HTTPStatusError: If the API returns an error.
    """
    response = client.get(f"videosequences/name/{name}")
    response.raise_for_status()
    return response.json()

models

Application data models.

timestamps

Timestamp parsing and formatting utilities.

format_timestamp(dt)

Format a datetime object as a timestamp string.

Parameters:

Name	Type	Description	Default
dt	datetime	The datetime object.	required

Returns:

Name	Type	Description
str	str	The formatted timestamp string.

Examples:

>>> format_timestamp(datetime(2021, 1, 1, 0, 0))
'2021-01-01T00:00:00.000000Z'

Source code in dfg_tool/lib/timestamps.py

def format_timestamp(dt: datetime) -> str:
    """
    Format a datetime object as a timestamp string.

    Args:
        dt (datetime): The datetime object.

    Returns:
        str: The formatted timestamp string.

    Examples:
        >>> format_timestamp(datetime(2021, 1, 1, 0, 0))
        '2021-01-01T00:00:00.000000Z'
    """
    return dt.strftime("%Y-%m-%dT%H:%M:%S.%fZ")

parse_timecode(timecode)

Parse a timecode string into a timedelta object.

Parameters:

Name	Type	Description	Default
timecode	str	The timecode string.	required

Returns:

Name	Type	Description
timedelta	timedelta	The parsed timedelta object.

Raises:

Type	Description
ValueError	If the timecode string is invalid.

Examples:

>>> parse_timecode("00:01:23")
datetime.timedelta(seconds=83)

>>> parse_timecode("01:23:45")
datetime.timedelta(seconds=5025)

Source code in dfg_tool/lib/timestamps.py

def parse_timecode(timecode: str) -> timedelta:
    """
    Parse a timecode string into a timedelta object.

    Args:
        timecode (str): The timecode string.

    Returns:
        timedelta: The parsed timedelta object.

    Raises:
        ValueError: If the timecode string is invalid.

    Examples:
        >>> parse_timecode("00:01:23")
        datetime.timedelta(seconds=83)

        >>> parse_timecode("01:23:45")
        datetime.timedelta(seconds=5025)
    """
    hours, minutes, seconds, *_ = timecode.split(":")
    return timedelta(
        hours=int(hours),
        minutes=int(minutes),
        seconds=int(seconds),
    )

parse_timestamp(timestamp)

Parse a timestamp string into a datetime object.

Parameters:

Name	Type	Description	Default
timestamp	str	The timestamp string.	required

Returns:

Name	Type	Description
datetime	datetime	The parsed datetime object.

Raises:

Type	Description
ValueError	If the timestamp string is invalid.

Examples:

>>> parse_timestamp("2021-01-01T00:00:00Z")
datetime.datetime(2021, 1, 1, 0, 0)

>>> parse_timestamp("2021-01-01T00:00:00.123Z")
datetime.datetime(2021, 1, 1, 0, 0, 0, 123000)

Source code in dfg_tool/lib/timestamps.py

def parse_timestamp(timestamp: str) -> datetime:
    """
    Parse a timestamp string into a datetime object.

    Args:
        timestamp (str): The timestamp string.

    Returns:
        datetime: The parsed datetime object.

    Raises:
        ValueError: If the timestamp string is invalid.

    Examples:
        >>> parse_timestamp("2021-01-01T00:00:00Z")
        datetime.datetime(2021, 1, 1, 0, 0)

        >>> parse_timestamp("2021-01-01T00:00:00.123Z")
        datetime.datetime(2021, 1, 1, 0, 0, 0, 123000)
    """
    for fmt in (
        "%Y-%m-%dT%H:%M:%S.%fZ",
        "%Y-%m-%dT%H:%M:%S.%f",
        "%Y-%m-%dT%H:%M:%SZ",
        "%Y-%m-%dT%H:%M:%S",
    ):
        try:
            return datetime.strptime(timestamp, fmt)
        except ValueError:
            pass
    raise ValueError(f"Invalid timestamp: '{timestamp}'")