imageupload.py

import asyncio
import os
from pathlib import Path

from dotenv import load_dotenv
from viam.rpc.dial import DialOptions, Credentials
from viam.app.viam_client import ViamClient

# ---------- config ----------
load_dotenv()

API_KEY = os.environ["VIAM_API_KEY"]
API_KEY_ID = os.environ["VIAM_API_KEY_ID"]
PART_ID = os.environ["PART_ID"]  # required for upload to associate image w part
SCRIPT_DIR = Path(__file__).parent
IMAGES_DIR = SCRIPT_DIR / "images"  # Change this to your images folder
TAGS = ["OK", "Volano", "SenzaDima"]
# --------------------------------


async def connect() -> ViamClient:
    return await ViamClient.create_from_dial_options(
        DialOptions(
            credentials=Credentials(type="api-key", payload=API_KEY),
            auth_entity=API_KEY_ID,
        )
    )


def get_image_files(directory: Path):
    exts = {".jpg", ".jpeg", ".png", ".bmp", ".gif"}
    return [f for f in directory.iterdir() if f.suffix.lower() in exts and f.is_file()]


async def main():
    client = await connect()
    try:
        image_files = get_image_files(IMAGES_DIR)
        if not image_files:
            print(f"No images found in {IMAGES_DIR}")
            return
        for img_path in image_files:
            fid = await client.data_client.file_upload_from_path(
                filepath=str(img_path),
                part_id=PART_ID,
                tags=TAGS,
            )
            print(f"✅ uploaded {img_path.name} as {fid}")
    finally:
        client.close()


if __name__ == "__main__":
    asyncio.run(main())