向 Data Bundle 上传数据

本指南说明如何通过 RealityConnect API 将一次采集的原始输入文件上传到 Data Bundle。

所有上传端点都需要 write:bundle 作用域；列出会话需要 read:bundle。

上传流程

Data Bundle 保存单次采集的原始输入：点云、轨迹、图像、每张图像的元数据等。Bundle 接受的确切文件集取决于该 Bundle 创建时所针对的设备。

文件在上传会话中上传。会话是服务器为一个 Bundle 的一批文件创建的容器。当你打开它时，你声明该批次将包含多少个文件（expectedFileCount）；一旦该数量的文件完成，会话就会自动完成。

每个单独的文件通过 S3 分段上传：

在 Bundle 上打开一个会话，声明你将发送的文件总数。
对每个文件：发起上传，将其各分段直接 PUT 到返回的 URL，然后完成它。
当完成的文件数量达到 expectedFileCount 时，会话被标记为 completed。

sequenceDiagram
  participant Client as Client
  participant RCAPI as RCAPI
  participant S3 as S3

  Client->>RCAPI: POST upload-sessions
  RCAPI-->>Client: sessionId + requirements

  loop for each file
    Client->>RCAPI: POST initiate file
    RCAPI-->>Client: fileId + presigned part URLs
    loop for each part
      Client->>S3: PUT part
      S3-->>Client: ETag
    end
    Client->>RCAPI: POST finalize
    RCAPI-->>Client: file completed + sessionStatus
  end

上传 requirements（在你打开会话时返回）会准确告诉你 Bundle 接受哪些文件类型、哪些是强制的，以及它们之间如何相互依赖。

完整示例：ZF SLAM 采集

ZF SLAM Bundle 是一个很好的示例，因为它接受四种不同的文件类型且存在跨类型依赖。一次完整的 ZF 采集由以下部分组成：

文件	类型	备注
`scan.las`	点云	恰好一个
`…laser_0.txt`	轨迹	恰好一个
`panorama/*.jpg`	全景图像	多个，每个都与元数据配对
`panorama/*.json`	每张图像的元数据	多个，每张图像一个

因此，一个完整的 ZF 批次是 1 个点云 + 1 条轨迹 + N 张图像 + N 个元数据文件。我们将在单个会话中上传全部文件。

端点

所有路径都相对于你的区域 api_url。Bundle 通过其自身的 id（bundleId）来寻址。

方法	路径	作用域	用途
`POST`	`/v1/bundles/{bundleId}/upload-sessions`	`write:bundle`	打开会话
`GET`	`/v1/bundles/{bundleId}/upload-sessions`	`read:bundle`	列出会话（恢复 / 进度）
`POST`	`/v1/bundles/{bundleId}/upload-sessions/{sessionId}/files`	`write:bundle`	发起文件上传
`GET`	`/v1/bundles/{bundleId}/upload-sessions/{sessionId}/files/{fileId}/refresh/{startIndex}`	`write:bundle`	重新签发过期的分段 URL
`POST`	`/v1/bundles/{bundleId}/upload-sessions/{sessionId}/files/{fileId}/finalize`	`write:bundle`	完成文件上传

开始之前

要向 Data Bundle 上传数据，你需要：

已完成的 OAuth 流程，持有 access_token 和你的区域 api_url
一个现有的 Data Bundle（bundleId）

如果你还没有 Bundle，请先按照创建 Data Bundle创建一个。该流程会获取设备输入树、选择叶子 dbuPath，并在父节点下创建 Bundle。

步骤 1：打开上传会话

声明该批次将包含的文件总数。对于包含 13 张图像的完整 ZF 采集，即 1 + 1 + 13 + 13 = 28。

POST {api_url}/v1/bundles/{bundleId}/upload-sessions
Authorization: Bearer {access_token}
Content-Type: application/json

{ "expectedFileCount": 28 }

响应返回 sessionId 和 Bundle 的 requirements：

{
  "sessionId": "6ebf0fe5-4428-4b26-a7e4-e3274980123b",
  "expectedFileCount": 28,
  "requirements": [
    { "type": "pointclouds",      "allowedExtensions": ["las", "laz"],         "optional": false, "allowMultiples": false, "dependencies": [] },
    { "type": "trajectories",     "allowedExtensions": ["txt"],                "optional": false, "allowMultiples": false, "dependencies": [] },
    { "type": "pictures",         "allowedExtensions": ["jpg", "png", "jpeg"], "optional": true,  "allowMultiples": true,  "dependencies": ["picturesMetadata"] },
    { "type": "picturesMetadata", "allowedExtensions": ["json"],               "optional": true,  "allowMultiples": true,  "dependencies": ["pictures"] }
  ]
}

