Python API¶
The main entry points. Every tool is also callable through the registry (see
Tools): registry.get("eeg.preprocess").run({...}, ctx).
Running tools¶
braino.context.RunContext(dataset_root=None, run_id=new_run_id(), record_external_paths=False)
dataclass
¶
Where a run reads its data from and writes its outputs to.
dataset_root = None
class-attribute
instance-attribute
¶
Root of the BIDS dataset, or None for runs that do not touch a dataset.
record_external_paths = False
class-attribute
instance-attribute
¶
Record the full paths of input files outside the dataset. Off by default, because the names of original recordings often identify the person.
provenance = field(init=False, repr=False, compare=False)
class-attribute
instance-attribute
¶
Records every tool call of the run; written under runs/<run_id>/ if there is a
dataset.
proposals = field(init=False, repr=False, compare=False)
class-attribute
instance-attribute
¶
Proposals made by decision tools, awaiting or holding the user's decision.
dataset
property
¶
The dataset root; raises :class:NoDatasetError if the run has none.
derivatives_root
property
¶
<dataset>/derivatives/braino, a BIDS-Derivatives dataset holding all outputs.
record_input(path)
¶
Record (and hash) a file the running tool reads. Returns path.
record_output(path)
¶
Declare a file the running tool writes; it is hashed when the tool finishes.
Returns path.
output_dir(*parts)
¶
Create and return a folder under the derivatives root, e.g.
ctx.output_dir("sub-01", "eeg").
Also writes the derivatives dataset_description.json the first time it is needed.
braino.tools.registry.ToolRegistry()
¶
braino.tools.decisions.decide(proposal_id, context, *, approve, decided_by, edits=None, note=None, tools=None)
¶
Approve or reject a pending proposal and record the decision.
On approval, edits (top-level fields to change, e.g. a shorter list of channels) are
merged into the proposal, the result is validated against the tool's output model, and the
tool's apply function commits it. If apply fails, the proposal stays pending.
Pipelines¶
braino.pipelines.spec.PipelineSpec
¶
Bases: BaseModel
braino.pipelines.spec.load_spec(source, *, tools=None)
¶
Read and check a spec from a .yaml/.yml/.json file, or the name of a built-in
spec (see :func:builtin_specs).
braino.pipelines.runner.run_pipeline(spec, ctx, *, subjects=None, from_step=None, auto_approve=None, tools=None, progress=None)
¶
Run spec over the dataset of ctx.
subjects limits the run to some of the spec's subjects (the group steps then use only
those). auto_approve is the identity recorded as approving every proposal, e.g.
"user:alice (auto-approved in a pipeline run)"; without it, decision steps wait.
braino.pipelines.runner.PipelineReport
¶
Bases: BaseModel
Provenance and export¶
braino.provenance.store.load_run(derivatives_root, run_id)
¶
braino.provenance.export.render_script(steps, *, title, environment, inputs)
¶
The Python script that replays steps.
braino.provenance.replay.Replay(context, recorded, differences=list(), n_steps=0, out=None)
dataclass
¶
The state of a running exported script.
run(tool, arguments, *, step)
¶
Run one tool call as recorded, and compare its result.
decide(pending, *, step, approve, edits=None, decided_by, note=None)
¶
Make the recorded decision again on the new proposal: the same approval or rejection, with the same edits.
finish()
¶
Print whether the results were reproduced; returns the exit code.
Local model runtime¶
braino.runtime.tiers.recommend(hw, cat=None)
¶
Propose a tier and model for hw.
braino.runtime.manager.setup(model=None, backend=None, *, benchmark=True, progress=None, hw=None, cat=None)
¶
Install the runtime and model for this computer, benchmark it, and save the choice.
progress(label) may return a callback for the download of label.
braino.runtime.manager.start_server(settings=None, *, cat=None, port=None, start=True)
¶
A server for the chosen model (started unless start=False).
braino.runtime.server.LlamaServer(model, *, binary, placement='cpu', ctx_size=DEFAULT_CTX, threads=None, port=None, alias=None, extra_args=(), log_path=None)
¶
A llama-server process serving one model on localhost.
binary is the server executable, or a command prefix (used by the tests to run a
stand-in server). placement="cpu" keeps everything off the GPU; otherwise llama.cpp
fits as much of the model as the GPU holds (--fit, its default), which also puts a
mixture-of-experts model's experts in RAM when the GPU is too small.
Use it as a context manager, or call :meth:start and :meth:stop::
with LlamaServer(model_path, binary=find_server()) as server:
server.post("/completion", {"prompt": "Hello", "n_predict": 8})
start(timeout=600.0)
¶
Start the server and wait until the model is loaded (/health answers 200).
LLM client¶
braino.llm.client.ChatClient(base_url, *, api_key=None, model='local', llama_cpp=True, thinking=False, timeout=600.0, retries=2, retry_pause=1.0)
¶
Talks to one model on one server.
base_url ends in /v1 (e.g. http://127.0.0.1:9931/v1). Requests that fail to
connect, or meet a server that is still loading (HTTP 503), are retried retries times.
for_server(server, **kwargs)
classmethod
¶
A client for a started :class:~braino.runtime.LlamaServer.
chat(messages, *, json_schema=None, max_tokens=512, temperature=0.0, thinking=None, extra=None)
¶
One chat completion. With json_schema, the reply is constrained to match it
(on llama.cpp by a grammar, so it cannot do otherwise).
count_tokens(text)
¶
Tokens in text by the model's tokenizer (llama.cpp only; else None).
braino.llm.conversation.Conversation(client, request, *, tools=None, preload=None, ctx_size=DEFAULT_CTX, max_tokens=768, max_loads=3, can_ask=True)
¶
One request, step by step. :meth:next_step returns the model's next action
(call, decide, ask or answer; tool loading is handled here), and the
caller reports what came of it with :meth:add_result.
tokens
property
¶
Tokens in the conversation so far (measured at the last reply, estimated after).
next_step()
¶
Ask the model for its next action.
add_result(result, *, proposal_id=None)
¶
Report what the last action led to: a tool's result (or error), the user's
decision, or the user's reply to a question. A decision tool's proposal_id
becomes something the model can ask the user to decide.
add_user_reply(reply)
¶
The user's answer to an ask step.
braino.llm.steps.step_schema(loaded, loadable=(), proposals=(), *, can_ask=True)
¶
The JSON schema of every action possible at this step.
braino.llm.select.select_tools(request, tools, *, limit=MAX_PRELOAD)
¶
Up to limit tools the request's words point to, best match first; the dataset
summary when nothing matches (the usual first step).
Agent evaluation¶
braino.agent.session.Session(context, *, tools=None)
¶
Bases: ABC
One request worked on by one agent, in one run context.
user_name
abstractmethod
property
¶
Recorded as who decided, e.g. user:alice.
schemas(names=None)
¶
LLM-facing schemas of all tools, or of the names given.
call(tool, arguments=None)
¶
Run a tool; errors are returned, not raised.
request_decision(proposal_id, message='')
¶
Ask the user about a proposal; returns their decision (and its application).
braino.evals.runner.run_suite(tasks, factory, *, workdir=None, tier=None, model=None, progress=None)
¶
braino.evals.tasks.EvalTask
¶
Bases: BaseModel
Importing¶
braino.io.importer.import_recording(source, bids_root, *, key, original_id, task, session=None, run=None, datatype=None, line_freq=None, rename=None, channel_types=None, montage='auto', stim_channels=None, cnt_data_format='auto', identifiers=(), overwrite=False)
¶
Import the recording at source as the person original_id (looked up in, or added
to, key) into bids_root. See :func:~braino.io.normalise,
:func:~braino.io.deidentify and :func:~braino.io.bids.write_bids for the options.
braino.io.deidentify.SubjectKey(path, subjects)
¶
The private table linking subject labels to original identifiers, kept in a JSON file outside the dataset folder.
open(path, *, dataset_root)
classmethod
¶
Open the key at path, or start a new one there.
Raises :class:SubjectKeyError if path is inside dataset_root.
entry(original_id, meas_date)
¶
The subject for original_id, assigning a label and date shift the first time.
Saves the key when it changes. Raises :class:SubjectKeyError if meas_date cannot
be shifted with the subject's existing shift into the allowed window.
Validation data¶
braino.datasets.erpcore.fetch(task, subjects=None, *, stimuli=False, verify=False)
¶
Download (or reuse) the BIDS data of task for subjects; returns the BIDS root.
Files already in the cache are reused if their size matches; verify=True re-checks
their hashes too. Raises :class:DownloadError if a download does not match its hash.
braino.validation.erpcore.run(component, subjects=None, *, progress=None)
¶
Download the ERP CORE data of component (all participants by default), run the
built-in erpcore-<component> pipeline on it, and compare the scores with the published
ones.
Decisions (ICA components, bad channels) are approved as proposed, in the name of the person running this, and recorded as auto-approved. Results are cached: running again only redoes what changed.
braino.validation.erpcore.compare(component, scores_uv, *, excluded=())
¶
Compare BrainO's per-participant scores (subject label -> µV, for the participants BrainO
kept) with the published component result. excluded lists the participants BrainO
excluded.