INNORIX Guide | Branch & Factory Edge Data Collection

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 ──────┐
Branch A ──────┐
Factory B ──────┼────→ Central Collection Environment ────→ Headquarters Server
             │              │
Factory B ──────┼────→ Central Collection Environment ────→ Headquarters Server

Edge Device C ─┘ └──────────→ 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.

Site File Creation
      │
      ▼
File Change Detection
      │
      ▼
Check Collection Conditions
      │
      ▼
Transfer to Central Environment
      │
      ▼
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 ────────┼──→ Central Collection ───→ Headquarters Server
               │         │
Edge Device C ───┘         └───────→ 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
               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 ────┼──→ Central Collection ───→ Data Processing
                │                       │
Edge Data ───────┘                       ▼
                                    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
├── Status      Completed
├── Files       128
└── Last Run    Completed

Factory-01
├── Status      Running
Branch-A
└── Last Run    In Progress

Edge-Server-01
├── Status      Completed
└── Files       2,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 Required
 │          │
 ▼          ▼
Review Result  Check Device · Path · File Status
                │
             Adjust Configuration
             Adjust Configuration
                │
             Rerun Operation
              Rerun Operation
                │
              Review Result
               Review Result

Developer#

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)

// InnorixClient.java
public static final String BASE_URL =
        env("INNORIX_BASE_URL", "https://app.innorix.com").replaceAll("/+
quot;, "");
public static final String WORKSPACE_ID = env("INNORIX_WORKSPACE_ID", null);

public static final int STATUS_COMPLETE = 2;
// States the transfer no longer moves out of
public static final Set<Integer> TERMINAL = Set.of(2, 4, 5, 9, 99);
// Terminal states that are not a full success
public static final Set<Integer> NOT_SUCCEEDED = Set.of(4, 5, 9, 99);

private HttpRequest.Builder headers(HttpRequest.Builder builder) {
    builder.header("Content-Type", "application/json")
           .header("Authorization", "Bearer " + session.accessToken());

    // When omitted the account's current workspace is used.
    if (workspaceId != null) builder.header("x-workspace-id", workspaceId);

    return builder;
}

/** Unwraps and returns data from the response. Throws ApiError on failure. */
public Object api(String method, String path, Object body, Map<String, Object> params) {
    Resp response = request(method, path, body, params);
    Object payload = null;

    try {
        payload = Json.parse(response.text());
    } catch (RuntimeException ignored) {
        payload = null;
    }

    if (!response.ok()) {
        Map<String, Object> map = Json.asObj(payload);
        String message = Json.str(map, "message", Json.str(map, "error", "unknown error"));
        throw new ApiError(response.status, message, map);
    }

    return Json.get(payload, "data");
}

/** Use the server flag when present, otherwise fall back to the status code. */
public static boolean isTerminal(Map<String, Object> record) {
    Boolean flag = Json.boolOrNull(record, "isTerminal");
    if (flag != null) return flag;

    Integer status = Json.intOrNull(record, "status");
    return status != null && TERMINAL.contains(status);
}

// innorix-client.js
const BASE_URL = (process.env.INNORIX_BASE_URL
  || "https://app.innorix.com").replace(/\/+$/, "");
const TOKEN = process.env.INNORIX_ACCESS_TOKEN;
const WORKSPACE_ID = process.env.INNORIX_WORKSPACE_ID || null;

export const STATUS_COMPLETE = 2;
export const TERMINAL = new Set([2, 4, 5, 9, 99]);       // complete / error / cancelled / partial / failed
export const NOT_SUCCEEDED = new Set([4, 5, 9, 99]);

export async function api(method, path, body = null, params = null) {
  const url = new URL(BASE_URL + path);

  if (params) {
    for (const [key, value] of Object.entries(params)) {
      if (value === undefined || value === null) continue;
      url.searchParams.set(key, String(value));
    }
  }

  const headers = {
    "Content-Type": "application/json",
    Authorization: `Bearer ${TOKEN}`,
  };

  // When omitted the account's current workspace is used.
  if (WORKSPACE_ID) headers["x-workspace-id"] = WORKSPACE_ID;

  const response = await fetch(url, {
    method,
    headers,
    body: body === null ? undefined : JSON.stringify(body),
    signal: AbortSignal.timeout(30_000),
  });

  const payload = await response.json().catch(() => ({}));

  if (!response.ok) {
    throw new Error(payload.message || `HTTP ${response.status}`);
  }

  return payload.data;
}

export function isTerminal(detail) {
  return detail.isTerminal !== undefined
    ? detail.isTerminal
    : TERMINAL.has(detail.status);
}

// InnorixClient.cs
public static readonly string BaseUrl =
    Env("INNORIX_BASE_URL", "https://app.innorix.com").TrimEnd('/');
public static readonly string WorkspaceIdFromEnv = Env("INNORIX_WORKSPACE_ID", null);

public const int StatusComplete = 2;
/// <summary>States the transfer no longer moves out of</summary>
public static readonly HashSet<int> Terminal = new HashSet<int> { 2, 4, 5, 9, 99 };
/// <summary>Terminal states that are not a full success</summary>
public static readonly HashSet<int> NotSucceeded = new HashSet<int> { 4, 5, 9, 99 };

// Applied on every request
request.Headers.TryAddWithoutValidation("Authorization", "Bearer " + Session.AccessToken);
// When omitted the account's current workspace is used.
if (WorkspaceId != null) request.Headers.TryAddWithoutValidation("x-workspace-id", WorkspaceId);

public async Task<JsonNode> ApiAsync(string method, string path, JsonNode body = null,
    IDictionary<string, object> parameters = null)
{
    Resp response = await RequestAsync(method, path, body, parameters).ConfigureAwait(false);
    JsonNode payload = null;

    try
    {
        payload = J.Parse(response.Text());
    }
    catch (Exception)
    {
        payload = null;
    }

    if (!response.Ok)
    {
        JsonObject map = J.AsObj(payload);
        string message = J.Str(map, "message", J.Str(map, "error", "unknown error"));
        throw new ApiError(response.Status, message, map);
    }

    return J.Get(payload, "data");
}

/// <summary>Use the server flag when present, otherwise fall back to the status code.</summary>
public static bool IsTerminal(JsonObject record)
{
    bool? flag = J.BoolOrNull(record, "isTerminal");
    if (flag != null) return flag.Value;

    int? status = J.IntOrNull(record, "status");
    return status != null && Terminal.Contains(status.Value);
}

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

public static String encodePath(String deviceId, String rawPath) {
    String normalized = (rawPath == null ? "" : rawPath).replace("\\", "/");
    return deviceId + "_ino_"
            + Base64.getEncoder().encodeToString(normalized.getBytes(StandardCharsets.UTF_8));
}

public static String nowIso() {
    return Instant.now().truncatedTo(ChronoUnit.SECONDS).toString().replace("Z", ".000Z");
}

export function encodePath(deviceId, rawPath) {
  const normalized = String(rawPath ?? "").replace(/\\/g, "/");
  const token = Buffer.from(normalized, "utf8").toString("base64");
  return `${deviceId}_ino_${token}`;
}

export function nowIso() {
  return new Date().toISOString().replace(/\.\d{3}Z$/, ".000Z");
}

public static string EncodePath(string deviceId, string rawPath)
{
    string normalized = (rawPath ?? "").Replace("\\", "/");
    return deviceId + "_ino_" + Convert.ToBase64String(Encoding.UTF8.GetBytes(normalized));
}

public static string NowIso()
{
    return DateTime.UtcNow.ToString("yyyy-MM-dd'T'HH:mm:ss'.000Z'",
        System.Globalization.CultureInfo.InvariantCulture);
}

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']}"

record Site(String device, String path, String target) {}

static final List<Site> SITES = List.of(
        new Site("branch-seoul", "/report/daily", "/data/branch"),
        new Site("branch-busan", "/report/daily", "/data/branch"),
        new Site("factory-01", "/production/result", "/data/factory"),
        new Site("edge-line-01", "/logs/device", "/data/edge"));

static final String CENTRAL = "device-hq-01";

