aymara_ai.lib.examples_utils

aymara_ai.lib.examples_utils#

Attributes#

`logger`
`CacheMetadata`
`video_cache_dir`
`video_cache_videos_dir`
`video_cache_metadata_file`
`VIDEO_CACHE_DIR`
`VIDEO_CACHE_VIDEOS_DIR`
`VIDEO_CACHE_METADATA_FILE`

Functions#

`configure_video_cache`([path])	Override the cache directory location and return the resolved path.
`ensure_cache_dir`(*[, verbose])	Create the cache directory structure if it does not exist.
`setup_video_cache`([path, verbose])	Configure and initialize the cache directory, returning the resolved path.
`load_cache_metadata`()	Load cache metadata from JSON file.
`save_cache_metadata`(metadata)	Persist cache metadata to disk.
`add_to_cache`(local_path, *, provider, prompt, s3_uri)	Copy a generated video into the cache and update metadata.
`list_cached_videos`([extension])	Return all cached video files that match the expected extension.
`generate_presigned_url_from_s3_uri`(s3_uri[, expiration])	Convert an S3 URI (s3://bucket/key) into a pre-signed HTTP URL.
`validate_s3_bucket_configuration`(s3_client, bucket_name, *)	Validate that the configured S3 bucket exists and is accessible.
`generate_video_async_bedrock`(prompt, prompt_uuid, *, ...)	Generate a Nova Reel video and return the S3 URI. Returns None on moderation/failure.
`generate_video_async_sora`(prompt, prompt_uuid, *, ...)	Generate a Sora video, upload to S3, and return the S3 URI.
`upload_cached_video_async`(prompt_uuid, *, client)	Upload a cached video through the SDK and return the new file UUID.
`answer_prompts`(prompts, *, client[, provider, ...])	Generate/upload videos for prompts and build response payloads.
`display_eval_run_results`(client, eval_run_uuid, *[, ...])	Pretty-print scored responses for an evaluation run and embed videos when available.

Module Contents#

aymara_ai.lib.examples_utils.logger#

aymara_ai.lib.examples_utils.CacheMetadata#

aymara_ai.lib.examples_utils.video_cache_dir#

aymara_ai.lib.examples_utils.video_cache_videos_dir#

aymara_ai.lib.examples_utils.video_cache_metadata_file#

aymara_ai.lib.examples_utils.VIDEO_CACHE_DIR#

aymara_ai.lib.examples_utils.VIDEO_CACHE_VIDEOS_DIR#

aymara_ai.lib.examples_utils.VIDEO_CACHE_METADATA_FILE#

aymara_ai.lib.examples_utils.configure_video_cache(path=None)#

Override the cache directory location and return the resolved path.

Parameters:: path (str | pathlib.Path | None)
Return type:: pathlib.Path

aymara_ai.lib.examples_utils.ensure_cache_dir(*, verbose=True)#

Create the cache directory structure if it does not exist.

Parameters:: verbose (bool)
Return type:: pathlib.Path

aymara_ai.lib.examples_utils.setup_video_cache(path=None, *, verbose=True)#

Configure and initialize the cache directory, returning the resolved path.

Parameters:

path (str | pathlib.Path | None)
verbose (bool)

Return type:

pathlib.Path

aymara_ai.lib.examples_utils.load_cache_metadata()#

Load cache metadata from JSON file.

Return type:: CacheMetadata

aymara_ai.lib.examples_utils.save_cache_metadata(metadata)#

Persist cache metadata to disk.

Parameters:: metadata (CacheMetadata)
Return type:: None

aymara_ai.lib.examples_utils.add_to_cache(local_path, *, provider, prompt, s3_uri, verbose=True)#

Copy a generated video into the cache and update metadata.

Parameters:

local_path (pathlib.Path)
provider (str)
prompt (str)
s3_uri (str)
verbose (bool)

Return type:

pathlib.Path

aymara_ai.lib.examples_utils.list_cached_videos(extension='.mp4')#

Return all cached video files that match the expected extension.

Parameters:: extension (str)
Return type:: List[pathlib.Path]

aymara_ai.lib.examples_utils.generate_presigned_url_from_s3_uri(s3_uri, expiration=3600)#

Convert an S3 URI (s3://bucket/key) into a pre-signed HTTP URL.

Parameters:

s3_uri (str)
expiration (int)

Return type:

str

aymara_ai.lib.examples_utils.validate_s3_bucket_configuration(s3_client, bucket_name, *, default_bucket=None)#

Validate that the configured S3 bucket exists and is accessible.

Parameters:

s3_client (Any)
bucket_name (str)
default_bucket (Optional[str])

Return type:

str

async aymara_ai.lib.examples_utils.generate_video_async_bedrock(prompt, prompt_uuid, *, bedrock_client, s3_client, bucket_name, model_id)#

Generate a Nova Reel video and return the S3 URI. Returns None on moderation/failure.

Parameters:

prompt (str)
prompt_uuid (str)
bedrock_client (Any)
s3_client (Any)
bucket_name (str)
model_id (str)

Return type:

Optional[str]

async aymara_ai.lib.examples_utils.generate_video_async_sora(prompt, prompt_uuid, *, openai_client, s3_client, bucket_name, sora_output_folder, video_duration, model_id)#

Generate a Sora video, upload to S3, and return the S3 URI.

Parameters:

prompt (str)
prompt_uuid (str)
openai_client (Any)
s3_client (Any)
bucket_name (str)
sora_output_folder (str)
video_duration (int)
model_id (str)

Return type:

Optional[str]

async aymara_ai.lib.examples_utils.upload_cached_video_async(prompt_uuid, *, client)#

Upload a cached video through the SDK and return the new file UUID.

Parameters:

prompt_uuid (str)
client (Any)

Return type:

Optional[str]

async aymara_ai.lib.examples_utils.answer_prompts(prompts, *, client, provider='nova', provider_handlers=None, generate_video_async_bedrock=None, generate_video_async_sora=None, upload_cached_video_async=None)#

Generate/upload videos for prompts and build response payloads.

Parameters:

prompts (List[aymara_ai.types.eval_prompt.EvalPrompt])
client (Any)
provider (str)
provider_handlers (Optional[Dict[str, Callable[Ellipsis, Awaitable[Optional[str]]]]])
generate_video_async_bedrock (Optional[Callable[[str, str], Awaitable[Optional[str]]]])
generate_video_async_sora (Optional[Callable[[str, str], Awaitable[Optional[str]]]])
upload_cached_video_async (Optional[Callable[[str], Awaitable[Optional[str]]]])

Return type:

List[aymara_ai.types.eval_response_param.EvalResponseParam]

aymara_ai.lib.examples_utils.display_eval_run_results(client, eval_run_uuid, *, prompts=None, fallback_s3_bucket=None)#

Pretty-print scored responses for an evaluation run and embed videos when available.

Parameters:

client (Any)
eval_run_uuid (str)
prompts (Optional[List[aymara_ai.types.eval_prompt.EvalPrompt]])
fallback_s3_bucket (Optional[str])

Return type:

None