AI 데이터·모델 워크플로

import os
import requests

BASE_URL = os.getenv("INNORIX_BASE_URL", "https://app.innorix.com").rstrip("/")
TOKEN = os.environ["INNORIX_ACCESS_TOKEN"]
WORKSPACE_ID = os.getenv("INNORIX_WORKSPACE_ID")   # optional; falls back to the current workspace

STATUS_COMPLETE = 2
TERMINAL = {2, 4, 5, 9, 99}          # complete / error / cancelled / partial / failed
NOT_SUCCEEDED = {4, 5, 9, 99}


def api(method, path, body=None, params=None):
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {TOKEN}",
    }

    if WORKSPACE_ID:
        headers["x-workspace-id"] = WORKSPACE_ID

    response = requests.request(
        method, BASE_URL + path,
        headers=headers, json=body, params=params, timeout=30,
    )

    payload = response.json() if response.content else {}

    if not response.ok:
        raise RuntimeError(payload.get("message") or f"HTTP {response.status_code}")

    return payload.get("data")


def is_terminal(detail):
    return detail.get("isTerminal", detail.get("status") in TERMINAL)

import base64
import time


def encode_path(device_id, raw_path):
    normalized = str(raw_path or "").replace("\\", "/")
    token = base64.b64encode(normalized.encode("utf-8")).decode("ascii")
    return f"{device_id}_ino_{token}"


def now_iso():
    return time.strftime("%Y-%m-%dT%H:%M:%S.000Z", time.gmtime())

DATA_ROOT = "/data"
WORK_ROOT = "/work"
ARCHIVE_ROOT = "/archive"


# source dataset on the storage device
def dataset_path(dataset, version):
    return f"{DATA_ROOT}/datasets/{dataset}/{version}"


# path the training job reads on the node
def node_input_path(dataset, version, run_id):
    return f"{WORK_ROOT}/{run_id}/input/{dataset}/{version}"


# path the node writes its results to
def node_output_path(run_id):
    return f"{WORK_ROOT}/{run_id}/output"


# archive path where results are collected
def archive_path(run_id):
    return f"{ARCHIVE_ROOT}/runs/{run_id}"


run_id = f"r-{time.strftime('%Y%m%d-%H%M%S')}"

def deploy(storage, nodes, dataset, version, run_id):
    source_path = dataset_path(dataset, version)
    target_path = node_input_path(dataset, version, run_id)

    return {
        node: api("POST", "/api/transfers/manual", {
            "sourceDevice": storage,
            "targetDevice": node,
            "targetPath": target_path,
            "sourcePaths": [source_path],
            "sendAllFolder": True,
            "checkIntegrity": True,
            "transferOptions": {"target-action": "overwrite"},
        })["monitorId"]
        for node in nodes
    }

def wait(monitor_id, timeout=3600, interval=3):
    deadline = time.time() + timeout

    while time.time() < deadline:
        detail = api("GET", f"/api/transfers/{monitor_id}")

        if is_terminal(detail):
            return detail

        time.sleep(interval)

    raise TimeoutError(monitor_id)


def failed_files(monitor_id):
    result = api("GET", f"/api/transfers/{monitor_id}/files", params={
        "state": "any", "size": 500,
    }) or {}

    return [r for r in (result.get("children") or [])
            if r.get("status") in NOT_SUCCEEDED]


def retry_failed(monitor_id):
    rows = failed_files(monitor_id)

    if not rows:
        return 0

    api("POST", f"/api/transfers/{monitor_id}/retry", {
        "filesRetry": [
            {"filePath": r["sourceFilePath"], "isDir": bool(r.get("isFolder"))}
            for r in rows
        ]
    })

    return len(rows)

transfers = deploy(STORAGE, NODES, "imagenet", "v3", run_id)
results = {node: wait(mid) for node, mid in transfers.items()}

incomplete = [n for n, d in results.items() if d["status"] != STATUS_COMPLETE]

if incomplete:
    raise RuntimeError(f"nodes not fully delivered: {', '.join(incomplete)}")