String siteTarget(Site site) {
    // sites reuse the same file names, so put the site id in the target path
    return site.target() + "/" + site.device();
}

const 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" },
];

const CENTRAL = "device-hq-01";

// sites reuse the same file names, so put the site id in the target path
const siteTarget = (site) => `${site.target}/${site.device}`;

public record Site(string Device, string Path, string Target);

static readonly List<Site> Sites = new()
{
    new Site("branch-seoul", "/report/daily", "/data/branch"),
    new Site("branch-busan", "/report/daily", "/data/branch"),
    new Site("factory-01", "/production/result", "/data/factory"),
    new Site("edge-line-01", "/logs/device", "/data/edge"),
};

const string Central = "device-hq-01";

// sites reuse the same file names, so put the site id in the target path
string SiteTarget(Site site) => 
quot;{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
}

Map<String, Object> buildCollection(Site site, String central,
                                    Map<String, Object> schedule,
                                    Map<String, Object> options) {
    String name = "collect " + site.device();
    boolean sync = schedule == null;

    Map<String, Object> transferOptions = new LinkedHashMap<>(Json.newObj(
            "noSchedule", sync, "target-action", "numbering",
            "send-fileoption", Json.newObj()));
    if (sync) transferOptions.put("syncType", 1);
    if (options != null) transferOptions.putAll(options);

    Map<String, Object> detail = Json.newObj(
            "senderId", site.device(), "receiverId", central,
            "sourceItem", List.of(Json.newObj(
                    "hash", InnorixClient.encodePath(site.device(), site.path()),
                    "filePath", site.path(), "isDir", true)),
            "targetPath", InnorixClient.encodePath(central, siteTarget(site)),
            "step", 1, "transferOptions", transferOptions);

    Map<String, Object> defaultSchedule = Json.newObj(
            "type", "none", "startDateType", "now",
            "startDate", InnorixClient.nowIso(), "timezone", "Asia/Seoul");

    return Json.newObj(
            "name", name, "flowName", name,
            "transferType", sync ? "sync" : "normal",
            "timezone", "Asia/Seoul", "step", 1, "isUpcoming", false,
            "details", List.of(detail),
            "schedules", List.of(schedule != null ? schedule : defaultSchedule));
}

function buildCollection(site, central, schedule = null, options = null) {
  const name = `collect ${site.device}`;
  const sync = schedule === null;

  const transferOptions = {
    noSchedule: sync,
    "target-action": "numbering",
    "send-fileoption": {},
    ...(sync ? { syncType: 1 } : {}),
    ...(options || {}),
  };

  const detail = {
    senderId: site.device,
    receiverId: central,
    sourceItem: [{
      hash: encodePath(site.device, site.path),
      filePath: site.path,
      isDir: true,
    }],
    targetPath: encodePath(central, siteTarget(site)),
    step: 1,
    transferOptions,
  };

  return {
    name,
    flowName: name,
    transferType: sync ? "sync" : "normal",
    timezone: "Asia/Seoul",
    step: 1,
    isUpcoming: false,
    details: [detail],
    schedules: [schedule ?? {
      type: "none", startDateType: "now",
      startDate: nowIso(), timezone: "Asia/Seoul",
    }],
  };
}

JsonObject BuildCollection(Site site, string central,
    JsonObject schedule = null, JsonObject options = null)
{
    string name = 
quot;collect {site.Device}";
    bool sync = schedule == null;

    var transferOptions = new JsonObject
    {
        ["noSchedule"] = sync,
        ["target-action"] = "numbering",
        ["send-fileoption"] = new JsonObject(),
    };
    if (sync) transferOptions["syncType"] = 1;
    if (options != null)
        foreach (var kv in options) transferOptions[kv.Key] = kv.Value?.DeepClone();

    var detail = new JsonObject
    {
        ["senderId"] = site.Device,
        ["receiverId"] = central,
        ["sourceItem"] = new JsonArray
        {
            new JsonObject
            {
                ["hash"] = InnorixClient.EncodePath(site.Device, site.Path),
                ["filePath"] = site.Path,
                ["isDir"] = true,
            },
        },
        ["targetPath"] = InnorixClient.EncodePath(central, SiteTarget(site)),
        ["step"] = 1,
        ["transferOptions"] = transferOptions,
    };

    return new JsonObject
    {
        ["name"] = name,
        ["flowName"] = name,
        ["transferType"] = sync ? "sync" : "normal",
        ["timezone"] = "Asia/Seoul",
        ["step"] = 1,
        ["isUpcoming"] = false,
        ["details"] = new JsonArray { detail },
        ["schedules"] = new JsonArray
        {
            schedule ?? new JsonObject
            {
                ["type"] = "none", ["startDateType"] = "now",
                ["startDate"] = InnorixClient.NowIso(), ["timezone"] = "Asia/Seoul",
            },
        },
    };
}

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"]

Map<String, Object> findAutomation(String 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 (int page = 1; page <= 5; page++) {
        Map<String, Object> result = client.apiObj("GET", "/api/automations", null,
                Json.newObj("page", page, "size", 100, "search", name));

        List<Map<String, Object>> items = new ArrayList<>();
        for (Object flow : Json.arrOf(result, "automations")) {
            for (Object item : Json.arrOf(Json.asObj(flow), "automations")) {
                items.add(Json.asObj(item));
            }
        }

        for (Map<String, Object> item : items) {
            if (name.equals(Json.str(item, "flowName"))) return item;
        }

        if (items.size() < 100) return null;
    }

    return null;
}

async function findAutomation(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 (let page = 1; page <= 5; page += 1) {
    const result = (await client.api("GET", "/api/automations", null,
      { page, size: 100, search: name })) || {};

    const items = (result.automations || [])
      .flatMap((flow) => flow.automations || []);

    for (const item of items) {
      if (item.flowName === name) return item;
    }

    if (items.length < 100) return null;
  }

  return null;
}

async Task<JsonObject> FindAutomationAsync(string 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 (int page = 1; page <= 5; page++)
    {
        JsonObject result = await client.ApiObjAsync("GET", "/api/automations", null,
            new Dictionary<string, object> { ["page"] = page, ["size"] = 100, ["search"] = name });

        var items = J.ArrOf(result, "automations")
            .SelectMany(flow => J.ArrOf(J.AsObj(flow), "automations"))
            .Select(J.AsObj).ToList();

        foreach (JsonObject item in items)
        {
            if (J.Str(item, "flowName") == name) return item;
        }

        if (items.Count < 100) return null;
    }

    return null;
}

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}")

Object[] siteState(String deviceId) {
    Map<String, Object> state = client.apiObj("GET",
            "/api/devices/" + deviceId + "/connectivity");
    return new Object[]{Json.bool(state, "isConnected", false), Json.str(state, "stateLabel")};
}

for (Site site : SITES) {
    Object[] s = siteState(site.device());
    if (!(boolean) s[0]) {
        System.out.printf("%-20s %s%n", site.device(), s[1]);
    }
}

async function siteState(deviceId) {
  const state = (await client.api("GET",
    `/api/devices/${deviceId}/connectivity`)) || {};
  return [Boolean(state.isConnected), state.stateLabel];
}

for (const site of SITES) {
  const [connected, label] = await siteState(site.device);
  if (!connected) {
    console.log(`${site.device.padEnd(20)} ${label}`);
  }
}

