Start Free

Centrally Collecting Files from Branches, Factories, and Edge Devices

Collect files from branch, factory, and edge locations with configured paths, conditions, status tracking, and offline-site handling.

IT EngineersDevelopers

Getting Started#

Core Concept#

Bring files generated across multiple sites into a single central environment

Business systems at branches, production equipment in factories, and edge devices in the field generate various files, including business documents, production data, logs, and result files.

Branch, factory, and edge data collection connects file-generation locations at each site to a central collection environment and automatically transfers files to headquarters servers or cloud storage according to configured conditions.

Branch A
Factory B
Edge Device C
Central Collection Environment
Headquarters Server
Cloud

Connecting files from each site into a single central collection flow lets you gather data generated across distributed environments into designated storage locations for subsequent analysis and business processing.

Collection Flow#

Detect files generated at sites and automatically move them to the central storage location

When a file is created or changed at a site, the configured collection conditions are checked and the files in scope are transferred to a central server or cloud.

  1. Site File Creation
  2. File Change Detection
  3. Check Collection Conditions
  4. Transfer to Central Environment
  5. Review Collection Result

You can configure collection jobs to start when files are created or changed, or according to a defined schedule. Collected files can then be used for analysis and follow-up work in the central environment.

Operational Benefits#

Manage data flows from multiple sites together from a central location

Each site may use different equipment, file-generation locations, and collection times. Connecting them to a central collection environment lets you manage files generated across multiple sites as a single flow.

Site Environment Generated Files Central Collection Location Use
Branch text text, text Headquarters Server Business Review
Factory Production Data, Inspection Results Cloud text, Quality Management
Edge Device Sensor Data, text Central Analysis Environment Data Processing

IT Engineer#

Collection Environment#

Connect branches, factories, edge devices, and file-generation locations

First, connect the branches, production equipment, and edge devices from which files will be collected to the central management environment.

Specify the folders or storage locations where files are generated on each device to configure the targets and file paths that collection jobs should monitor.

  • Branch-A
  • /data/report
  • Factory-01
  • /production/result
  • Edge-Server-01
  • /logs/device

Collection Paths#

Connect site-specific file locations to the central storage environment

After connecting the collection targets, configure each site's file path and central storage location as a single collection path.

You can gather files from each site to the headquarters server, or transfer them to cloud storage or an analysis environment according to the data type and intended use.

Branch A
Factory B
Edge Device C
Central Collection
Headquarters Server
Cloud Storage
collection source file path central storage text
Branch A /report/daily /data/branch
Factory B /production/result /data/factory
Edge C /logs/device /data/edge

Collection Conditions#

Start collection jobs based on file events and schedules

You can start collection jobs when files are created or changed, or configure them to collect required files according to a defined schedule.

You can define the collection scope based on file paths and types so that only the required data is collected from files generated at each site.

Collection Start Condition
File Creation
File Change
Recurring Schedule
File Collection

Collection Automation#

Connect central collection to data processing and result storage

After collecting data from multiple sites centrally, collection jobs can continue into analysis, transformation, or separate storage operations.

Branch Data
Factory Data
Edge Data
Central Collection
Data Processing
Result Storage

By connecting collected files to processing equipment, you can build an automated flow from site data collection → central transfer → data processing → result storage.

Collection Status#

Review collection jobs and processing results from multiple sites centrally

Runstext Dataset text text text Branchtext Factory, Edge Devicetext collection jobtext text reviewtext text text.

devicetext execution statustext text, processingtext file text, text execution resulttext text text Collection Statustext text.

  • Branch-A
  • StatusCompleted
  • Files128
  • Last RunCompleted
  • Factory-01
  • StatusRunning
  • Progress68%
  • Last RunIn Progress
  • Edge-Server-01
  • StatusCompleted
  • Files2,431

text text text sitetext File Collection jobtext text reviewtext text text, central text text text devicetext processing statustext text text text text.

Operational Response#

site textresult file processing statustext reviewtext required job again executiontext

When a collection job requires review, use the execution details and Activity Log to check the site's device connection, file path, access scope, and processing result.

Collection Job Execution
Execution Status Check
Completed
Review Result
Review Required
Check Device, Path, File Status
Adjust Configuration
Rerun Operation
Review Result

Developer#

Register collection automation for each site and aggregate connection status and collection metrics

Integration Preparation#