理解 `requirements` 和跨依赖

requirements 中的每个条目描述 Bundle 接受的一种文件类型：

字段	含义
`type`	发起文件上传时你作为 `type` 传入的标识符。
`allowedExtensions`	此类型接受的文件扩展名。
`optional`	如果为 `false`，则 Bundle 在没有至少一个此类型文件的情况下无法处理。
`allowMultiples`	如果为 `false`，则此类型只能上传一个文件。
`dependencies`	每当此类型存在时也必须存在的其他类型。

对于 ZF 示例：

pointclouds 和 trajectories 是强制的（optional: false）且单一的（allowMultiples: false）。每个 ZF Bundle 都恰好需要各一个。
pictures 和 picturesMetadata 是可选的，接受多个文件。
这两种图像类型相互依赖：pictures.dependencies = ["picturesMetadata"] 且 picturesMetadata.dependencies = ["pictures"]。

依赖意味着：如果某个类型存在，则它列出的每个类型也必须存在。 由于 ZF 图像类型相互引用，它们作为一对是全有或全无的。你可以将图像及其元数据一起上传，或两者都跳过，但不能只上传其中一个而不上传另一个。

有效的 ZF 批次：

最小 — 1 个点云 + 1 条轨迹（无图像）。expectedFileCount = 2。
完整 — 1 个点云 + 1 条轨迹 + N 张图像 + N 个 picturesMetadata。expectedFileCount = 2 + 2N。

无效：

图像没有其元数据（或反之），因为跨依赖未得到满足。
没有点云或没有轨迹的批次，因为缺少强制类型。

在打开会话之前，请统计你计划上传的每个文件，并将该总数设置为 expectedFileCount。可选类型仅在你将其包含在批次中时才计入。会话在完成该数量的文件时结束，因此你声明的数字必须与你实际上传的相符。

步骤 2：上传每个文件

对批次中的每个文件重复以下三个调用。

2a. 发起

传入文件名、其字节大小及其 type（requirements 类型之一）：

POST {api_url}/v1/bundles/{bundleId}/upload-sessions/{sessionId}/files
Authorization: Bearer {access_token}
Content-Type: application/json

{ "fileName": "scan.las", "fileSize": 734003200, "type": "pointclouds" }

响应描述了分段上传：要发送多少个 parts，以及每个分段一个预签名 url。

{
  "fileId": "c8c8cb0b-f476-4211-a3a4-72e4e04b91d5",
  "parts": 8,
  "urls": [
    { "partNumber": 1, "url": "https://s3…/part-1?…" },
    { "partNumber": 2, "url": "https://s3…/part-2?…" }
  ]
}

2b. 上传各分段

将文件分割为 parts 个每段 ceil(fileSize / parts) 字节的连续块，并将每个块 PUT 到其预签名 url。这些请求不发送 auth 标头；URL 已经过签名。保留每个分段的 ETag 响应标头，你在完成时需要它。

import math, requests

def upload_parts(file_path, file_size, urls):
    part_size = math.ceil(file_size / len(urls))
    parts = []
    with open(file_path, "rb") as handle:
        for entry in sorted(urls, key=lambda u: u["partNumber"]):
            res = requests.put(entry["url"], data=handle.read(part_size))
            res.raise_for_status()
            parts.append({"partNumber": entry["partNumber"], "etag": res.headers["ETag"]})
    return parts

将 ETag 完全按返回原样传回，包括其两侧的引号。

2c. 完成

发送收集到的分段以完成文件。该端点返回 200 OK。

POST {api_url}/v1/bundles/{bundleId}/upload-sessions/{sessionId}/files/{fileId}/finalize
Authorization: Bearer {access_token}
Content-Type: application/json

{ "parts": [ { "partNumber": 1, "etag": "\"a1b2…\"" }, { "partNumber": 2, "etag": "\"c3d4…\"" } ] }

响应回显文件并报告会话状态：

{
  "fileId": "c8c8cb0b-f476-4211-a3a4-72e4e04b91d5",
  "fileName": "scan.las",
  "type": "pointclouds",
  "size": 734003200,
  "status": "completed",
  "sessionStatus": "pending"
}

sessionStatus 保持 pending，直到最后一个预期文件完成，届时它变为 completed。

刷新过期的 URL

预签名 URL 会过期。如果上传耗时较长且 PUT 开始返回 403，请从你停止的那个分段重新签发 URL 并继续：

