Examples

The MecaPy repo ships pedagogical examples under repos/functions/. The snippets below mirror those — example01 (mode A) and example02 (mode B) are deployable as-is.

Mode A — pure Python (example01)

Single function, managed image, typed I/O explicitly declared.

version: "1"

package:
  name: example01
  version: 0.1.0
  description: |
    Pedagogical example — analytical bolt stress under a force/moment
    torsor. Single function showcasing canonical types Torsor + Length →
    Stress on a managed Python runtime (mode A).
  author: MecaPy Team
  license: MIT
  visibility: public
  tags: [example, bolt, stress, torsor, mechanics]
  resources:
    recommended: { cpu: 1, memory_mb: 512 }
    timeout: 60

runtime:
  kind: python
  version: "3.12"
  requirements: false

functions:
  stress_bolt:
    handler: handler:stress_bolt
    description: |
      Compute the Von Mises equivalent stress on the ISO metric stress
      section of a bolt loaded by a force/moment torsor.
    inputs:
      torsor:
        type: Torsor
        description: |
          Force/moment torsor at the bolt section centroid, x-axis = bolt
          axis. F in newtons, M in newton-metres.
        required: true
      diameter:
        type: Length
        description: Bolt nominal diameter (m).
        required: true
      pitch:
        type: Length
        description: Thread pitch (m).
        required: true
    outputs:
      sigma_eq:
        type: Stress
        description: Von Mises equivalent stress (Pa) at the most loaded fibre.

def stress_bolt(torsor: dict, diameter: float, pitch: float) -> dict:
    """Compute Von Mises stress at the bolt's ISO stress section."""
    d_s = diameter - 0.9382 * pitch
    # ... combine axial + bending + shear + torsion at d_s ...
    return {"sigma_eq": ...}

The runner reads in/data.json, calls stress_bolt(**inputs), and serialises the returned dict to out/data.json. You write zero file plumbing.

Mode A — typed I/O omitted (introspection)

For a quick prototype you can drop the inputs: / outputs: blocks entirely:

version: "1"
package: { name: quick, version: 0.1.0 }
runtime: { kind: python, version: "3.12" }
functions:
  add:
    handler: math_utils:add

def add(a: float, b: float) -> float:
    return a + b

The deploy parser introspects the signature: a: Float, b: Float, output result: Float (the result key is the conventional name when the return is a single non-dict value). The function deploys and runs, but cannot be wired into a workflow — the workflow type-check requires explicit typed declarations. Use this shape for scratch packages only.

Mode B — custom Dockerfile (example02)

Native solver (code-aster) shipped as a custom image built from your Dockerfile. The wrapper reads in/, runs the solver, writes out/.

version: "1"

package:
  name: example02
  version: 0.1.0
  description: |
    Pedagogical example — linear static FEA of a 4-bolt support under a
    body torsor + linear acceleration, run on top of code-aster.
  visibility: public
  tags: [example, fea, code-aster, static]
  resources:
    recommended: { cpu: 2, memory_mb: 4096 }
    timeout: 600

runtime:
  kind: dockerfile
  dockerfile: Dockerfile
  context: .