async Task<(bool Connected, string Label)> SiteStateAsync(string deviceId)
{
    JsonObject state = await client.ApiObjAsync("GET",
        
quot;/api/devices/{deviceId}/connectivity");
    return (J.Bool(state, "isConnected", false), J.Str(state, "stateLabel"));
}

foreach (Site site in Sites)
{
    var (connected, label) = await SiteStateAsync(site.Device);
    if (!connected)
    {
        Console.WriteLine(
quot;{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"]

Map<String, Long> listFiles(String deviceId, String path) {
    Map<String, Long> found = new LinkedHashMap<>();
    int page = 1;

    while (true) {
        Map<String, Object> result = client.apiObj("GET",
                "/api/devices/" + deviceId + "/files", null,
                Json.newObj("path", path, "page", page, "size", 200, "type", "file"));

        for (Object node : Json.arrOf(result, "items")) {
            Map<String, Object> item = Json.asObj(node);
            found.put(Json.str(item, "name"), Json.longOrNull(item, "size"));
        }

        if (page >= Json.intOr(result, "lastPage", 1)) return found;

        page++;
    }
}

String catchUp(Site site, String central) {
    Map<String, Long> sourceFiles = listFiles(site.device(), site.path());
    Set<String> missing = new TreeSet<>(sourceFiles.keySet());
    missing.removeAll(listFiles(central, siteTarget(site)).keySet());

    if (missing.isEmpty()) return null;

    List<Object> items = new ArrayList<>();
    for (String name : missing) {
        items.add(Json.newObj(
                "path", site.path().replaceAll("/+
quot;, "") + "/" + name,
                "isDir", false, "fileSize", sourceFiles.get(name)));
    }

    Map<String, Object> transfer = client.apiObj("POST", "/api/transfers/manual", Json.newObj(
            "sourceDevice", site.device(),
            "targetDevice", central,
            "targetPath", siteTarget(site),
            "sourceItem", items,
            "sendAllFolder", false,
            "transferOptions", Json.newObj("target-action", "numbering")));

    return Json.str(transfer, "monitorId");
}

async function listFiles(deviceId, path) {
  const found = new Map();
  let page = 1;

  for (;;) {
    const result = (await client.api("GET", `/api/devices/${deviceId}/files`, null, {
      path, page, size: 200, type: "file",
    })) || {};

    for (const item of result.items || []) found.set(item.name, item.size);

    if (page >= (result.lastPage || 1)) return found;

    page += 1;
  }
}

async function catchUp(site, central) {
  const sourceFiles = await listFiles(site.device, site.path);
  const targetFiles = await listFiles(central, siteTarget(site));
  const missing = [...sourceFiles.keys()].filter((n) => !targetFiles.has(n)).sort();

  if (missing.length === 0) return null;

  const transfer = await client.api("POST", "/api/transfers/manual", {
    sourceDevice: site.device,
    targetDevice: central,
    targetPath: siteTarget(site),
    sourceItem: missing.map((name) => ({
      path: `${site.path.replace(/\/+$/, "")}/${name}`,
      isDir: false,
      fileSize: sourceFiles.get(name),
    })),
    sendAllFolder: false,
    transferOptions: { "target-action": "numbering" },
  });

  return transfer.monitorId;
}

async Task<Dictionary<string, long?>> ListFilesAsync(string deviceId, string path)
{
    var found = new Dictionary<string, long?>();
    int page = 1;

    while (true)
    {
        JsonObject result = await client.ApiObjAsync("GET",
            
quot;/api/devices/{deviceId}/files", null,
            new Dictionary<string, object>
            {
                ["path"] = path, ["page"] = page, ["size"] = 200, ["type"] = "file",
            });

        foreach (JsonNode node in J.ArrOf(result, "items"))
        {
            JsonObject item = J.AsObj(node);
            found[J.Str(item, "name")] = J.LongOrNull(item, "size");
        }

        if (page >= J.Int(result, "lastPage", 1)) return found;

        page++;
    }
}

async Task<string> CatchUpAsync(Site site, string central)
{
    var sourceFiles = await ListFilesAsync(site.Device, site.Path);
    var targetFiles = await ListFilesAsync(central, SiteTarget(site));

    var missing = sourceFiles.Keys.Where(n => !targetFiles.ContainsKey(n))
        .OrderBy(n => n).ToList();

    if (missing.Count == 0) return null;

    var items = new JsonArray();
    foreach (string name in missing)
    {
        items.Add(new JsonObject
        {
            ["path"] = 
quot;{site.Path.TrimEnd('/')}/{name}",
            ["isDir"] = false,
            ["fileSize"] = sourceFiles[name],
        });
    }

    JsonObject transfer = await client.ApiObjAsync("POST", "/api/transfers/manual",
        new JsonObject
        {
            ["sourceDevice"] = site.Device,
            ["targetDevice"] = central,
            ["targetPath"] = SiteTarget(site),
            ["sourceItem"] = items,
            ["sendAllFolder"] = false,
            ["transferOptions"] = new JsonObject { ["target-action"] = "numbering" },
        });

    return J.Str(transfer, "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

public List<Map<String, Object>> paginate(String path, Map<String, Object> params,
                                          int limit, int maxPages) {
    Map<String, Object> query = new LinkedHashMap<>(params == null ? Map.of() : params);
    query.put("limit", limit);

    List<Map<String, Object>> rows = new ArrayList<>();
    String cursor = null;

    for (int page = 0; page < maxPages; page++) {
        if (cursor != null) query.put("cursor", cursor);

        Map<String, Object> result = apiObj("GET", path, null, query);

        for (Object record : Json.arrOf(result, "data")) rows.add(Json.asObj(record));

        Map<String, Object> pagination = Json.objOf(result, "pagination");
        if (!Json.bool(pagination, "hasMore", false)) return rows;

        cursor = Json.str(pagination, "nextCursor");
        if (cursor == null) return rows;
    }

    return rows;
}

export async function* paginate(path, params = {}, limit = 200, maxPages = 50) {
  const query = { ...params, limit };
  let cursor = null;

  for (let page = 0; page < maxPages; page += 1) {
    if (cursor) query.cursor = cursor;

    const result = (await api("GET", path, null, query)) || {};

    for (const record of result.data || []) yield record;

    const pagination = result.pagination || {};

    if (!pagination.hasMore || !pagination.nextCursor) return;

    cursor = pagination.nextCursor;
  }
}

public async Task<List<JsonObject>> PaginateAsync(string path,
    IDictionary<string, object> parameters = null, int limit = 200, int maxPages = 50)
{
    var query = new Dictionary<string, object>(
        parameters ?? new Dictionary<string, object>()) { ["limit"] = limit };

    var rows = new List<JsonObject>();
    string cursor = null;

    for (int page = 0; page < maxPages; page++)
    {
        if (cursor != null) query["cursor"] = cursor;

        JsonObject result = await ApiObjAsync("GET", path, null, query).ConfigureAwait(false);

        foreach (JsonNode record in J.ArrOf(result, "data")) rows.Add(J.AsObj(record));

        JsonObject pagination = J.ObjOf(result, "pagination");
        if (!J.Bool(pagination, "hasMore", false)) return rows;

        cursor = J.Str(pagination, "nextCursor");
        if (cursor == null) return rows;
    }

    return rows;
}

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}")

Map<String, Integer> siteSummary(String deviceId, int days) {
    Instant end = Instant.now();
    DateTimeFormatter fmt = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss'Z'")
            .withZone(ZoneOffset.UTC);

    List<Map<String, Object>> rows = client.paginate(
            "/api/devices/" + deviceId + "/transfer-history",
            Json.newObj("startDate", fmt.format(end.minus(days, ChronoUnit.DAYS)),
                        "endDate", fmt.format(end)), 200, 50);

    int failed = 0;
    for (Map<String, Object> row : rows) {
        Integer status = Json.intOrNull(row, "status");
        if (status != null && InnorixClient.NOT_SUCCEEDED.contains(status)) failed++;
    }

    return Json.newObj("total", rows.size(), "failed", failed);
}

System.out.printf("%-20s %^6s %6s %6s%n", "site", "up", "collect", "fail");
for (Site site : SITES) {
    Object[] state = siteState(site.device());
    Map<String, Integer> summary = siteSummary(site.device(), 1);
    System.out.printf("%-20s %^6s %6d %6d%n", site.device(),
            (boolean) state[0] ? "O" : "X",
            summary.get("total"), summary.get("failed"));
}

async function siteSummary(deviceId, days = 1) {
  const end = new Date();
  const fmt = (d) => d.toISOString().replace(/\.\d{3}Z$/, "Z");

  const rows = [];
  for await (const row of paginate(`/api/devices/${deviceId}/transfer-history`, {
    startDate: fmt(new Date(end.getTime() - days * 86400000)),
    endDate: fmt(end),
  })) rows.push(row);

  const failed = rows.filter((r) => NOT_SUCCEEDED.has(r.status)).length;
  return { total: rows.length, failed };
}

console.log("site".padEnd(20), "up".padStart(6), "collect".padStart(6), "fail".padStart(6));
for (const site of SITES) {
  const [connected] = await siteState(site.device);
  const summary = await siteSummary(site.device);
  console.log(site.device.padEnd(20),
    (connected ? "O" : "X").padStart(6),
    String(summary.total).padStart(6),
    String(summary.failed).padStart(6));
}

async Task<(int Total, int Failed)> SiteSummaryAsync(string deviceId, int days = 1)
{
    DateTime end = DateTime.UtcNow;
    const string Fmt = "yyyy-MM-dd'T'HH:mm:ss'Z'";

    List<JsonObject> rows = await client.PaginateAsync(
        
quot;/api/devices/{deviceId}/transfer-history",
        new Dictionary<string, object>
        {
            ["startDate"] = end.AddDays(-days).ToString(Fmt),
            ["endDate"] = end.ToString(Fmt),
        });

    int failed = rows.Count(r =>
    {
        int? status = J.IntOrNull(r, "status");
        return status != null && InnorixClient.NotSucceeded.Contains(status.Value);
    });

    return (rows.Count, failed);
}

Console.WriteLine(
quot;{"site",-20} {"up",6} {"collect",6} {"fail",6}");
foreach (Site site in Sites)
{
    var (connected, _) = await SiteStateAsync(site.Device);
    var summary = await SiteSummaryAsync(site.Device);
    Console.WriteLine(
quot;{site.Device,-20} {(connected ? "O" : "X"),6} "
        + 
quot;{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)

public Map<String, Object> wait(String monitorId, int timeoutSeconds, int intervalSeconds) {
    long deadline = System.currentTimeMillis() + timeoutSeconds * 1000L;

    while (System.currentTimeMillis() < deadline) {
        Map<String, Object> detail = apiObj("GET", "/api/transfers/" + monitorId);

        if (isTerminal(detail)) return detail;

        sleep(intervalSeconds * 1000L);
    }

    throw new ApiError(0, "timeout waiting for " + monitorId);
}

public List<Map<String, Object>> failedFiles(String monitorId) {
    List<Map<String, Object>> failed = new ArrayList<>();

    for (Map<String, Object> row : transferFiles(monitorId)) {
        Integer status = Json.intOrNull(row, "status");
        if (status != null && NOT_SUCCEEDED.contains(status)) failed.add(row);
    }

    return failed;
}

public int retryFailed(String monitorId) {
    List<Map<String, Object>> rows = failedFiles(monitorId);
    if (rows.isEmpty()) return 0;

    List<Object> filesRetry = new ArrayList<>();

    for (Map<String, Object> row : rows) {
        String path = Json.str(row, "sourceFilePath");
        if (path == null) continue;
        filesRetry.add(Json.newObj("filePath", path, "isDir", Json.bool(row, "isFolder", false)));
    }

    api("POST", "/api/transfers/" + monitorId + "/retry", Json.newObj("filesRetry", filesRetry));

    return rows.size();
}

const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));

export async function wait(monitorId, { timeout = 3600, interval = 3 } = {}) {
  const deadline = Date.now() + timeout * 1000;

  while (Date.now() < deadline) {
    const detail = await api("GET", `/api/transfers/${monitorId}`);

    if (isTerminal(detail)) return detail;

    await sleep(interval * 1000);
  }

  throw new Error(`timeout waiting for ${monitorId}`);
}

export async function failedFiles(monitorId) {
  const result = (await api("GET", `/api/transfers/${monitorId}/files`, null, {
    state: "any", size: 500,
  })) || {};

  return (result.children || []).filter((row) => NOT_SUCCEEDED.has(row.status));
}

export async function retryFailed(monitorId) {
  const rows = await failedFiles(monitorId);

  if (rows.length === 0) return 0;

  await api("POST", `/api/transfers/${monitorId}/retry`, {
    filesRetry: rows
      .filter((row) => row.sourceFilePath)
      .map((row) => ({ filePath: row.sourceFilePath, isDir: Boolean(row.isFolder) })),
  });

  return rows.length;
}

public async Task<JsonObject> WaitAsync(string monitorId, int timeoutSeconds = 3600,
    int intervalSeconds = 3)
{
    long deadline = Environment.TickCount64 + timeoutSeconds * 1000L;

    while (Environment.TickCount64 < deadline)
    {
        JsonObject detail = await ApiObjAsync("GET", "/api/transfers/" + monitorId)
            .ConfigureAwait(false);

        if (IsTerminal(detail)) return detail;

        await Task.Delay(intervalSeconds * 1000).ConfigureAwait(false);
    }

    throw new ApiError(0, "timeout waiting for " + monitorId);
}

public async Task<List<JsonObject>> FailedFilesAsync(string monitorId)
{
    List<JsonObject> rows = await TransferFilesAsync(monitorId).ConfigureAwait(false);

    return rows.Where(row =>
    {
        int? status = J.IntOrNull(row, "status");
        return status != null && NotSucceeded.Contains(status.Value);
    }).ToList();
}

/// <summary>Only callable once the transfer has reached a terminal state.</summary>
public async Task<int> RetryFailedAsync(string monitorId)
{
    List<JsonObject> rows = await FailedFilesAsync(monitorId).ConfigureAwait(false);
    if (rows.Count == 0) return 0;

    var filesRetry = new JsonArray();

    foreach (JsonObject row in rows)
    {
        string path = J.Str(row, "sourceFilePath");
        if (path == null) continue;

        filesRetry.Add(new JsonObject
        {
            ["filePath"] = path,
            ["isDir"] = J.Bool(row, "isFolder", false),
        });
    }

    await ApiAsync("POST", "/api/transfers/" + monitorId + "/retry",
        new JsonObject { ["filesRetry"] = filesRetry }).ConfigureAwait(false);

    return rows.Count;
}

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

for (var entry : collectors.entrySet()) {
    List<Map<String, Object>> runs = client.executions(entry.getValue());

    if (runs.isEmpty()) {
        System.out.println(entry.getKey()
                + ": no run history - check registration and start condition");
        continue;
    }

    Map<String, Object> latest = runs.get(0);

    if (Json.intOr(latest, "status", -1) != InnorixClient.STATUS_COMPLETE) {
        int count = client.retryFailed(Json.str(latest, "monitorId"));
        System.out.println(entry.getKey() + ": retried " + count + " files");
    }
}

for (const [device, automationId] of Object.entries(collectors)) {
  const runs = (await client.executions(automationId)) || [];

  if (runs.length === 0) {
    console.log(`${device}: no run history - check registration and start condition`);
    continue;
  }

  const latest = runs[0];

  if (latest.status !== STATUS_COMPLETE) {
    console.log(`${device}: retried ${await retryFailed(latest.monitorId)} files`);
  }
}

foreach (var (device, automationId) in collectors)
{
    List<JsonObject> runs = J.AsList(await client.ExecutionsAsync(automationId));

    if (runs.Count == 0)
    {
        Console.WriteLine(
quot;{device}: no run history - check registration and start condition");
        continue;
    }

    JsonObject latest = runs[0];

    if (J.IntOrNull(latest, "status") != InnorixClient.StatusComplete)
    {
        int count = await client.RetryFailedAsync(J.Str(latest, "monitorId"));
        Console.WriteLine(
quot;{device}: retried {count} 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	`isConnected`text `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

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 A ──────┐
Branch A ──────┐
Factory B ──────┼────→ Central Collection Environment ────→ Headquarters Server
             │              │
Factory B ──────┼────→ Central Collection Environment ────→ Headquarters Server

Edge Device C ─┘ └──────────→ Cloud

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.

Site File Creation
      │
      ▼
File Change Detection
      │
      ▼
Check Collection Conditions
      │
      ▼
Transfer to Central Environment
      │
      ▼
Review Collection Result

Operational Benefits#

Manage data flows from multiple sites together from a central location

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 ────────┼──→ Central Collection ───→ Headquarters Server
               │         │
Edge Device C ───┘         └───────→ 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
               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 ────┼──→ Central Collection ───→ Data Processing
                │                       │
Edge Data ───────┘                       ▼
                                    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
├── Status      Completed
├── Files       128
└── Last Run    Completed

Factory-01
├── Status      Running
Branch-A
└── Last Run    In Progress

Edge-Server-01
├── Status      Completed
└── Files       2,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 Required
 │          │
 ▼          ▼
Review Result  Check Device · Path · File Status
                │
             Adjust Configuration
             Adjust Configuration
                │
             Rerun Operation
              Rerun Operation
                │
              Review Result
               Review Result

Developer#

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)

// InnorixClient.java
public static final String BASE_URL =
        env("INNORIX_BASE_URL", "https://app.innorix.com").replaceAll("/+
quot;, "");
public static final String WORKSPACE_ID = env("INNORIX_WORKSPACE_ID", null);

public static final int STATUS_COMPLETE = 2;
// States the transfer no longer moves out of
public static final Set<Integer> TERMINAL = Set.of(2, 4, 5, 9, 99);
// Terminal states that are not a full success
public static final Set<Integer> NOT_SUCCEEDED = Set.of(4, 5, 9, 99);

private HttpRequest.Builder headers(HttpRequest.Builder builder) {
    builder.header("Content-Type", "application/json")
           .header("Authorization", "Bearer " + session.accessToken());

    // When omitted the account's current workspace is used.
    if (workspaceId != null) builder.header("x-workspace-id", workspaceId);

    return builder;
}

/** Unwraps and returns data from the response. Throws ApiError on failure. */
public Object api(String method, String path, Object body, Map<String, Object> params) {
    Resp response = request(method, path, body, params);
    Object payload = null;

    try {
        payload = Json.parse(response.text());
    } catch (RuntimeException ignored) {
        payload = null;
    }

    if (!response.ok()) {
        Map<String, Object> map = Json.asObj(payload);
        String message = Json.str(map, "message", Json.str(map, "error", "unknown error"));
        throw new ApiError(response.status, message, map);
    }

    return Json.get(payload, "data");
}

/** Use the server flag when present, otherwise fall back to the status code. */
public static boolean isTerminal(Map<String, Object> record) {
    Boolean flag = Json.boolOrNull(record, "isTerminal");
    if (flag != null) return flag;

    Integer status = Json.intOrNull(record, "status");
    return status != null && TERMINAL.contains(status);
}

// innorix-client.js
const BASE_URL = (process.env.INNORIX_BASE_URL
  || "https://app.innorix.com").replace(/\/+$/, "");
const TOKEN = process.env.INNORIX_ACCESS_TOKEN;
const WORKSPACE_ID = process.env.INNORIX_WORKSPACE_ID || null;

export const STATUS_COMPLETE = 2;
export const TERMINAL = new Set([2, 4, 5, 9, 99]);       // complete / error / cancelled / partial / failed
export const NOT_SUCCEEDED = new Set([4, 5, 9, 99]);

export async function api(method, path, body = null, params = null) {
  const url = new URL(BASE_URL + path);

  if (params) {
    for (const [key, value] of Object.entries(params)) {
      if (value === undefined || value === null) continue;
      url.searchParams.set(key, String(value));
    }
  }

  const headers = {
    "Content-Type": "application/json",
    Authorization: `Bearer ${TOKEN}`,
  };

  // When omitted the account's current workspace is used.
  if (WORKSPACE_ID) headers["x-workspace-id"] = WORKSPACE_ID;

  const response = await fetch(url, {
    method,
    headers,
    body: body === null ? undefined : JSON.stringify(body),
    signal: AbortSignal.timeout(30_000),
  });

  const payload = await response.json().catch(() => ({}));

  if (!response.ok) {
    throw new Error(payload.message || `HTTP ${response.status}`);
  }

  return payload.data;
}

export function isTerminal(detail) {
  return detail.isTerminal !== undefined
    ? detail.isTerminal
    : TERMINAL.has(detail.status);
}

// InnorixClient.cs
public static readonly string BaseUrl =
    Env("INNORIX_BASE_URL", "https://app.innorix.com").TrimEnd('/');
public static readonly string WorkspaceIdFromEnv = Env("INNORIX_WORKSPACE_ID", null);

public const int StatusComplete = 2;
/// <summary>States the transfer no longer moves out of</summary>
public static readonly HashSet<int> Terminal = new HashSet<int> { 2, 4, 5, 9, 99 };
/// <summary>Terminal states that are not a full success</summary>
public static readonly HashSet<int> NotSucceeded = new HashSet<int> { 4, 5, 9, 99 };

// Applied on every request
request.Headers.TryAddWithoutValidation("Authorization", "Bearer " + Session.AccessToken);
// When omitted the account's current workspace is used.
if (WorkspaceId != null) request.Headers.TryAddWithoutValidation("x-workspace-id", WorkspaceId);

public async Task<JsonNode> ApiAsync(string method, string path, JsonNode body = null,
    IDictionary<string, object> parameters = null)
{
    Resp response = await RequestAsync(method, path, body, parameters).ConfigureAwait(false);
    JsonNode payload = null;

    try
    {
        payload = J.Parse(response.Text());
    }
    catch (Exception)
    {
        payload = null;
    }

    if (!response.Ok)
    {
        JsonObject map = J.AsObj(payload);
        string message = J.Str(map, "message", J.Str(map, "error", "unknown error"));
        throw new ApiError(response.Status, message, map);
    }

    return J.Get(payload, "data");
}

/// <summary>Use the server flag when present, otherwise fall back to the status code.</summary>
public static bool IsTerminal(JsonObject record)
{
    bool? flag = J.BoolOrNull(record, "isTerminal");
    if (flag != null) return flag.Value;

    int? status = J.IntOrNull(record, "status");
    return status != null && Terminal.Contains(status.Value);
}

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

public static String encodePath(String deviceId, String rawPath) {
    String normalized = (rawPath == null ? "" : rawPath).replace("\\", "/");
    return deviceId + "_ino_"
            + Base64.getEncoder().encodeToString(normalized.getBytes(StandardCharsets.UTF_8));
}

public static String nowIso() {
    return Instant.now().truncatedTo(ChronoUnit.SECONDS).toString().replace("Z", ".000Z");
}

export function encodePath(deviceId, rawPath) {
  const normalized = String(rawPath ?? "").replace(/\\/g, "/");
  const token = Buffer.from(normalized, "utf8").toString("base64");
  return `${deviceId}_ino_${token}`;
}

export function nowIso() {
  return new Date().toISOString().replace(/\.\d{3}Z$/, ".000Z");
}

public static string EncodePath(string deviceId, string rawPath)
{
    string normalized = (rawPath ?? "").Replace("\\", "/");
    return deviceId + "_ino_" + Convert.ToBase64String(Encoding.UTF8.GetBytes(normalized));
}

public static string NowIso()
{
    return DateTime.UtcNow.ToString("yyyy-MM-dd'T'HH:mm:ss'.000Z'",
        System.Globalization.CultureInfo.InvariantCulture);
}

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']}"

record Site(String device, String path, String target) {}

static final List<Site> SITES = List.of(
        new Site("branch-seoul", "/report/daily", "/data/branch"),
        new Site("branch-busan", "/report/daily", "/data/branch"),
        new Site("factory-01", "/production/result", "/data/factory"),
        new Site("edge-line-01", "/logs/device", "/data/edge"));

static final String CENTRAL = "device-hq-01";

String siteTarget(Site site) {
    // sites reuse the same file names, so put the site id in the target path
    return site.target() + "/" + site.device();
}

const 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" },
];

const CENTRAL = "device-hq-01";

// sites reuse the same file names, so put the site id in the target path
const siteTarget = (site) => `${site.target}/${site.device}`;

public record Site(string Device, string Path, string Target);

static readonly List<Site> Sites = new()
{
    new Site("branch-seoul", "/report/daily", "/data/branch"),
    new Site("branch-busan", "/report/daily", "/data/branch"),
    new Site("factory-01", "/production/result", "/data/factory"),
    new Site("edge-line-01", "/logs/device", "/data/edge"),
};

const string Central = "device-hq-01";

// sites reuse the same file names, so put the site id in the target path
string SiteTarget(Site site) => 
quot;{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
}

Map<String, Object> buildCollection(Site site, String central,
                                    Map<String, Object> schedule,
                                    Map<String, Object> options) {
    String name = "collect " + site.device();
    boolean sync = schedule == null;

    Map<String, Object> transferOptions = new LinkedHashMap<>(Json.newObj(
            "noSchedule", sync, "target-action", "numbering",
            "send-fileoption", Json.newObj()));
    if (sync) transferOptions.put("syncType", 1);
    if (options != null) transferOptions.putAll(options);

    Map<String, Object> detail = Json.newObj(
            "senderId", site.device(), "receiverId", central,
            "sourceItem", List.of(Json.newObj(
                    "hash", InnorixClient.encodePath(site.device(), site.path()),
                    "filePath", site.path(), "isDir", true)),
            "targetPath", InnorixClient.encodePath(central, siteTarget(site)),
            "step", 1, "transferOptions", transferOptions);

    Map<String, Object> defaultSchedule = Json.newObj(
            "type", "none", "startDateType", "now",
            "startDate", InnorixClient.nowIso(), "timezone", "Asia/Seoul");

    return Json.newObj(
            "name", name, "flowName", name,
            "transferType", sync ? "sync" : "normal",
            "timezone", "Asia/Seoul", "step", 1, "isUpcoming", false,
            "details", List.of(detail),
            "schedules", List.of(schedule != null ? schedule : defaultSchedule));
}

function buildCollection(site, central, schedule = null, options = null) {
  const name = `collect ${site.device}`;
  const sync = schedule === null;

  const transferOptions = {
    noSchedule: sync,
    "target-action": "numbering",
    "send-fileoption": {},
    ...(sync ? { syncType: 1 } : {}),
    ...(options || {}),
  };

  const detail = {
    senderId: site.device,
    receiverId: central,
    sourceItem: [{
      hash: encodePath(site.device, site.path),
      filePath: site.path,
      isDir: true,
    }],
    targetPath: encodePath(central, siteTarget(site)),
    step: 1,
    transferOptions,
  };

  return {
    name,
    flowName: name,
    transferType: sync ? "sync" : "normal",
    timezone: "Asia/Seoul",
    step: 1,
    isUpcoming: false,
    details: [detail],
    schedules: [schedule ?? {
      type: "none", startDateType: "now",
      startDate: nowIso(), timezone: "Asia/Seoul",
    }],
  };
}

JsonObject BuildCollection(Site site, string central,
    JsonObject schedule = null, JsonObject options = null)
{
    string name = 
quot;collect {site.Device}";
    bool sync = schedule == null;

    var transferOptions = new JsonObject
    {
        ["noSchedule"] = sync,
        ["target-action"] = "numbering",
        ["send-fileoption"] = new JsonObject(),
    };
    if (sync) transferOptions["syncType"] = 1;
    if (options != null)
        foreach (var kv in options) transferOptions[kv.Key] = kv.Value?.DeepClone();

    var detail = new JsonObject
    {
        ["senderId"] = site.Device,
        ["receiverId"] = central,
        ["sourceItem"] = new JsonArray
        {
            new JsonObject
            {
                ["hash"] = InnorixClient.EncodePath(site.Device, site.Path),
                ["filePath"] = site.Path,
                ["isDir"] = true,
            },
        },
        ["targetPath"] = InnorixClient.EncodePath(central, SiteTarget(site)),
        ["step"] = 1,
        ["transferOptions"] = transferOptions,
    };

    return new JsonObject
    {
        ["name"] = name,
        ["flowName"] = name,
        ["transferType"] = sync ? "sync" : "normal",
        ["timezone"] = "Asia/Seoul",
        ["step"] = 1,
        ["isUpcoming"] = false,
        ["details"] = new JsonArray { detail },
        ["schedules"] = new JsonArray
        {
            schedule ?? new JsonObject
            {
                ["type"] = "none", ["startDateType"] = "now",
                ["startDate"] = InnorixClient.NowIso(), ["timezone"] = "Asia/Seoul",
            },
        },
    };
}

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.

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"]

Map<String, Object> findAutomation(String 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 (int page = 1; page <= 5; page++) {
        Map<String, Object> result = client.apiObj("GET", "/api/automations", null,
                Json.newObj("page", page, "size", 100, "search", name));

        List<Map<String, Object>> items = new ArrayList<>();
        for (Object flow : Json.arrOf(result, "automations")) {
            for (Object item : Json.arrOf(Json.asObj(flow), "automations")) {
                items.add(Json.asObj(item));
            }
        }

        for (Map<String, Object> item : items) {
            if (name.equals(Json.str(item, "flowName"))) return item;
        }

        if (items.size() < 100) return null;
    }

    return null;
}

async function findAutomation(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 (let page = 1; page <= 5; page += 1) {
    const result = (await client.api("GET", "/api/automations", null,
      { page, size: 100, search: name })) || {};

    const items = (result.automations || [])
      .flatMap((flow) => flow.automations || []);

    for (const item of items) {
      if (item.flowName === name) return item;
    }

    if (items.length < 100) return null;
  }

  return null;
}

async Task<JsonObject> FindAutomationAsync(string 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 (int page = 1; page <= 5; page++)
    {
        JsonObject result = await client.ApiObjAsync("GET", "/api/automations", null,
            new Dictionary<string, object> { ["page"] = page, ["size"] = 100, ["search"] = name });

        var items = J.ArrOf(result, "automations")
            .SelectMany(flow => J.ArrOf(J.AsObj(flow), "automations"))
            .Select(J.AsObj).ToList();

        foreach (JsonObject item in items)
        {
            if (J.Str(item, "flowName") == name) return item;
        }

        if (items.Count < 100) return null;
    }

    return null;
}

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}")

Object[] siteState(String deviceId) {
    Map<String, Object> state = client.apiObj("GET",
            "/api/devices/" + deviceId + "/connectivity");
    return new Object[]{Json.bool(state, "isConnected", false), Json.str(state, "stateLabel")};
}

for (Site site : SITES) {
    Object[] s = siteState(site.device());
    if (!(boolean) s[0]) {
        System.out.printf("%-20s %s%n", site.device(), s[1]);
    }
}

async function siteState(deviceId) {
  const state = (await client.api("GET",
    `/api/devices/${deviceId}/connectivity`)) || {};
  return [Boolean(state.isConnected), state.stateLabel];
}

for (const site of SITES) {
  const [connected, label] = await siteState(site.device);
  if (!connected) {
    console.log(`${site.device.padEnd(20)} ${label}`);
  }
}

async Task<(bool Connected, string Label)> SiteStateAsync(string deviceId)
{
    JsonObject state = await client.ApiObjAsync("GET",
        
quot;/api/devices/{deviceId}/connectivity");
    return (J.Bool(state, "isConnected", false), J.Str(state, "stateLabel"));
}

foreach (Site site in Sites)
{
    var (connected, label) = await SiteStateAsync(site.Device);
    if (!connected)
    {
        Console.WriteLine(
quot;{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"]

Map<String, Long> listFiles(String deviceId, String path) {
    Map<String, Long> found = new LinkedHashMap<>();
    int page = 1;

    while (true) {
        Map<String, Object> result = client.apiObj("GET",
                "/api/devices/" + deviceId + "/files", null,
                Json.newObj("path", path, "page", page, "size", 200, "type", "file"));

        for (Object node : Json.arrOf(result, "items")) {
            Map<String, Object> item = Json.asObj(node);
            found.put(Json.str(item, "name"), Json.longOrNull(item, "size"));
        }

        if (page >= Json.intOr(result, "lastPage", 1)) return found;

        page++;
    }
}

String catchUp(Site site, String central) {
    Map<String, Long> sourceFiles = listFiles(site.device(), site.path());
    Set<String> missing = new TreeSet<>(sourceFiles.keySet());
    missing.removeAll(listFiles(central, siteTarget(site)).keySet());

    if (missing.isEmpty()) return null;

    List<Object> items = new ArrayList<>();
    for (String name : missing) {
        items.add(Json.newObj(
                "path", site.path().replaceAll("/+
quot;, "") + "/" + name,
                "isDir", false, "fileSize", sourceFiles.get(name)));
    }

    Map<String, Object> transfer = client.apiObj("POST", "/api/transfers/manual", Json.newObj(
            "sourceDevice", site.device(),
            "targetDevice", central,
            "targetPath", siteTarget(site),
            "sourceItem", items,
            "sendAllFolder", false,
            "transferOptions", Json.newObj("target-action", "numbering")));

    return Json.str(transfer, "monitorId");
}

async function listFiles(deviceId, path) {
  const found = new Map();
  let page = 1;

  for (;;) {
    const result = (await client.api("GET", `/api/devices/${deviceId}/files`, null, {
      path, page, size: 200, type: "file",
    })) || {};

    for (const item of result.items || []) found.set(item.name, item.size);

    if (page >= (result.lastPage || 1)) return found;

    page += 1;
  }
}

async function catchUp(site, central) {
  const sourceFiles = await listFiles(site.device, site.path);
  const targetFiles = await listFiles(central, siteTarget(site));
  const missing = [...sourceFiles.keys()].filter((n) => !targetFiles.has(n)).sort();

  if (missing.length === 0) return null;

  const transfer = await client.api("POST", "/api/transfers/manual", {
    sourceDevice: site.device,
    targetDevice: central,
    targetPath: siteTarget(site),
    sourceItem: missing.map((name) => ({
      path: `${site.path.replace(/\/+$/, "")}/${name}`,
      isDir: false,
      fileSize: sourceFiles.get(name),
    })),
    sendAllFolder: false,
    transferOptions: { "target-action": "numbering" },
  });

  return transfer.monitorId;
}

async Task<Dictionary<string, long?>> ListFilesAsync(string deviceId, string path)
{
    var found = new Dictionary<string, long?>();
    int page = 1;

    while (true)
    {
        JsonObject result = await client.ApiObjAsync("GET",
            
quot;/api/devices/{deviceId}/files", null,
            new Dictionary<string, object>
            {
                ["path"] = path, ["page"] = page, ["size"] = 200, ["type"] = "file",
            });

        foreach (JsonNode node in J.ArrOf(result, "items"))
        {
            JsonObject item = J.AsObj(node);
            found[J.Str(item, "name")] = J.LongOrNull(item, "size");
        }

        if (page >= J.Int(result, "lastPage", 1)) return found;

        page++;
    }
}

async Task<string> CatchUpAsync(Site site, string central)
{
    var sourceFiles = await ListFilesAsync(site.Device, site.Path);
    var targetFiles = await ListFilesAsync(central, SiteTarget(site));

    var missing = sourceFiles.Keys.Where(n => !targetFiles.ContainsKey(n))
        .OrderBy(n => n).ToList();

    if (missing.Count == 0) return null;

    var items = new JsonArray();
    foreach (string name in missing)
    {
        items.Add(new JsonObject
        {
            ["path"] = 
quot;{site.Path.TrimEnd('/')}/{name}",
            ["isDir"] = false,
            ["fileSize"] = sourceFiles[name],
        });
    }

    JsonObject transfer = await client.ApiObjAsync("POST", "/api/transfers/manual",
        new JsonObject
        {
            ["sourceDevice"] = site.Device,
            ["targetDevice"] = central,
            ["targetPath"] = SiteTarget(site),
            ["sourceItem"] = items,
            ["sendAllFolder"] = false,
            ["transferOptions"] = new JsonObject { ["target-action"] = "numbering" },
        });

    return J.Str(transfer, "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

public List<Map<String, Object>> paginate(String path, Map<String, Object> params,
                                          int limit, int maxPages) {
    Map<String, Object> query = new LinkedHashMap<>(params == null ? Map.of() : params);
    query.put("limit", limit);

    List<Map<String, Object>> rows = new ArrayList<>();
    String cursor = null;

    for (int page = 0; page < maxPages; page++) {
        if (cursor != null) query.put("cursor", cursor);

        Map<String, Object> result = apiObj("GET", path, null, query);

        for (Object record : Json.arrOf(result, "data")) rows.add(Json.asObj(record));

        Map<String, Object> pagination = Json.objOf(result, "pagination");
        if (!Json.bool(pagination, "hasMore", false)) return rows;

        cursor = Json.str(pagination, "nextCursor");
        if (cursor == null) return rows;
    }

    return rows;
}

export async function* paginate(path, params = {}, limit = 200, maxPages = 50) {
  const query = { ...params, limit };
  let cursor = null;

  for (let page = 0; page < maxPages; page += 1) {
    if (cursor) query.cursor = cursor;

    const result = (await api("GET", path, null, query)) || {};

    for (const record of result.data || []) yield record;

    const pagination = result.pagination || {};

    if (!pagination.hasMore || !pagination.nextCursor) return;

    cursor = pagination.nextCursor;
  }
}

public async Task<List<JsonObject>> PaginateAsync(string path,
    IDictionary<string, object> parameters = null, int limit = 200, int maxPages = 50)
{
    var query = new Dictionary<string, object>(
        parameters ?? new Dictionary<string, object>()) { ["limit"] = limit };

    var rows = new List<JsonObject>();
    string cursor = null;

    for (int page = 0; page < maxPages; page++)
    {
        if (cursor != null) query["cursor"] = cursor;

        JsonObject result = await ApiObjAsync("GET", path, null, query).ConfigureAwait(false);

        foreach (JsonNode record in J.ArrOf(result, "data")) rows.Add(J.AsObj(record));

        JsonObject pagination = J.ObjOf(result, "pagination");
        if (!J.Bool(pagination, "hasMore", false)) return rows;

        cursor = J.Str(pagination, "nextCursor");
        if (cursor == null) return rows;
    }

    return rows;
}

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}")

Map<String, Integer> siteSummary(String deviceId, int days) {
    Instant end = Instant.now();
    DateTimeFormatter fmt = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss'Z'")
            .withZone(ZoneOffset.UTC);

    List<Map<String, Object>> rows = client.paginate(
            "/api/devices/" + deviceId + "/transfer-history",
            Json.newObj("startDate", fmt.format(end.minus(days, ChronoUnit.DAYS)),
                        "endDate", fmt.format(end)), 200, 50);

    int failed = 0;
    for (Map<String, Object> row : rows) {
        Integer status = Json.intOrNull(row, "status");
        if (status != null && InnorixClient.NOT_SUCCEEDED.contains(status)) failed++;
    }

    return Json.newObj("total", rows.size(), "failed", failed);
}

System.out.printf("%-20s %^6s %6s %6s%n", "site", "up", "collect", "fail");
for (Site site : SITES) {
    Object[] state = siteState(site.device());
    Map<String, Integer> summary = siteSummary(site.device(), 1);
    System.out.printf("%-20s %^6s %6d %6d%n", site.device(),
            (boolean) state[0] ? "O" : "X",
            summary.get("total"), summary.get("failed"));
}

async function siteSummary(deviceId, days = 1) {
  const end = new Date();
  const fmt = (d) => d.toISOString().replace(/\.\d{3}Z$/, "Z");

  const rows = [];
  for await (const row of paginate(`/api/devices/${deviceId}/transfer-history`, {
    startDate: fmt(new Date(end.getTime() - days * 86400000)),
    endDate: fmt(end),
  })) rows.push(row);

  const failed = rows.filter((r) => NOT_SUCCEEDED.has(r.status)).length;
  return { total: rows.length, failed };
}

console.log("site".padEnd(20), "up".padStart(6), "collect".padStart(6), "fail".padStart(6));
for (const site of SITES) {
  const [connected] = await siteState(site.device);
  const summary = await siteSummary(site.device);
  console.log(site.device.padEnd(20),
    (connected ? "O" : "X").padStart(6),
    String(summary.total).padStart(6),
    String(summary.failed).padStart(6));
}

async Task<(int Total, int Failed)> SiteSummaryAsync(string deviceId, int days = 1)
{
    DateTime end = DateTime.UtcNow;
    const string Fmt = "yyyy-MM-dd'T'HH:mm:ss'Z'";

    List<JsonObject> rows = await client.PaginateAsync(
        
quot;/api/devices/{deviceId}/transfer-history",
        new Dictionary<string, object>
        {
            ["startDate"] = end.AddDays(-days).ToString(Fmt),
            ["endDate"] = end.ToString(Fmt),
        });

    int failed = rows.Count(r =>
    {
        int? status = J.IntOrNull(r, "status");
        return status != null && InnorixClient.NotSucceeded.Contains(status.Value);
    });

    return (rows.Count, failed);
}

Console.WriteLine(
quot;{"site",-20} {"up",6} {"collect",6} {"fail",6}");
foreach (Site site in Sites)
{
    var (connected, _) = await SiteStateAsync(site.Device);
    var summary = await SiteSummaryAsync(site.Device);
    Console.WriteLine(
quot;{site.Device,-20} {(connected ? "O" : "X"),6} "
        + 
quot;{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)

public Map<String, Object> wait(String monitorId, int timeoutSeconds, int intervalSeconds) {
    long deadline = System.currentTimeMillis() + timeoutSeconds * 1000L;

    while (System.currentTimeMillis() < deadline) {
        Map<String, Object> detail = apiObj("GET", "/api/transfers/" + monitorId);

        if (isTerminal(detail)) return detail;

        sleep(intervalSeconds * 1000L);
    }

    throw new ApiError(0, "timeout waiting for " + monitorId);
}

public List<Map<String, Object>> failedFiles(String monitorId) {
    List<Map<String, Object>> failed = new ArrayList<>();

    for (Map<String, Object> row : transferFiles(monitorId)) {
        Integer status = Json.intOrNull(row, "status");
        if (status != null && NOT_SUCCEEDED.contains(status)) failed.add(row);
    }

    return failed;
}

public int retryFailed(String monitorId) {
    List<Map<String, Object>> rows = failedFiles(monitorId);
    if (rows.isEmpty()) return 0;

    List<Object> filesRetry = new ArrayList<>();

    for (Map<String, Object> row : rows) {
        String path = Json.str(row, "sourceFilePath");
        if (path == null) continue;
        filesRetry.add(Json.newObj("filePath", path, "isDir", Json.bool(row, "isFolder", false)));
    }

    api("POST", "/api/transfers/" + monitorId + "/retry", Json.newObj("filesRetry", filesRetry));

    return rows.size();
}

const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));

export async function wait(monitorId, { timeout = 3600, interval = 3 } = {}) {
  const deadline = Date.now() + timeout * 1000;

  while (Date.now() < deadline) {
    const detail = await api("GET", `/api/transfers/${monitorId}`);

    if (isTerminal(detail)) return detail;

    await sleep(interval * 1000);
  }

  throw new Error(`timeout waiting for ${monitorId}`);
}

export async function failedFiles(monitorId) {
  const result = (await api("GET", `/api/transfers/${monitorId}/files`, null, {
    state: "any", size: 500,
  })) || {};

  return (result.children || []).filter((row) => NOT_SUCCEEDED.has(row.status));
}

export async function retryFailed(monitorId) {
  const rows = await failedFiles(monitorId);

  if (rows.length === 0) return 0;

  await api("POST", `/api/transfers/${monitorId}/retry`, {
    filesRetry: rows
      .filter((row) => row.sourceFilePath)
      .map((row) => ({ filePath: row.sourceFilePath, isDir: Boolean(row.isFolder) })),
  });

  return rows.length;
}

public async Task<JsonObject> WaitAsync(string monitorId, int timeoutSeconds = 3600,
    int intervalSeconds = 3)
{
    long deadline = Environment.TickCount64 + timeoutSeconds * 1000L;

    while (Environment.TickCount64 < deadline)
    {
        JsonObject detail = await ApiObjAsync("GET", "/api/transfers/" + monitorId)
            .ConfigureAwait(false);

        if (IsTerminal(detail)) return detail;

        await Task.Delay(intervalSeconds * 1000).ConfigureAwait(false);
    }

    throw new ApiError(0, "timeout waiting for " + monitorId);
}

public async Task<List<JsonObject>> FailedFilesAsync(string monitorId)
{
    List<JsonObject> rows = await TransferFilesAsync(monitorId).ConfigureAwait(false);

    return rows.Where(row =>
    {
        int? status = J.IntOrNull(row, "status");
        return status != null && NotSucceeded.Contains(status.Value);
    }).ToList();
}

/// <summary>Only callable once the transfer has reached a terminal state.</summary>
public async Task<int> RetryFailedAsync(string monitorId)
{
    List<JsonObject> rows = await FailedFilesAsync(monitorId).ConfigureAwait(false);
    if (rows.Count == 0) return 0;

    var filesRetry = new JsonArray();

    foreach (JsonObject row in rows)
    {
        string path = J.Str(row, "sourceFilePath");
        if (path == null) continue;

        filesRetry.Add(new JsonObject
        {
            ["filePath"] = path,
            ["isDir"] = J.Bool(row, "isFolder", false),
        });
    }

    await ApiAsync("POST", "/api/transfers/" + monitorId + "/retry",
        new JsonObject { ["filesRetry"] = filesRetry }).ConfigureAwait(false);

    return rows.Count;
}

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

for (var entry : collectors.entrySet()) {
    List<Map<String, Object>> runs = client.executions(entry.getValue());

    if (runs.isEmpty()) {
        System.out.println(entry.getKey()
                + ": no run history - check registration and start condition");
        continue;
    }

    Map<String, Object> latest = runs.get(0);

    if (Json.intOr(latest, "status", -1) != InnorixClient.STATUS_COMPLETE) {
        int count = client.retryFailed(Json.str(latest, "monitorId"));
        System.out.println(entry.getKey() + ": retried " + count + " files");
    }
}

for (const [device, automationId] of Object.entries(collectors)) {
  const runs = (await client.executions(automationId)) || [];

  if (runs.length === 0) {
    console.log(`${device}: no run history - check registration and start condition`);
    continue;
  }

  const latest = runs[0];

  if (latest.status !== STATUS_COMPLETE) {
    console.log(`${device}: retried ${await retryFailed(latest.monitorId)} files`);
  }
}

foreach (var (device, automationId) in collectors)
{
    List<JsonObject> runs = J.AsList(await client.ExecutionsAsync(automationId));

    if (runs.Count == 0)
    {
        Console.WriteLine(
quot;{device}: no run history - check registration and start condition");
        continue;
    }

    JsonObject latest = runs[0];

    if (J.IntOrNull(latest, "status") != InnorixClient.StatusComplete)
    {
        int count = await client.RetryFailedAsync(J.Str(latest, "monitorId"));
        Console.WriteLine(
quot;{device}: retried {count} 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	`isConnected`text `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

Centrally Collecting Files from Branches, Factories, and Edge Devices

Getting Started#

Core Concept#

Collection Flow#

Operational Benefits#

IT Engineer#

Collection Environment#

Collection Paths#

Collection Conditions#

Collection Automation#

Collection Status#

Operational Response#

Developer#

Integration Preparation#

Configure Site List#

Register Collection Automation#

Preventing Duplicate Registration#

Offline Site Response#

Aggregate Collection Status#

Exception Handling#

Centrally Collecting Files from Branches, Factories, and Edge Devices

Getting Started#

Core Concept#

Collection Flow#

Operational Benefits#

IT Engineer#

Collection Environment#

Collection Paths#

Collection Conditions#

Collection Automation#

Collection Status#

Operational Response#

Developer#

Integration Preparation#

Configure Site List#

Register Collection Automation#

Preventing Duplicate Registration#

Offline Site Response#

Aggregate Collection Status#

Exception Handling#