GET {api_url}/v1/bundles/{bundleId}/upload-sessions/{sessionId}/files/{fileId}/refresh/0
Authorization: Bearer {access_token}

{ "startIndex": 0, "urls": [ { "partNumber": 1, "url": "https://s3…/part-1?…" } ] }

文件无需重新发起；仅更新分段 URL。

步骤 3：跟踪进度并恢复

列出 Bundle 的会话，以检查进度或恢复中断的批次，而无需将 sessionId 保存在内存中。该列表支持分页和排序，并可选地按 status 过滤。

查询参数：page（默认 1）、limit（默认 20，最大 20）、sortBy（createdAt 或 status，默认 createdAt）、sortDir（ASC 或 DESC，默认 ASC）、status（可选过滤：pending 或 completed）。

GET {api_url}/v1/bundles/{bundleId}/upload-sessions?page=1&limit=20&sortBy=createdAt&sortDir=DESC
Authorization: Bearer {access_token}

{
  "items": [
    {
      "sessionId": "6ebf0fe5-4428-4b26-a7e4-e3274980123b",
      "status": "pending",
      "expectedFileCount": 28,
      "finalizedFileCount": 14
    }
  ],
  "total": 1
}

finalizedFileCount 与 expectedFileCount 的对比会告诉你还剩多少文件。要恢复，只需发起并完成尚未完成的文件。

验证与限制

请仔细阅读，因为它决定了你在使用 RealityConnect API 上传数据时需要承担的责任。

上传端点会将每个文件记录在你提供的 type 下，并纯粹根据数量完成会话：一旦 finalizedFileCount 等于 expectedFileCount，会话即为 completed。在上传时，服务器不会验证 type 是否为 requirements 之一、文件扩展名是否被允许、强制类型是否存在，或跨依赖是否被满足。
所有规则（强制类型、allowMultiples 以及类型之间的 dependencies）都在 Bundle 被送去处理之后强制执行。忽略 requirements 的批次会成功上传但无法处理。
上传 API 从不执行深层内容验证。 .las 是否为真正的 ZF 点云，或 .json 是否正确引用其图像，仅在处理时检查。

将 requirements 视为契约。在你的上传批次中遵守类型、扩展名、可选性、多重性和依赖，以便 Bundle 在上传后可被处理。由于上传结构不正确而导致的处理失败不予退款。

完整示例

将其组合起来用于 ZF 采集。files 是你从本地数据集构建的列表，每个条目将一个路径与其来自 requirements 的 type 配对。

import math, requests

API_URL = "https://<your-regional-api-url>"
TOKEN = "<access_token>"
BUNDLE_ID = "<bundleId>"

HEADERS = {"Authorization": f"Bearer {TOKEN}"}

# (local path, schema type) for every file in the batch
files = [
    ("scan.las", "pointclouds"),
    ("2024-07-scan.laser_laser_0.txt", "trajectories"),
    ("panorama/2024-07-scan-idx40.jpg", "pictures"),
    ("panorama/2024-07-scan-idx40.json", "picturesMetadata"),
    # … remaining image / metadata pairs …
]


def upload_one(session_id, path, file_type):
    size = __import__("os").path.getsize(path)
    base = f"{API_URL}/v1/bundles/{BUNDLE_ID}/upload-sessions/{session_id}/files"

    initiated = requests.post(
        base, headers=HEADERS,
        json={"fileName": path.split("/")[-1], "fileSize": size, "type": file_type},
    ).json()

    part_size = math.ceil(size / initiated["parts"])
    parts = []
    with open(path, "rb") as handle:
        for entry in sorted(initiated["urls"], key=lambda u: u["partNumber"]):
            res = requests.put(entry["url"], data=handle.read(part_size))
            res.raise_for_status()
            parts.append({"partNumber": entry["partNumber"], "etag": res.headers["ETag"]})

    return requests.post(
        f"{base}/{initiated['fileId']}/finalize", headers=HEADERS, json={"parts": parts},
    ).json()


# 1. Open the session for the exact number of files we will send.
session = requests.post(
    f"{API_URL}/v1/bundles/{BUNDLE_ID}/upload-sessions",
    headers=HEADERS, json={"expectedFileCount": len(files)},
).json()

# 2. Upload every file under its type.
result = None
for path, file_type in files:
    result = upload_one(session["sessionId"], path, file_type)

# 3. The last finalize reports the session as completed.
assert result["sessionStatus"] == "completed"

下一步

一旦每个文件都已上传且会话为 completed，请继续前往处理与监控，将原始输入转化为可查看的输出。