Workflows¶

The client.workflows namespace composes the resource endpoints (projects, studies, recordings, analyses) into single calls: create a project and study, upload the recording, submit an analysis, wait for it to finish, and fetch results, without threading resource IDs by hand or constructing request models.

It adds no routes of its own. The low-level endpoints remain available.

All workflows methods are keyword-only.

Analyze a file¶

analyze_file runs the whole pipeline and keeps everything it creates:

async with client:
    result = await client.workflows.analyze_file(
        file="night.edf",
        analysis_config_id="u-sleep-research-v2",
    )

    print(result.project_id, result.study_id, result.recording_id)
    print(f"{len(result.events)} events")
    print(f"TST (min): {result.biomarkers.biomarkers.tst_min}")

flowchart LR
    A[create project<br/>and study] --> B[upload]
    B --> C[submit]
    C --> D[poll until done]
    D --> E[fetch events,<br/>statistics, report]

When study_name is omitted it is inferred from the filename (night.edf → study night). The default fetch returns both events and biomarkers (biomarkers only when the analysis produced sleep staging).

Subject metadata¶

Pass metadata as plain scalars; no model construction is required. To set metadata across many existing studies at once, see study metadata in bulk:

result = await client.workflows.analyze_file(
    file="night.edf",
    analysis_config_id="u-sleep-research-v2",
    study_name="Subject 001",
    age=35,
    sex="male",
)

Analyze into an existing project or study¶

Supply IDs to reuse existing resources instead of creating new ones:

result = await client.workflows.analyze_file(
    file="night.edf",
    analysis_config_id="u-sleep-research-v2",
    project_id=existing_project_id,
    study_id=existing_study_id,   # requires project_id
)

Choosing what to fetch¶

result = await client.workflows.analyze_file(
    file="night.edf",
    analysis_config_id="u-sleep-research-v2",
    fetch=("events", "biomarkers", "report"),   # any subset; default is events + biomarkers
)

Ephemeral analysis (auto-cleanup)¶

analyze_file_temporary is an async context manager that deletes whatever it created when the block exits, including on exception. A caller-supplied project_id or study_id is never deleted.

async with client.workflows.analyze_file_temporary(
    file="night.edf",
    analysis_config_id="u-sleep-research-v2",
    fetch=("biomarkers",),
) as result:
    store_biomarkers(result.biomarkers)
# the created project, study, recording, and analysis are gone here

Cleanup deletes the single top-most resource the workflow created (the project when it made one, otherwise the study, otherwise the recording); the server cascades the deletion to descendants and frees quota. Cleanup is best-effort: a delete failure is logged and never masks an error from inside the block.

Choosing between them. Use analyze_file when the recording and results should persist on the platform. Use analyze_file_temporary for a one-off analysis where you only need the results in-process. There is no cleanup flag; the choice is made by the method name.

Handling failures¶

If the analysis finishes in a non-completed state, an AnalysisWorkflowError is raised, carrying the terminal AnalysisOut and (when available) its logs:

from yousleep_common.client import AnalysisWorkflowError

try:
    result = await client.workflows.analyze_file(
        file="night.edf", analysis_config_id="u-sleep-research-v2"
    )
except AnalysisWorkflowError as exc:
    print(f"Analysis {exc.analysis.id} ended as {exc.analysis.status.value}")
    if exc.logs:
        print(exc.logs.stderr)

analyze_file accepts a timeout (default 3600s); exceeding it raises ClientTimeoutError from yousleep_common.client.exceptions. The builtin TimeoutError does not catch it.

Analyzing multiple files¶

analyze_files scores a whole set of recordings in one call. It creates one project and one study per file (names inferred from filenames), uploads with bounded concurrency, then submits and polls everything through the server's batch endpoints: one submission call and one status call per poll, regardless of batch size, up to the server's batch limit of 500 files by default:

result = await client.workflows.analyze_files(
    files=["sub-01.edf", "sub-02.edf", "sub-03.edf"],
    analysis_config_id="u-sleep-research-v2",
)
for outcome in result.succeeded:
    print(outcome.file, outcome.result.biomarkers)
for outcome in result.failed:
    print(outcome.file, outcome.status, outcome.error)

Unlike analyze_file, per-file failures never raise: each input file yields a FileAnalysisOutcome with status "completed", "failed", or "timeout" (the shared timeout bounds the whole batch), and the rest of the batch continues. Only argument errors and whole-batch failures raise.

analyze_files creates one study per file. study_id targets an existing study and is therefore only valid with a single file, because a study holds exactly one recording; passing it with several files raises ValueError.

Several nights of one subject are several studies, one per night, linked by a shared subject_key; see study metadata in bulk.

Per-file subject metadata is not supported here. Create studies with client.batch.create_studies and use the lower-level endpoints if you need it.

Track upload progress across the batch with on_progress, which receives the original file reference and its percentage (calls from concurrent uploads interleave):

result = await client.workflows.analyze_files(
    files=paths,
    analysis_config_id="u-sleep-research-v2",
    on_progress=lambda file, pct: print(f"{file}: {pct}%"),
)

analyze_files_temporary is the ephemeral variant, mirroring analyze_file_temporary:

async with client.workflows.analyze_files_temporary(
    files=paths, analysis_config_id="u-sleep-research-v2"
) as result:
    export([o.result.biomarkers for o in result.succeeded])
# the created project (and everything under it) is gone here

Waiting on an analysis directly¶

When orchestrating the steps yourself, use analyses.wait_for to block until an analysis reaches a terminal state. It is the same helper the workflows use internally:

analysis = await client.analyses.wait_for(
    analysis_id, poll_interval=5.0, timeout=3600
)
if analysis.status == AnalysisStatus.COMPLETED:
    events = await client.analyses.get_events(analysis.id)

Result object¶

analyze_file (and the context manager) return an AnalysisWorkflowResult:

Attribute	Description
`project_id` / `study_id`	UUIDs of the (created or supplied) project and study
`recording`	The uploaded `RecordingOut` (`recording_id` is a shortcut)
`analysis`	The completed `AnalysisOut` (`analysis_id` is a shortcut)
`events`	`list[Event]` if `"events"` was fetched, else `None`
`biomarkers`	`AnalysisBiomarkersResponse` (the result plus its presentation `table`) if `"biomarkers"` was fetched and staging ran, else `None`
`report`	`StudyReportResponse` — event stats plus server-computed `hypnograms` and presentation `tables` — if `"report"` was fetched, else `None`

events is always the flat list. For dense per-epoch output at scale, where a full night of 1 Hz staging is tens of thousands of events, omit "events" from the fetch set and read client.analyses.get_event_blocks instead.

analyze_files returns a BatchAnalysisWorkflowResult: the project_id plus one FileAnalysisOutcome per input file (in input order), each carrying the file, a status ("completed"/"failed"/"timeout"), the per-file AnalysisWorkflowResult on success, and an error description otherwise. .succeeded and .failed filter the outcomes.