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:
    tier: nano

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:
    tier: medium

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:
      tier: medium
      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
  tags: [openfoam, cfd]
  resources:
    tier: large

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
  tags: [bolts, mechanics]
  resources:
    tier: small        # default for every function below

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:
      tier: medium     # this one needs more memory
    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 JSON test cases (referenced per-function) at deploy time and surfaces results in the deploy response:

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     # block deploy if any case fails

[
  {
    "name": "vertical bolt under axial load",
    "inputs": {
      "torsor":   { "F": [10000, 0, 0], "M": [0, 0, 0] },
      "diameter": 0.010,
      "pitch":    0.0015
    },
    "expected": {
      "sigma_eq": { "approx": 1.95e8, "rel_tol": 0.01 }
    }
  }
]

The platform runs each case and reports per-function status. With fail_on_error: true, a single failure turns the deploy into a 4xx; with false (default), failures are warnings and the deploy completes.