monitor_id = api("POST", "/api/transfers/manual", {
    "sourceDevice": STORAGE,
    "targetDevice": node,
    "targetPath": node_input_path("imagenet", "v3", run_id),
    "sourcePaths": [dataset_path("imagenet", "v3")],
    "sendAllFolder": True,
    "incremental": True,
    "transferOptions": {"target-action": "overwrite"},
})["monitorId"]

detail = wait(monitor_id)
print(f"transferred {detail['fileCount']} files, {detail['totalSize']} bytes")

def verify(monitor_id, timeout=1800, interval=10):
    api("POST", f"/api/transfers/{monitor_id}/verification", {})

    deadline = time.time() + timeout

    while time.time() < deadline:
        result = api("GET", f"/api/transfers/{monitor_id}/verification") or {}

        if result.get("verified"):
            return result

        time.sleep(interval)

    raise TimeoutError(f"verification: {monitor_id}")


result = verify(monitor_id)

if result["sourceFileCount"] != result["targetFileCount"]:
    raise RuntimeError("file counts differ - part of the transfer is missing")

if not result["checksumMatched"]:
    for row in result.get("mismatchedFiles") or []:
        print("mismatch:", row)

def collect(nodes, archive, run_id):
    transfers = {}

    for node in nodes:
        target_path = f"{archive_path(run_id)}/{node}"

        transfers[node] = api("POST", "/api/transfers/manual", {
            "sourceDevice": node,
            "targetDevice": archive,
            "targetPath": target_path,
            "sourcePaths": [node_output_path(run_id)],
            "sendAllFolder": True,
            "checkIntegrity": True,
            "transferOptions": {"target-action": "overwrite"},
        })["monitorId"]

    return transfers

import uuid

flow_id = str(uuid.uuid4())


def build_step(name, source, source_path, target, target_path,
               step, trigger_id=None, webhook=None):
    schedule = {
        "type": "none",
        "startDateType": "now",
        "startDate": now_iso(),
        "timezone": "Asia/Seoul",
    }

    if trigger_id:
        schedule["triggerAutomation"] = {"value": trigger_id}

    body = {
        "name": name,
        "flowName": name,
        "flowId": flow_id,
        "transferType": "normal",
        "timezone": "Asia/Seoul",
        "step": step,
        "isUpcoming": False,
        "details": [
            {
                "senderId": source,
                "receiverId": target,
                "sourceItem": [
                    {
                        "hash": encode_path(source, source_path),
                        "filePath": source_path,
                        "isDir": True,
                    }
                ],
                "targetPath": encode_path(target, target_path),
                "step": step,
                "transferOptions": {
                    "noSchedule": False,
                    "target-action": "overwrite",
                    "send-fileoption": {},
                },
            }
        ],
        "schedules": [schedule],
    }

    if webhook:
        body["processors"] = [{
            "category": "run",
            "type": "http",
            "config": {"url": webhook, "method": "POST"},
        }]

    return body


deploy_id = api("POST", "/api/automations", build_step(
    f"deploy imagenet:v3", STORAGE, dataset_path("imagenet", "v3"),
    node, node_input_path("imagenet", "v3", run_id),
    step=1, webhook=TRAIN_HOOK))["automationId"]

collect_id = api("POST", "/api/automations", build_step(
    f"collect {run_id}", node, node_output_path(run_id),
    ARCHIVE, f"{archive_path(run_id)}/{node}",
    step=2, trigger_id=deploy_id))["automationId"]

def on_train_hook(payload):
    monitor_id = payload.get("monitorId")

    if monitor_id:
        detail = wait(monitor_id)

        if detail["status"] != STATUS_COMPLETE:
            return abort_run(payload)

    start_training(payload)

runs = api("GET", f"/api/automations/{deploy_id}/executions") or []
latest = runs[0] if runs else None

if latest and latest["status"] != STATUS_COMPLETE:
    print("deploy failed:", latest["monitorId"],
          "retried", retry_failed(latest["monitorId"]), "files")