Prepare common request code and path representation

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

Determine the transfer status using the values below. There are five terminal states, and the successful value is Complete (2).

Status Value Meaning Terminal
2 Complete Yes
4 Error Yes
5 Cancelled Yes
9 Partially Complete Yes
99 Failed Yes
1, 6, 12, 13 Started, Transferring, Synchronizing, Receiving No

Configure Site List#

Keep site information as data and create collection jobs in bulk

When there are dozens of sites, creating them one by one in the UI is difficult. Keep the site list as data and iterate through it.

SITES = [
    {"device": "branch-seoul", "path": "/report/daily", "target": "/data/branch"},
    {"device": "branch-busan", "path": "/report/daily", "target": "/data/branch"},
    {"device": "factory-01", "path": "/production/result", "target": "/data/factory"},
    {"device": "edge-line-01", "path": "/logs/device", "target": "/data/edge"},
]

CENTRAL = "device-hq-01"


def site_target(site):
    # sites reuse the same file names, so put the site id in the target path
    return f"{site['target']}/{site['device']}"

If destination paths are not separated, files with the same name, such as result.csv, will overwrite one another between sites.

Register Collection Automation#

Create a collection job for each site and define its execution conditions

def build_collection(site, central, schedule=None, options=None):
    name = f"collect {site['device']}"
    sync = schedule is None

    body = {
        "name": name,
        "flowName": name,
        "transferType": "sync" if sync else "normal",
        "timezone": "Asia/Seoul",
        "step": 1,
        "isUpcoming": False,
        "details": [
            {
                "senderId": site["device"],
                "receiverId": central,
                "sourceItem": [
                    {
                        "hash": encode_path(site["device"], site["path"]),
                        "filePath": site["path"],
                        "isDir": True,
                    }
                ],
                "targetPath": encode_path(central, site_target(site)),
                "step": 1,
                "transferOptions": {
                    "noSchedule": sync,
                    "target-action": "numbering",
                    "send-fileoption": {},
                    **({"syncType": 1} if sync else {}),
                    **(options or {}),
                },
            }
        ],
        "schedules": [schedule or {
            "type": "none", "startDateType": "now",
            "startDate": now_iso(), "timezone": "Asia/Seoul",
        }],
    }

    return body


DAILY = {
    "type": "day", "startDateType": "now",
    "hour": "02", "minute": "00", "ampm": "am",
    "startDate": now_iso(), "timezone": "Asia/Seoul",
}

collectors = {
    site["device"]: api("POST", "/api/automations",
                        build_collection(site, CENTRAL, DAILY))["automationId"]
    for site in SITES
}

There are four items that must be followed in an automation request.

Item Specification
isUpcoming Must be false. The server default true ignores the schedule in the request and replaces it with a five-minute one-time schedule. Steps with triggerAutomation are forced to false by the server, so specify it directly only on the first step without a trigger.
step Include it at both the top level and in details. It represents the hop position in the flow.
sourceItem Include both hash (path token) and filePath (plain-text path).
syncType Put it inside transferOptions. 1 is one-way and 2 is bidirectional.

All four items can be omitted and registration will still succeed, but behavior changes at execution time. If a recurring schedule was registered but runs only once and stops, check isUpcoming first.

Execution Mode Configuration Suitable Site
Scheduled Execution transferType: normal + schedule Branches where files are created at a fixed time
Creation Detection transferType: sync + transferOptions.syncType Production equipment and edge devices where files are created intermittently

Because file-generation times differ by site, you do not have to standardize on a single method.

Preventing Duplicate Registration#

Prevent the same collection job from being created twice

A new automation is created even when an automation with the same name already exists. Iterating through the site list again would execute collection twice.

def find_automation(name):
    # the name we send is stored as flowName in the response
    # automationName is a server generated id like T4037-8500-1815, not the name we set.
    for page in range(1, 6):
        result = api("GET", "/api/automations",
                     params={"page": page, "size": 100, "search": name}) or {}

        items = [item
                 for flow in result.get("automations") or []
                 for item in flow.get("automations") or []]

        for item in items:
            if item.get("flowName") == name:
                return item

        if len(items) < 100:
            return None

    return None


def ensure_collection(site, central, schedule=None):
    name = f"collect {site['device']}"

    if find_automation(name):
        return None

    return api("POST", "/api/automations",
               build_collection(site, central, schedule))["automationId"]