functions:
  static_support:
    entrypoint: ["python3", "/workspace/_runner/wrapper.py"]
    description: Linear static analysis of a 4-bolt support.
    resources:
      recommended: { cpu: 2, memory_mb: 4096 }
      timeout: 1800
    inputs:
      mesh:
        type: File[med]
        description: Tetrahedral mesh with named groups (BOLT_1…BOLT_4, LOAD).
        required: true
      young_modulus: { type: Stress, description: Young's modulus (Pa) }
      poisson_ratio: { type: Float,  description: Poisson's ratio }
      torsor:        { type: Torsor, description: External torsor at LOAD group }
      acceleration:  { type: Vector3, description: Linear acceleration (m/s²) }
    outputs:
      result:        { type: File[med],     description: Result MED file }
      bolt_torsors:  { type: list[Torsor],  description: Reaction torsors at the 4 bolts }
      max_stress:    { type: Stress,        description: Max nodal Von Mises (Pa) }

FROM registry.scaleway.com/mecapy/code-aster:15.4
COPY wrapper.py /workspace/_runner/wrapper.py

"""Entrypoint — read /workspace/in, run code_aster, write /workspace/out."""
import json
from pathlib import Path

WS = Path("/workspace")
inputs = json.loads((WS / "in/data.json").read_text())
mesh = next((WS / "in/files").glob("mesh.*"))

# ... run aster, parse results ...

(WS / "out/data.json").write_text(json.dumps({
    "bolt_torsors": [...], "max_stress": ...
}))
# Result MED file written to /workspace/out/files/result.med

The full runtime contract is your responsibility in modes B/C. Mode-A’s runner.py is not present in this image.

Mode C — pre-built image

Reuses an image already published to a registry — no build, no context, just a pull-and-run.

version: "1"

package:
  name: openfoam-mvp
  version: 0.1.0
  visibility: internal
  organization: acme-cfd
  tags: [openfoam, cfd]
  resources:
    recommended: { cpu: 4, memory_mb: 8192 }
    timeout: 1800

runtime:
  kind: image
  image: ghcr.io/myorg/openfoam-mecapy:11.0

functions:
  channel_flow:
    entrypoint: ["/opt/run.sh", "--solver=simpleFoam"]
    inputs:
      case: { type: File[zip], description: OpenFOAM case directory zipped }
      reynolds: { type: Float,  description: Target Reynolds number }
    outputs:
      converged: { type: Boolean }
      cD:        { type: Float, description: Drag coefficient }
      cL:        { type: Float, description: Lift coefficient }

The image referenced by runtime.image must already implement the runtime contract — read /workspace/in/, write /workspace/out/, exit non-zero on failure. The platform pulls it once and reuses it for every run; tag pinning (:11.0 here) is recommended over :latest so reruns stay reproducible.

Multi-function package

Several functions in one package share the image. They distinguish themselves by handler (mode A) or entrypoint (modes B/C). Common metadata moves up to package; per-function overrides stay scoped:

version: "1"

package:
  name: bolt-analysis
  version: 0.3.1
  visibility: internal
  organization: acme-mech
  tags: [bolts, mechanics]
  resources:
    recommended: { cpu: 2, memory_mb: 2048 }   # default for every function below
    timeout: 300

runtime:
  kind: python
  version: "3.12"

functions:
  static_check:
    handler: bolts:static_check
    inputs:
      torsor:    { type: Torsor }
      diameter:  { type: Length }
    outputs:
      sigma_eq:  { type: Stress }

  fatigue_check:
    handler: bolts:fatigue_check
    description: Fatigue analysis using the Wöhler curve.
    resources:
      recommended: { cpu: 2, memory_mb: 4096 }   # this one needs more memory
      timeout: 600
    inputs:
      load_cycles: { type: list[Torsor] }
      diameter:    { type: Length }
      material:    { type: literal["8.8", "10.9", "12.9"] }
    outputs:
      cycles_to_failure: { type: Float }

  validate_assembly:
    handler: bolts:validate_assembly
    inputs:
      assembly:
        type:
          n_bolts: Integer
          spacing: Length
          torsor:  Torsor
    outputs:
      verdict: { type: Boolean }
      reason:  { type: String }

Three functions, one image, independent typed I/O, per-function resource overrides. The package version (0.3.1) bumps each time any of them changes — workflows using a specific version pin all three at once.

With testcase gating

quality.validation_testcases.on_deployment runs each function’s JSON test cases after the build and gates the version on the result:

version: "1"
package: { name: bolt-analysis, version: 0.3.1 }
runtime: { kind: python, version: "3.12" }

functions:
  static_check:
    handler: bolts:static_check
    testcases: tests/static_cases.json
    inputs:  { ... }
    outputs: { ... }

quality:
  validation_testcases:
    on_deployment: true
    fail_on_error: true     # gate the version if any case fails

{
  "version": "1",
  "function": "static_check",
  "test_cases": [
    {
      "name": "vertical bolt under axial load",
      "inputs": {
        "torsor":   { "F": [10000, 0, 0], "M": [0, 0, 0] },
        "diameter": 0.010,
        "pitch":    0.0015
      },
      "expected_ranges": {
        "sigma_eq": { "min": 1.93e8, "max": 1.97e8 }
      }
    }
  ]
}

After the build, the package enters the validating state while each case runs. With fail_on_error: true, any failed case moves the version to validation_failed (not servable); with false (default), failures are recorded as warnings and the version still becomes ready. Results appear on the package detail page.

Assertion blocks

Each case declares one or more assertion blocks:

• expected_outputs — exact match on non-File output ports.
• expected_ranges — {min, max} numeric bounds per output port.
• expected_files — assertions on a produced file, keyed by its filename. This covers both free artifacts (anything written to out/artifacts/) and declared File / list[File] output ports. Listing a filename asserts it exists; the value adds optional checks — min_size_bytes / max_size_bytes, sha256 (bit-exact), or equals (text equivalence to a reference fixture, with ignore_lines_matching to mask non-deterministic lines).
• expect_failure: true — the case passes only if the run fails (negative test); the expected_* blocks are then ignored.

A File input is passed as { "$file": "path" } and a list[File] input as { "$files": ["a", "b"] } — paths are relative to the test-case file and the fixtures ship in the package repo.