Automatically Transfer Media Sources and Processing Results Step by Step

Start하기

기본 개념

Automatically connect file collection through processing and result delivery

In media and data workflows, the location where files are created, the equipment that processes them, and the location where the results are used may all be different.

For example, you can collect video files generated by multiple recording devices, transfer them to a conversion server, store the processed results in storage, and then transfer them to the next work environment.

File Creation
    │
    ▼
File Collection
    │
    ▼
File Processing
    │
    ▼
Result Use

The result of each stage becomes the input for the next task, allowing you to connect everything from file creation through collection, processing, and result use in a single flow.

Automated Flow

Collect files from multiple devices and process them in a defined order

In an automated flow, files generated by multiple devices are collected as one task and the collected results are connected to the required processing equipment.

When processing is complete, you can transfer the resulting files to a designated location for use in the next task.

Source A ───┐
            │
Source B ───┼──→ Collect
            │       │
Source C ───┘       ▼
                  Process
                     │
                     ▼
                  Result

By configuring each next stage to run based on file status and task completion conditions, you can automatically connect file tasks in sequence.

Automation Benefits

Reduce repetitive file movement and management at each stage

When multi-stage file tasks are handled separately, file preparation, transfer, and processing-result checks are repeated.

By configuring an automated flow, you can connect the result of each previous stage to the next task and manage the process after file creation as one flow.

This allows file tasks distributed across multiple devices and systems to be connected in a single automated flow.

IT 엔지니어

Configure an automated processing flow for media and data files

Collection Environment

Connect file-generating devices to collection locations

First, connect the devices where files are generated to the locations where the files will be collected.

Configure environments where files are generated, such as recording devices, business servers, data collection equipment, and storage, as Sources, and specify the file path to use on each device.

Source Devices
      │
 ┌────┼────┐
 ▼    ▼    ▼
Cam  Server Storage
 │      │      │
 └──────┼──────┘
        ▼
    Collection

Configuring the collection environment lets you connect files generated in multiple locations into one automated flow.

Processing Connection

Connect collected files to the required processing equipment

Connect collected files to environments that perform the required tasks, such as conversion servers, analysis servers, and AI processing equipment.

You can consolidate files from multiple Sources into one processing device or connect them to different tasks based on file type and processing conditions.

Source A ───┐
            ├──→ Collect ───→ Process Server
Source B ───┤                       │
            │                       ▼
Source C ───┘                   Processing

When processing is complete, connect the resulting files to the next result stage.

Result Delivery

Store processing results and connect them to the next work environment

Transfer processed result files to locations used by the next task, such as storage, servers, and applications.

By connecting result storage and transfer to the next work environment as one result stage, you can automatically continue from processing completion to file use.

Processing
     │
     ▼
Result Files
     │
 ┌───┴───────┐
 ▼           ▼
Storage   Next System

If needed, you can extend the file-use flow by connecting notifications or additional processing tasks after result delivery.

Execution 조 matches

Run the next stage based on file and task status

Each stage can be configured to run according to specified conditions.

For example, you can configure the flow to start collection when a file is created, run processing when collection is complete, and start result delivery when the processing result is ready.

File Ready
    │
    ▼
Collect Complete?
    │
   Yes
    │
    ▼
Start Processing
    │
    ▼
Result Ready?
    │
   Yes
    │
    ▼
Deliver Result

By using file status and the result of the previous task as execution conditions, you can connect multiple tasks in a defined sequence.

Result Check

Check the overall flow and processing status by stage

When the automated flow runs, you can check the overall workflow status and the processing result of each stage together.

You can check how far the current task has progressed based on collected files, processing tasks, and result delivery status.

Workflow Run
     │
 ┌───┼──────────────┐
 ▼   ▼       ▼      ▼
Collect Process Result Complete
  ✓      ✓       ●
                 │
              Running

Checking the overall flow and individual stage statuses together lets you quickly understand the current progress and processing results.

Yes외 대응

Review stages that need attention and rerun the task

If additional review is required during execution, check the details and execution history for the relevant stage.

You can select the task requiring attention among file collection, processing equipment, and result delivery, check device connections, file paths, and processing status, and then rerun the required task.

Workflow Run
     │
     ▼
Status Check
     │
 ┌───┴────────┐
 ▼            ▼
Completed   Attention
                │
                ▼
          View Details
                │
                ▼
          Check Settings
                │
                ▼
              Retry
                │
                ▼
          Result Confirmed

Developers

Select targets by extension and size, collect them on processing equipment, and pass results to the next device

Integration Setup

Prepare common API-call code and path notation

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())

Transfer status is determined by the values below. There are five terminal states, and the value corresponding to success is Complete (2).

Target Selection

Filter files to send by extension, size, and name conditions

Specify filters in the transfer options so that only files matching the conditions are sent instead of the entire source folder.