The automation list is returned nested by flow group, so you must iterate through the inner arrays as well. Server search uses partial matching, so select only the item whose name exactly matches the name received.

Offline Site Response#

Find disconnected sites and send queued files after recovery

Site equipment can experience unstable network connectivity. The response differs depending on whether collection failed or the device was disconnected.

def site_state(device_id):
    state = api("GET", f"/api/devices/{device_id}/connectivity") or {}
    return bool(state.get("isConnected")), state.get("stateLabel")


for site in SITES:
    connected, label = site_state(site["device"])

    if not connected:
        print(f"{site['device']:20} {label}")

The connection status in the response is isConnected. The accompanying stateLabel can be used directly in the UI.

When the connection returns, send the files that accumulated during the outage in one batch. Compare the source and target file lists and select only missing files.

def list_files(device_id, path):
    found, page = {}, 1

    while True:
        result = api("GET", f"/api/devices/{device_id}/files", params={
            "path": path, "page": page, "size": 200, "type": "file",
        }) or {}

        for item in result.get("items") or []:
            found[item["name"]] = item.get("size")

        if page >= (result.get("lastPage") or 1):
            return found

        page += 1


def catch_up(site, central):
    source_files = list_files(site["device"], site["path"])
    missing = sorted(set(source_files) - set(list_files(central, site_target(site))))

    if not missing:
        return None

    # send file lists through sourceItem; sourcePaths treats every path as a folder
    return api("POST", "/api/transfers/manual", {
        "sourceDevice": site["device"],
        "targetDevice": central,
        "targetPath": site_target(site),
        "sourceItem": [
            {"path": f"{site['path'].rstrip('/')}/{name}",
             "isDir": False,
             "fileSize": source_files[name]}
            for name in missing
        ],
        "sendAllFolder": False,
        "transferOptions": {"target-action": "numbering"},
    })["monitorId"]

Passing the file size along lets the server skip querying the size of each item again, improving performance.

When the destination policy is numbering, files accumulate with numbered names. In that configuration, find files based on transfer history rather than comparing names.

Aggregate Collection Status#

Review collection results from multiple sites at once

def paginate(path, params=None, limit=200, max_pages=50):
    query = dict(params or {})
    query["limit"] = limit
    cursor = None

    for _ in range(max_pages):
        if cursor:
            query["cursor"] = cursor

        result = api("GET", path, params=query) or {}

        for record in result.get("data") or []:
            yield record

        pagination = result.get("pagination") or {}

        if not pagination.get("hasMore"):
            return

        cursor = pagination.get("nextCursor")

        if not cursor:
            return
from collections import Counter
from datetime import datetime, timedelta, timezone


def site_summary(device_id, days=1):
    end = datetime.now(timezone.utc)
    fmt = "%Y-%m-%dT%H:%M:%SZ"

    rows = list(paginate(f"/api/devices/{device_id}/transfer-history", params={
        "startDate": (end - timedelta(days=days)).strftime(fmt),
        "endDate": end.strftime(fmt),
    }))

    failed = [r for r in rows if r.get("status") in NOT_SUCCEEDED]
    return {"total": len(rows), "failed": len(failed)}


header = "{:20} {:^6} {:>6} {:>6}".format("site", "up", "collect", "fail")
print(header)
print("-" * len(header))

for site in SITES:
    connected, _ = site_state(site["device"])
    summary = site_summary(site["device"])

    print(f"{site['device']:20} {'O' if connected else 'X':^6}"
          f" {summary['total']:>6} {summary['failed']:>6}")

Transfer history is returned in the data.data array, and pagination information is returned in data.pagination.

Exception Handling#

Review failed collections and rerun them

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 device, automation_id in collectors.items():
    runs = api("GET", f"/api/automations/{automation_id}/executions") or []

    if not runs:
        print(f"{device}: no run history - check registration and start condition")
        continue

    latest = runs[0]

    if latest["status"] != STATUS_COMPLETE:
        print(f"{device}: retried {retry_failed(latest['monitorId'])} files")

execution text text text text text text. text text text text text text executiontext text text isUpcomingtext Started conditiontext reviewtext.

review Item review text
site device isConnectedtext stateLabel
collection job sitetext text execution condition
text path site text text storage text
Collection Status sitetext collection text Failed
text file text text text file