59 lines
2.0 KiB
Python
59 lines
2.0 KiB
Python
"""단계 산출물의 blob 저장·회수
|
|
경로는 <파이프라인>/<task_id>/<이름>
|
|
"""
|
|
import io
|
|
|
|
from PIL import Image
|
|
|
|
from utils import blob
|
|
|
|
JPEG_QUALITY = 90
|
|
|
|
# 이름이 고정된 산출물. 경로가 정해져 있어 컬럼에 다시 적지 않고 만들어 쓴다
|
|
THUMBNAIL = "thumbnail.jpg"
|
|
|
|
|
|
def artifact_path(pipeline: str, task_id: str, name: str) -> str:
|
|
return f"{pipeline}/{task_id}/{name}"
|
|
|
|
|
|
def artifact_url(pipeline: str, task_id: str, name: str) -> str:
|
|
return blob.public_url(artifact_path(pipeline, task_id, name))
|
|
|
|
|
|
async def store_bytes(pipeline: str, task_id: str, name: str, data: bytes, *,
|
|
download_as: str | None = None) -> str:
|
|
"""download_as를 주면 브라우저가 그 주소를 열 때 그 이름으로 내려받는다"""
|
|
return await blob.upload_bytes(
|
|
data, artifact_path(pipeline, task_id, name),
|
|
disposition=blob.as_attachment(download_as) if download_as else None)
|
|
|
|
|
|
async def store_image(pipeline: str, task_id: str, name: str, image: Image.Image) -> str:
|
|
buffer = io.BytesIO()
|
|
if name.endswith(".png"):
|
|
image.save(buffer, "PNG")
|
|
else:
|
|
image.convert("RGB").save(buffer, "JPEG", quality=JPEG_QUALITY)
|
|
return await store_bytes(pipeline, task_id, name, buffer.getvalue())
|
|
|
|
|
|
async def load_image(url: str) -> Image.Image:
|
|
return Image.open(io.BytesIO(await blob.download_bytes(url)))
|
|
|
|
|
|
async def load_optional(url: str | None) -> bytes | None:
|
|
return await blob.download_bytes(url) if url else None
|
|
|
|
|
|
def contact_sheet(images: list[Image.Image], tile_width: int = 240,
|
|
gap: int = 8) -> Image.Image:
|
|
"""검수 화면에 한 줄로 늘어놓는 시트"""
|
|
tiles = [image.resize((tile_width, round(tile_width * image.height / image.width)))
|
|
for image in images]
|
|
height = max(tile.height for tile in tiles)
|
|
sheet = Image.new("RGB", ((tile_width + gap) * len(tiles) - gap, height), "black")
|
|
for order, tile in enumerate(tiles):
|
|
sheet.paste(tile, (order * (tile_width + gap), 0))
|
|
return sheet
|