Developer utilities - ULJ-Yale/qunexsdk GitHub Wiki
QuNex ships a number of reusable classes and helper utilities that developers can build on when writing new commands, rather than re-implementing the same logic each time. This page collects them. It is intended to grow as more shared building blocks are documented.
QuNex commands report what they did through a log object, which collects the
report as the command works and hands it on when the command is done. The
classes live in python/qx_utilities/general/log/, and they divide the job in
three: one set describes what a report says, one owns the files it is written
to, and one decides whether and where anything is written at all.
| Class or helper | What it provides |
|---|---|
ReportLog |
The report itself: the level methods (step, detail, warning, error, info, blank, raw), nesting (section, indent, depth=), a failure count derived from the errors recorded on it, and the (session_id, summary, failed) status read off it. Used by per-BOLD and per-group executors, study-level commands, and utility commands |
SessionLog |
A ReportLog for a command working on one session. It writes the session header as it is created, adds the closing footer at finish(), and carries the session id. Used by session- and subject-level processing commands |
log_or_console(_log) |
The caller's log, or — when there is none — a stand-in that echoes to the console, so that a function reached from both kinds of caller keeps one body |
RunContext |
One per qunex invocation: the run's log folder, its single runlog, the closing manifest, and the status record --logstatus asks for |
ComContext |
One comlog: the record of a single external call. It creates the file as tmp_…, renames it to done_, error_ or incomplete_ on close, and offers its open handle for subprocess
|
LogSettings, resolve_logging()
|
The resolved logging configuration for one invocation, folded from the built-in defaults, the user and study settings files, the command's registry declaration, and the command line |
Both contexts stay usable when logging is switched off — the writes are dropped — so a command can record its report without asking whether anyone is listening.
A call site states what a line is, and the log renders the marker and the indent that belong to it. The comment after each call below is the line it produces.
import qx_utilities.general.log as gl
# a session log writes its header — session id, start time, pipeline name
log = gl.SessionLog(sinfo, options, "My pipeline")
# a utility command instead receives its log, and opens with:
# log = gl.log_or_console(_log)
log.info("Processing 3 BOLD runs.") # Processing 3 BOLD runs.
log.blank() # a blank line
log.step("checking the data") # ---> checking the data
log.detail(f"found {path}") # ... found /studies/...
log.warning(
"no fieldmap, using the default"
) # ---> WARNING: no fieldmap, using the default
log.error(f"{path} does not exist!") # ---> ERROR: /studies/... does not exist!
with log.section("preparing the BOLD runs"): # ---> preparing the BOLD runs
log.detail("bold 1: rest") # ... bold 1: rest
# the block is indented under the heading
log.action("Running", "FSL feat", options["run"])
# ---> Running FSL feat
# ---> Test running FSL feat (under --test)
log.pipeline_command(command) # the assembled call, framed, one flag per line
log.raw(str(errormessage)) # verbatim: a tool's output, an exception's text
log.blank() # a blank line
log.rule() # a full width rule, always the same width
log.rule(char=".") # a lighter division, for a block inside the report
return log.finish("My pipeline completed") # the summary, and the log itselflog.error() does more than choose a marker: it adds to the log's error count,
which is what decides the session's status, what the run's closing manifest
reports, and the exit status of the qunex process. So a failure is best stated
with error() rather than described inside a step() or an info().
The rules that go with these methods — where a command's log comes from, why the
parameter is spelled _log, what a processing command returns, and the pitfalls
worth knowing — are on the Logging page, with the whole picture in
The QuNex logging whitepaper.
When QuNex reads the sessions listed in a batch file, it returns them as a SessionList
(defined in python/qx_utilities/general/batch_io.py). A SessionList is a light subclass of
Python's list (UserList), so it behaves exactly like a list of session dictionaries —
you can iterate over it, index it, and pass it around as usual. On top of that, it offers
a handful of convenience methods for the operations that commonly come up when working
with sets of sessions: filtering, grouping, and pulling out values. These are used both
internally by QuNex (for example, subject-level processing groups the sessions with
group_by_key("subject")) and in command code.
The available methods:
| Method | Returns |
|---|---|
filter_by_key(key, value) |
A new SessionList of the items where key matches value. value may be a comma-separated list, and values may use glob patterns (*, ?, [abc]). |
filter_by_string(filter) |
A new SessionList selected with <key>:<value> pairs joined by | (OR) or & (AND) — one operator type at a time. Values are treated as globs. |
get_list_by_key(key, sep=",") |
The unique values found for key, as a separator-joined string; pass an empty sep to get a Python list instead. |
group_by_key(key) |
A list of SessionLists, one for each distinct value of key. |
have_key(key) |
A new SessionList of the items that have a non-empty value for key. |
dont_have_key(key) |
A new SessionList of the items that are missing key or have it empty. |
copy() |
A deep copy of the SessionList. |
Because the filtering and grouping methods return new SessionLists, they can be chained
and combined naturally:
# all sessions in the "patients" group that have a rest BOLD
patients = sessions.filter_by_string("group:patients&task:rest")
# the distinct subjects present in the list
subjects = sessions.get_list_by_key("subject", sep="")
# process each subject's sessions together
for subject_sessions in sessions.group_by_key("subject"):
...To obtain a SessionList in the first place, call gc.resolve_sessions() — the single
entry point for "which batch file, which sessions", which also returns the parameters the
batch file's header states:
import qx_utilities.general.core as gc
sessions, header = gc.resolve_sessions(
batchfile=options["batchfile"], # the path to the batch file
sessions=options["sessions"], # ids or globs, selecting within it
filter=options.get("filter"), # <key>:<value> criteria
sessionsfolder=options["sessionsfolder"],
command="my_command", # named in the error if this fails
)The parsing and selection themselves live in general/batch_io.py, which imports nothing
outside the standard library because bin/qunex_container carries a spliced copy of it.
The working guide is on the Invocation page; the whole picture is in
The QuNex invocation whitepaper.