Uploading data¶
Collections are where your project's data lives. They organize simulation files, keep them under a known structure, and provide the foundation for everything else in Nexus. Before you can transform, query, or train on data, it first has to exist in a collection.
This section shows how to bring data into Nexus and inspect what was uploaded. Attach labels and splits in Describing your data when you are ready to select work for recipes.
You create a collection, upload data, and explore its contents.
Describe what your collection should contain¶
A collection schema is the admissibility contract for folder layouts and file names. Before bytes move, Ingestion runs a plan gate: collection.plan(...) compares your local tree to the schema and reports blocking mismatches (plan.gate_status, plan.is_admissible, plan.print_validation_report(...)). Fix the data or adjust the schema while the plan is still a dry run.
case_grouping= is optional. Upload without it when you just need the files in the collection; pass it when a recipe must process each simulation as a unit, because deploy_recipe(..., cases=…) binds cases that were already admitted and will not admit them for you. When you do pass case_grouping= to plan, a second gate validates how paths form cases (roles, composition, profiles) — after the schema gate, and only if layout checks pass. See Case grouping. plan.is_admissible spans every gate that ran (fail-closed for unknown statuses). Per-gate detail lives under plan.validation_gates; the admitted cases are plan.cases (upload.cases after upload), which is what you pass to deploy_recipe(..., cases=…).
With ignore_extras=True on the collection schema, local ingest drops undeclared paths before the plan gate and reports them on plan.ignored_extras. On CaseStructure.compose, pass ignore_extras=True to structure.schema(...) so nested ResourceMap paths count as declared (the default compose projection lists top-level roots only).
Permissive schema¶
Schema.permissive() accepts every path — no layout checks. Use it only when you intentionally skip admissibility; prefer an explicit schema when you control the tree yourself. When grouping still needs a shared resource vocabulary, pass it with resources= (resource_catalog= remains accepted):
from miura.sdk import Schema
schema = Schema.permissive()
# schema = Schema.permissive(resources=resources) # optional shared ResourceMap
collection = project.create_collection(
collection_name="scratch",
schema=schema,
)
Structured schema¶
For folder-shaped cases — one repeated folder is one case, with named files as
members — author layout and case grouping together with
CaseStructure.folder, then project
.schema(count=...) and .case_grouping(). You need more than this only when a
single case draws its files from separate directory trees, when the name a file
has on disk differs from the part it plays in the case, when several collections
must share one set of names, or when a case member is a whole directory rather
than a file — see multi-root cases and
case grouping.
from miura.sdk import CaseStructure, File, Number, capture, one_or_more
case_id = capture("case_id", Number(min_digits=6, max_digits=6))
structure = CaseStructure.folder(
path="sim_" + case_id,
members={
"mesh": File("mesh.stl"),
"result_field": File("field.xdmf"),
},
)
schema = structure.schema(count=one_or_more())
case_grouping = structure.case_grouping()
schema(count=...) is how many such folders the collection may contain.
Pass both projections into create/plan as usual.
When the layout is a deeper nested tree (for example a full OpenFOAM case),
declare the folder tree on Schema directly:
from miura.sdk import File, Folder, Number, Schema, one_or_more
case_id = Number(min_digits=6, max_digits=6)
openfoam_schema = Schema(
roots=[
Folder(
"case_" + case_id,
count=one_or_more(),
children=[
Folder(
"constant",
children=[
Folder(
"polyMesh",
children=[
File("points"),
File("faces"),
File("owner"),
File("neighbour"),
File("boundary"),
],
),
File("transportProperties"),
],
),
Folder(
"system",
children=[
File("controlDict"),
File("fvSchemes"),
File("fvSolution"),
],
),
Folder(
"0",
children=[
File("U"),
File("p"),
],
),
],
)
]
)
Folder(..., count=optional()) does not change child count= values. An absent
optional parent does not require descendants; a present optional parent still
requires default children. See
Optional folders and child count.
A serial OpenFOAM case tree has one case_XXXXXX/ folder per run:
case_000001/
├── constant/
│ ├── polyMesh/ # mesh topology (points, faces, owner, neighbour, boundary)
│ └── transportProperties
├── system/ # solver settings
│ ├── controlDict
│ ├── fvSchemes
│ └── fvSolution
└── 0/ # initial / boundary fields (time directory)
├── U
└── p
Pass the schema to collection creation:
collection = project.create_collection(
collection_name="openfoam-campaign",
schema=openfoam_schema,
)
Preview the gate before upload:
plan = collection.plan(LocalDataSource(data_path), mode=UploadMode.REPLACE)
if not plan.is_admissible:
plan.print_validation_report(max_issues=20)
Plan validation report (failed)
──────────────────────────────────────────────────
Plan ID: igplan_demo igplan_demo
Failed (1):
• [SCHEMA_PATH_NOT_PERMITTED] /case_000001/postProcessing/forces/0/force.dat: path not allowed by collection schema
Passed (3):
• /case_000001/constant/transportProperties
• /case_000001/system/controlDict
• /case_000001/0/U
Reuse schemas.py (openfoam_case_schema()).
Bring data into Nexus¶
Ingestion copies files from a datasource into the Collection. You can upload directly or plan first to inspect gate status and work summaries.
Datasources¶
A datasource is the origin of files for upload. Whichever you use, the calls are the same — collection.plan(datasource) to validate, then plan.apply() to import. Only the datasource changes.
Local datasource¶
LocalDataSource points at a directory on disk. The client walks that tree, checks it against the collection schema, then copies admitted paths into Nexus.
from miura.sdk import LocalDataSource
datasource = LocalDataSource("/my-simulations/openfoam-quickstart")
Connector datasource¶
ConnectorDataSource points at a remote bucket that an administrator has registered as a connector. Nexus inventories the remote prefix server-side, seals the resulting file list, and validates it against the schema before anything is copied. Credentials stay with the platform: the client never sees or sends them.
from miura.sdk import ConnectorDataSource, ConnectorInventoryScope
datasource = ConnectorDataSource(
connector_id="conn_5329aaa438a94473",
inventory_scope=ConnectorInventoryScope(source_prefix="s3://customer-bucket/for_dev"),
transfer_binding_id=transfer_binding_id,
)
plan = await collection.plan(datasource, mode=UploadMode.INSERT)
await plan.print_validation_report() # review before any bytes move
result = await plan.apply()
Connector inventory and transfer polling are async-only, so this path needs AsyncNexus; sync collection.plan() raises TypeError for a ConnectorDataSource.
transfer_binding_id is a platform routing value for your deployment — ask your administrator. Because it travels on the datasource, apply() is identical to the local case. Nexus picks the transfer DAG itself.
Connector plans differ from local ones in a few ways that follow from the file list being sealed remotely: max_files= is rejected, and properties that only describe a local tree (data_root, ignored_extras, work_summary) raise AttributeError rather than report a misleading zero. case_grouping= is supported on connector plans by default (same API as local); opt out with MIURA_CONNECTOR_CASE_GROUPING=false on incompatible stacks. Import files from a remote connector covers the whole flow — finding and testing a connector, controlling what the inventory scan picks up, reading the validation report, importing, and replaying a sealed manifest — with runnable scripts.
Upload modes (local ingest)¶
mode= on collection.plan() and collection.upload() controls how local paths interact with items already in the collection (same UploadMode enum recipes use for target writes — see Target upload mode).
| Mode | Behaviour | Typical use |
|---|---|---|
INSERT |
New paths only; existing skipped | Incremental adds |
UPDATE |
Existing paths only | Refresh changed files |
UPSERT |
Insert or update per path | Safe re-runs of a full tree |
REPLACE |
Replace the logical group for this upload scope | Full snapshot refresh |
APPEND is a legacy alias for INSERT when calling collection.upload() without an explicit mode. Runnable walkthrough of all four ingestion modes: upload.py.
After planning, plan.fetch_validation_report() returns a PlanValidationReport. Gate failures are in report.failed_records; ingest candidates are in report.passed_records.
With sync Nexus(), plan.apply() blocks until the transfer finishes (no await). The async API uses the same contract with await plan.apply().
Preview then apply a full refresh:
from miura.display import print_ingestion_plan
from miura.sdk import LocalDataSource, UploadMode
datasource = LocalDataSource("/my-simulations/openfoam-quickstart")
plan = collection.plan(datasource, mode=UploadMode.REPLACE)
print_ingestion_plan(plan)
if plan.is_admissible and plan.requires_execution:
plan.apply()
Ingestion plan
──────────────────────────────────────────────────
Plan ID: igplan_demo123456 igplan_demo123456
Mode: REPLACE
Gate: passed
State: SUCCEEDED
Admissible: True
Candidates: 3 Size: 6.0 KiB
Transfer: 3 to upload, 0 skipped existing
Mapping: 0 write, 0 replace required
Apply: 3 new files to upload
Data root: /my-simulations/openfoam-quickstart
List ingest candidates from passed_records:
report = plan.fetch_validation_report()
if report.failed_records:
plan.print_validation_report(max_issues=20)
for issue in report.passed_records:
print(issue.path)
Plan validation report (failed)
──────────────────────────────────────────────────
Plan ID: igplan_demo igplan_demo
Failed (1):
• [SCHEMA_PATH_NOT_PERMITTED] /case_000001/postProcessing/forces/0/force.dat: path not allowed by collection schema
Passed (3):
• /case_000001/constant/transportProperties
• /case_000001/system/controlDict
• /case_000001/0/U
/case_000001/constant/transportProperties
/case_000001/system/controlDict
/case_000001/0/U
Direct upload:
from miura.display import print_upload_result
result = collection.upload(datasource, mode=UploadMode.UPSERT)
print_upload_result(result)
Upload result
──────────────────────────────────────────────────
Status: success
Session: ups_demo… ups_demo1234567890abcdef
Run: igrun_demo… igrun_demo1234567890abcdef
Plan: igplan_demo… igplan_demo1234567890abcdef
Uploaded: 3 Failed: 0 Size: 6.0 KiB
Explore what was uploaded¶
Each uploaded path is an item (file or folder) with a stable URI.
Browse at one level (immediate children of a path, like Path.iterdir()):
list_items(path)— returns alist(use when the folder is small)iter_items(path)— streams pages from the API (use when the folder is large)
Browse recursively under a path (like os.walk()):
walk_items(path)— yields(item, depth)for every descendant
To print every file in the collection:
for item, depth in collection.walk_items(path="/", include_dirs=False):
print(item.item_uri, item.file_size)
Look up one path or move through the tree without re-listing:
collection.get_item(path).children,.siblings,.parent(sync API; async items useawait item.get_parent())