def build_filter(exts=None, min_size=None, exclude=None):
    file_option = {}

    if exts:
        # extension whitelist, without the leading dot
        file_option["extension"] = {
            "extension": [e.lstrip(".").lower() for e in exts],
            "allow": True,
        }

    if min_size is not None:
        # over and equal both True means size or larger
        file_option["fileSize"] = {"size": min_size, "over": True, "equal": True}

    if exclude:
        # allow=False excludes files whose name contains this. Server matching is case sensitive.
        file_option["fileName"] = {"name": exclude, "allow": False}

    return {"send-fileoption": file_option} if file_option else {}

The extension filter uses send-fileoption.extension. send-filetype-cus The regular expression matches only the filename without its extension, so it does not work as an extension condition.

When multiple filters are provided, they are combined with AND. Only files that pass all filters are transferred.

FILTER = build_filter(exts=["mp4", "mov", "wav"], min_size=1024, exclude=".tmp")

Verify through a search that the conditions work as intended before adding them to the automation.

page = api("POST", f"/api/devices/{SOURCE_ID}/files/search",
           {"path": SOURCE_PATH, "pageSize": 500})

matched = [i for i in page["items"]
           if i["type"] == "file" and i["name"].lower().endswith((".mp4", ".mov"))]

print(f"{len(matched)} matched")

The search does not accept condition parameters, so evaluate the returned results.

File Collection

Collect files from multiple devices on one processing device

One transfer handles one source device. Create a transfer for each device and separate the destination path for each source.

SOURCES = [
    ("device-cam-01", "/media/raw"),
    ("device-cam-02", "/media/raw"),
    ("device-mic-01", "/audio/raw"),
]

for source, source_path in SOURCES:
    # give each source its own folder so file names do not collide
    target_path = f"/work/incoming/{source}"

    monitor_id = api("POST", "/api/transfers/manual", {
        "sourceDevice": source,
        "targetDevice": PROCESS_ID,
        "targetPath": target_path,
        "sourcePaths": [source_path],
        "sendAllFolder": True,
        "transferOptions": {"target-action": "numbering", **FILTER},
    })["monitorId"]

    print(source, "->", target_path, monitor_id)

When sources are separated by device, collection from the remaining devices continues even if one device is offline. If destination paths are not separated, files with the same name from different devices will overwrite one another.

Storage Location 제어

Define the folder structure below the destination path

TARGET_OPTIONS = {
    "savepath": True,          # keep the source folder structure (lowercase p)
    "optionPath": 3,           # how many trailing path segments to keep
    "target-action": "numbering",
}

If files with the same names arrive from multiple devices, optionPath increase the value to distinguish their sources. To organize files by date, put the date directly in the destination path rather than using an option.

from datetime import date

target_path = f"/work/incoming/{date.today():%Y/%m/%d}"

Stage Connection

Automatically start result delivery when collection is complete

Create each segment as an automation and group them under the same flowId. Put the previous stage's automationId in the next stage's triggerAutomation.value so the server continues execution.

import uuid

flow_id = str(uuid.uuid4())


def build_step(name, source, source_path, target, target_path,
               step, trigger_id=None, options=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": "numbering",
                    "send-fileoption": {},
                    **(options or {}),
                },
            }
        ],
        "schedules": [schedule],
    }

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

    return body


collect_id = api("POST", "/api/automations", build_step(
    "collect", "device-cam-01", "/media/raw",
    PROCESS_ID, "/work/incoming", step=1,
    options=FILTER, webhook=ENCODE_HOOK))["automationId"]

archive_id = api("POST", "/api/automations", build_step(
    "archive", PROCESS_ID, "/work/output",
    ARCHIVE_ID, "/archive", step=2,
    trigger_id=collect_id, options=TARGET_OPTIONS))["automationId"]

There are four items that must be followed in automation requests.

Registration succeeds even if all four items are omitted, but behavior changes at execution time. If a recurring schedule runs only once and then stops, check isUpcoming first.

The source and target of a stage must be different devices. Specifying the same device returns 400.

The processor is called after the transfer completes. Specify which events it responds to with config.events; to receive failures as well, check the status on the receiving side.

Changes Transfer

Send only files changed since the last execution

transfer = api("POST", "/api/transfers/manual", {
    "sourceDevice": "device-cam-01",
    "targetDevice": PROCESS_ID,
    "targetPath": "/work/incoming",
    "sourcePaths": ["/media/raw"],
    "sendAllFolder": True,
    "incremental": True,
    "transferOptions": {"target-action": "overwrite"},
})

The default is off. The device agent calculates the changes and transfers only files added or modified since the last execution.

For incremental transfers, the destination policy must be set to overwrite. With a numbering policy, modified files accumulate under new names, causing the processing equipment to keep reading older files.

Result Check과 Retransmission

Check each stage run and retransmit only failed files

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)

for step, automation_id in enumerate([collect_id, archive_id], start=1):
    runs = api("GET", f"/api/automations/{automation_id}/executions") or []
    latest = runs[0] if runs else {}

    if latest.get("status") not in (None, STATUS_COMPLETE):
        print(f"step {step} failed, retried {retry_failed(latest['monitorId'])} files")
        break

Execution history is returned in full without pagination, with the latest run at the beginning of the array.