Conveyor
The conveyor feeds many files to decoders on one machine, in pure Python. It is not the pipeline (the steps inside a decoder). Read from the docstrings in src/conveyor/.
The package
The conveyor: storage (or network) -> memory -> decoders -> sinks, in pure Python.
A worked example of the kernels at scale, and a test of them: nothing in the kernels needs it, and
its file formats are the example's choice.
Not to be confused with the *pipeline* (decode, mask, untranspose, collect, route, dot), the chain
of steps one decoder runs on every block. The conveyor moves work to the decoders:
units consecutive frames of a file, one large contiguous read each (make_work_units), seconds of
work, so workers synchronise with the rest about once a second
workers each owns two buffers it first touches itself and a reader; decodes all the frames of
a unit straight from a buffer, gives the sink the result
readers one per worker: when a buffer is free, take the next unit from the shared queue and
read it with large preadv calls, in file order, while the worker decodes the previous
one (a worker that decodes faster takes more units: the load balances itself)
sinks where results go (a file format is the sink's business: see Sink)
Workers are threads or forked processes (run(mode=...)) around the same code; the decoders release
the GIL, so threads are the default. Pure Python 3: reading and decoding need numpy only, the chunk index of each file is read from a
sidecar file (tools/index_chunks.py builds them, with h5py, once), so reading touches no HDF5. The
parts that name them need scipy and pyFAI (Integrator), h5py and hdf5plugin (the writers),
ImageD11 (min_pixels) or cupy (the GPU backend). The writers refuse, before the run starts, a powder
of more than a tenth of the uncompressed frame (check_powder_size): the point is to reduce data.
Chunk indexes and units
make_work_units
make_work_units(paths, dset, max_unit_bytes=268435456, index_dir=None, harvest=False)
(units, frame shape, dtype): the work of reading and decoding the files, cut into WorkUnits.
A unit is as many consecutive frames of one file as fit in max_unit_bytes (bytes of the
compressed file, the input, not of any output), the whole file when it fits: a read is one large
contiguous request, and a unit is seconds of work, so workers synchronise with the rest about
once a second. Where the chunks lie is read from each file's sidecar index (see write_index).
index_dir where the sidecar indexes are kept (None: beside each file)
harvest when a file has no sidecar, read its index from HDF5 instead of raising
WorkUnit
class WorkUnit()
A unit of work (nothing to do with pyFAI's radial units): consecutive frames of one file that
one worker reads and decodes. file[start:start+span] holds frames frame0..frame0+n-1, the chunk
of frame frame0+i at offsets[i] (relative to start), lengths[i] bytes long; nfile is the number
of frames in the whole file.
Index
class Index(meta, offsets, lengths)
shape (nframes, ni, nj) of the frames stored (meta["declared_nframes"] is what HDF5 says;
fewer when the acquisition stopped early), dtype, codec (CODEC_LZ4 / CODEC_ZSTD), offsets
int64[nframes], lengths int32[nframes] (bytes of each frame's chunk in the file), dataset, meta
(the header).
read_index
read_index(path, index_dir=None, check=True)
Load the sidecar of path (an Index). check=True refuses one whose recorded file size or
modification time differ from the data file's. FileNotFoundError when there is none.
write_index
write_index(index, path, index_dir=None)
Write an Index to its small sidecar file beside the data file `path` (12 bytes per frame), so
that every later run finds the chunks without HDF5; returns the sidecar's path. It is written
under a temporary name and renamed, so a reader never sees a half-written index.
harvest
harvest(path, dset)
Read the chunk index of the HDF5 file path (h5py, once, serially). Returns an Index: where each
frame's compressed chunk lies in the file (.offsets, bytes from the start; .lengths) and what
the frames are (.shape, .dtype, .codec). Nothing is written: see write_index.
sidecar_path
sidecar_path(path, index_dir=None)
Where the chunk index of the data file `path` is kept: beside it, or under `index_dir`.
StaleIndex
class StaleIndex
The sidecar does not describe the data file as it is now.
Running
run
run(units, backend, workers, sink=None, mode='threads', batch=16, slots_per_worker=2, worker_cpus=None, reader_cpus=None, io_bytes=0, meta=None, start_timeout=900)
Run the units through the conveyor; returns a dict of what it took.
units from make_work_units; handed out in order to whichever worker is ready for one
backend CpuBackend, DecompressBackend or GpuBackend (anything with make_worker())
workers decoders; each has its own reader thread (or process, in "processes" mode)
sink factory k -> Sink, called in each worker (default: NullSink)
mode "threads" (default) or "processes" (forked): the same code, to compare
batch frames per decoder call
slots_per_worker buffers per worker (2: one decoding while the next is read)
worker_cpus list: worker k is pinned to worker_cpus[k % len]; None leaves it to the OS
reader_cpus list: every reader may run on any of these; None leaves it to the OS
io_bytes size of each read call (0: the whole unit in one call)
meta a dict the sinks get in begin(), with shape, dtype (from the backend's mask), nbins
(the rows of its matrix, None without one) and powder_scale (from an exact
backend) added; a writer refuses there, before any frame, what it will not write
start_timeout seconds to wait for every worker to start (pinned, buffers touched, decoder
built) before giving up
iterate
iterate(units, backend, workers, depth=4, **kw)
A generator of (file, frame0, npx, values, indices, powder), one tuple per batch of frames,
the arguments of Sink.write but copies the consumer may keep. Batches come in the order the
workers finish them, not file order: file and frame0 say where each belongs.
for file, frame0, npx, values, indices, powder in iterate(units, backend, workers=8):
...
The results wait in a queue of `depth` batches per worker: a consumer that is slower than the
decoders slows them down instead of filling memory. Threads only. Stopping the loop early
(break, or an exception) stops the run. Other keywords go to run (batch, ...).
report
report(r)
The run result r as text: input and output bytes, their ratio, and the rates.
numa_nodes
numa_nodes(allowed=None)
[[cpu, ...], ...]: the CPUs this process may use, grouped by NUMA node.
Decoders
CpuBackend
class CpuBackend(mask, dtype, csc=None, cut=0, howmany=None, thresholds=None, pipeline=None, exact=None, maxval=None)
bslz4_to_sparse.chunk2sparse, or chunk2sparseCSC with a matrix (the GIL is released).
mask, dtype: the detector mask and pixel type; csc: a pyFAI CSC matrix or None; cut, howmany,
thresholds: as chunk2sparse; pipeline: passed through to it; exact, maxval: as chunk2sparseCSC:
by default the powder is int64 fixed-point sums for 16-bit pixels (the scale goes to the sink's
meta as powder_scale) and float64 sums otherwise; exact=False for a matrix that is not a sum of
counts (rings, q-weighted rows).
CpuBackend.prepare
prepare(self)
Make what the workers share before they start, so that forked workers inherit it
instead of each making it.
CpuBackend.make_worker
make_worker(self)
A decoder for one worker: fn(buffer, offsets, lengths) -> (npx, values, indices,
powder).
DecompressBackend
class DecompressBackend(dtype, shape, unshuffle=True, batch=16)
The front half alone: bslz4_to_sparse.decompress into a preallocated array, no sparse step.
unshuffle=False gives each block as stored (before the bit unshuffle); True the pixel values.
Returns npx = 0 for every frame, values = the decompressed frames, no indices or powder.
dtype, shape: pixel type and frame shape; batch: the most frames per call (run's batch must
not be larger).
DecompressBackend.make_worker
make_worker(self)
A decompressor for one worker: fn(buffer, offsets, lengths) -> (npx, frames, None,
None).
GpuBackend
class GpuBackend(mask, dtype, csc=None, cut=0, batch=None, acc=None, howmany=None, thresholds=None, exact=None, maxval=None, group=None, unblock='gpu')
bslz4_to_sparse.cuda (cupy): one decoder per worker, each with its own CUDA stream.
Build the workers after any fork: CUDA must be initialised in the process that uses it.
mask, dtype, csc, cut: as CpuBackend; batch: frames decoded together (None: sized to the L2 cache);
group: frames uploaded together (None: the whole call), unblock: 'gpu' or 'cpu' (cuda.chunk2sparse); acc: accumulator type of the CSC
sum (None: 'i64' where exact applies, else 'f64'); howmany, thresholds: as chunk2sparse; exact,
maxval: as CpuBackend, with the same integers for the same matrix.
GpuBackend.make_worker
make_worker(self)
A GPU decoder for one worker: fn(buffer, offsets, lengths) -> (npx, values, indices,
powder).
Sinks
Layout and rules: sinks
Where results go. File formats are not the conveyor's business: implement Sink for yours.
A worker calls its sink (one per worker, made by the factory passed to run) once per batch:
write(file, frame0, npx, values, indices, powder)
file, frame0 which file (index into the paths given to make_work_units) and the first frame's number
npx (n,) int32: the number of sparse pixels of each of the n frames
values (n, npixels) pixel dtype, indices (n, npixels) uint32: row i holds frame i's pixels in
its first npx[i] entries
powder (n, nbins) float64 (int64 sums when the backend is exact: meta['powder_scale'] is the
integer to divide by for float), or None without a matrix
The arrays are the decoder's own buffers: they are overwritten after write returns, so a sink that
keeps anything copies it.
Layout and rules: sparsefile
ImageD11's sparse pixel file, one per input file, with the powder sums beside it.
ImageD11's lima_segmenter writes one sparse file per lima file, so workers never share a file and
a later step (assemble_label) gathers them. This sink does the same, so ImageD11 reads the result
as it reads its own. Layout, in the group named like the input dataset (the lima path):
row, col uint16 (npx,) pixel coordinates (dimension 0, dimension 1), frame after frame
intensity pixel dtype (npx,) the pixel values, always the input dtype
nnz uint32 (nframes,) the number of pixels of each frame
attributes nframes, shape0, shape1, npx, itype ("uint16", the type of row and col);
min_pixels_in_object when small objects were dropped
and, when the run has a matrix, in the same group
powder (nframes, nbins) the summed powder: int64 sums if the backend is exact (then the
attribute scale is the integer to divide by for float), else float64;
powder_dtype float32 or float64 stores it as float (the sums
were accumulated in 64 bits, scale applied, no scale attribute)
attributes of powder: scale (int64 only), integrator (the path of an Integrator file, if given)
With a stack (IntegratorStack) there is a dataset powder_<name> for every block, the rows of the
block, with the same attributes.
Only the sparse pixels are ImageD11's; powder and its attributes are ours, and ImageD11 ignores them.
A powder of more than a tenth of the frame (check_powder_size) is refused before the run starts.
No cleaning is applied unless min_pixels is given: then the pixels of connected
objects smaller than that are dropped (ImageD11's pixels_in_spot, its labelling). Datasets are compressed with the bitshuffle filter of hdf5plugin (compression
"bslz4", "bszstd", "lzf" or None).
A sink writes whole files: use make_work_units with max_unit_bytes large enough that a unit is a file.
Layout and rules: ringfile
Ring windows per frame, one file per input file, in the order the frames came in.
For texture pipelines: for every Bragg ring and every azimuth bin, the sum over a window of radial
bins of the normalised 2D cake, with no background subtracted, and, either side of the ring, the
same sum over two background windows. The reader decides what to do with them and how to sort the
frames into scan order. Layout, in the group named like the input dataset:
intensity (nframes, nwindows, nazim) float32 (or the dtype asked for): the window sums
window_ranges (nwindows, 2) lower and upper edge of each window, in `unit`
ring_index (nwindows,) the ring each window belongs to
role (nwindows,) int8: -1 background below the ring, 0 the ring, +1 background above
window_nbins (nwindows,) how many radial bins each window holds (the sums are not means)
azimuthal (nazim,) the azimuth of each bin, degrees
attributes unit, nframes, integrator (path of the Integrator file, if given)
Frame i of the file is frame i of the input file. Written like ImageD11SparseSink, one whole input
file at a time, and refused like it before the run when the run's powder is too large
(check_powder_size).
Sink
class Sink()
Where one worker's results go: subclass it and implement write (and begin, end if needed).
Sink.begin
begin(self, meta)
called once in the worker before the first batch; meta: dict (dtype, shape, ...)
Sink.begin_unit
begin_unit(self, unit)
called before the batches of a unit (a WorkUnit of make_work_units: file, frame0, n, lengths)
Sink.write
write(self, file, frame0, npx, values, indices, powder)
Receive one batch of frames (arguments in the module docstring).
Sink.end_unit
end_unit(self, unit)
called after the last batch of a unit
Sink.end
end(self)
flush and close; return a dict of statistics
NullSink
class NullSink()
Counts and throws everything away: measures the reading and decoding alone.
NullSink.write
write(self, file, frame0, npx, values, indices, powder)
Count the frames and pixels of the batch.
NullSink.end
end(self)
The counts: dict(frames, pixels).
TeeSink
class TeeSink(*sinks)
Several sinks in one: every call goes to each of them, in order (sparse pixels to a pixel
writer and the powder to a ring writer, say). end() merges their statistics.
TeeSink.begin
begin(self, meta)
called once in the worker before the first batch; meta: dict (dtype, shape, ...)
TeeSink.begin_unit
begin_unit(self, unit)
called before the batches of a unit (a WorkUnit of make_work_units: file, frame0, n, lengths)
TeeSink.write
write(self, file, frame0, npx, values, indices, powder)
Receive one batch of frames (arguments in the module docstring).
TeeSink.end_unit
end_unit(self, unit)
called after the last batch of a unit
TeeSink.end
end(self)
flush and close; return a dict of statistics
check_powder_size
check_powder_size(meta)
Refuse, before anything runs, a powder of more than POWDER_MAX_FRACTION of the frame it comes
from: nbins sums of 8 bytes against the uncompressed frame (shape x pixel size). The writers
call it in begin(); a 1D powder or ring windows pass, a cake as large as the image does not.
meta: the run's meta (shape, dtype, nbins; nbins None or absent: no powder, nothing to check).
ImageD11SparseSink
class ImageD11SparseSink(out_dir, dset, name=imaged11_name, compression='bslz4', powder_dtype=None, integrator=None, stack=None, min_pixels=None)
out_dir where the files go (created); dset the dataset path of the input, which is the
group written; name function input path -> file name (imaged11_name); compression see the
module docstring; powder_dtype None (as the decoder gives it: int64 or float64), np.float32 or
np.float64; integrator path of an Integrator file to record beside the powder; stack the
IntegratorStack the run's matrix is: each block is written as its own dataset powder_<name>
(raw sums of the block's rows, with the normalisation in the integrator file); min_pixels drop
the pixels of connected objects (by ImageD11's labelling) smaller than this, 2 or 3 say: single
pixels are mostly noise, and the file is much smaller (None: keep every pixel; needs ImageD11).
ImageD11SparseSink.begin
begin(self, meta)
called once in the worker before the first batch; meta: dict (dtype, shape, ...)
ImageD11SparseSink.begin_unit
begin_unit(self, unit)
called before the batches of a unit (a WorkUnit of make_work_units: file, frame0, n, lengths)
ImageD11SparseSink.write
write(self, file, frame0, npx, values, indices, powder)
Keep the batch (trimmed to its pixels) until the file is complete.
ImageD11SparseSink.end_unit
end_unit(self, unit)
Write the file of the unit.
ImageD11SparseSink.end
end(self)
The files written.
imaged11_name
imaged11_name(path)
ImageD11's name for the sparse file of an input file: / become _, .h5 becomes _sparse.h5
RingSink
class RingSink(integrator, out_dir, dset, ring_index=None, role=None, name=rings_name, compression='bslz4', dtype=numpy.float32, integrator_path=None, rows=None)
integrator the "rings" Integrator the run's backend uses (csc=integrator.csc, float dot)
out_dir, dset, name, compression as ImageD11SparseSink (the default name ends in _rings.h5)
dtype what intensity is stored as (np.float32); integrator_path recorded in attributes
ring_index, role from peak_windows: which ring each window is for, and below/ring/above
rows the rows of the powder that are this integrator's (a slice, from
IntegratorStack.slices) when the run's matrix is a stack
RingSink.begin
begin(self, meta)
called once in the worker before the first batch; meta: dict (dtype, shape, ...)
RingSink.begin_unit
begin_unit(self, unit)
called before the batches of a unit (a WorkUnit of make_work_units: file, frame0, n, lengths)
RingSink.write
write(self, file, frame0, npx, values, indices, powder)
Normalise the batch and keep it as (frames, windows, azimuth) until the file is done.
RingSink.end_unit
end_unit(self, unit)
Write the file of the unit.
RingSink.end
end(self)
The files written.
rings_name
rings_name(path)
The name of the ring file of an input file: / become _, .h5 becomes _rings.h5
peak_windows
peak_windows(centres, width, background_width=None, gap=0.0)
(ranges, ring_index, role): for each ring centre a window of `width` around it with the
background windows of background_width (default: width) either side, `gap` away from it, all
in the unit of the centres. Windows are half-open [lower, upper) like Integrator.rings. Order:
below, ring, above, for ring after ring.
Integrator state
Layout and rules: integrator
The integrator state: what is needed to turn the summed powder of a run into intensities.
A conveyor run returns, per frame, the sum over the pixels of each bin of weight x pixel value
(pyFAI's sum_signal). pyFAI's intensity is that divided by sum_normalization, the same sum for an
image of ones with the corrections (flat, solid angle, polarisation, ...) folded in. An Integrator
keeps what that needs, in one file that results files point to:
poni the pyFAI geometry, as the text of a .poni file
csc the sparse matrix (bins x pixels), CSC: data, indices, indptr
norm sum_normalization of an image of ones, one per bin (float64)
radial the bin positions in `unit` (the azimuth too for a 2D matrix); ring_ranges
(nrings, 2) in `unit` when the bins are rings; offset, per bin, added to the
normalised intensity (the background of rings)
Kinds: "powder1d", "powder2d", "rings", and "moment" (a 1D matrix and, below it, the same with every
pixel's weights multiplied by its q: the sum of q x signal per bin; Integrator.first_moment). An
IntegratorStack puts several of them in one matrix, so one pass gives all of them.
Rows of the matrix: "powder1d" the radial bins; "powder2d" radial-major, row = radial * nazim +
azimuth; "rings" azimuth-major, row = azimuth * nrings + ring (reshape to (nazim, nrings)).
File (HDF5; hdf5plugin supplies the bitshuffle filter): attributes bslz4_to_sparse_integrator
(format version), kind ("powder1d", "rings", "powder2d" or "moment"), unit, shape (the detector), nbins,
bins_shape (what the powder of one frame reshapes to); datasets poni, norm, radial,
azimuthal, ring_ranges and offset when present; group csc (attribute format = "CSC", shape = (nbins, npix))
with datasets data (float32), indices (int32), indptr (int32), as scipy and pyFAI name them.
Needs pyFAI only for from_pyfai and ai(), h5py only for save and load.
Integrator
class Integrator(csc, norm, shape, unit, radial=None, azimuthal=None, kind='powder1d', poni=None, ring_ranges=None, offset=None)
csc scipy.sparse CSC matrix (nbins, npix); from_pyfai, rings, first_moment and the stack
make float32 weights
norm (nbins,) float64, sum_normalization of an image of ones
shape (ni, nj) of the detector; unit the radial unit (pyFAI name, e.g. "q_nm^-1")
radial bin positions in unit: (nrad,) for 1D or rings, with azimuthal (nazim,) for 2D
kind "powder1d", "rings", "powder2d" or "moment"; poni text of a .poni file
ring_ranges (nrings, 2) lower and upper edge of each ring in unit (kind "rings");
offset (nbins,) added to the intensity (kind "rings": minus the background of the ring)
Integrator.nbins (property)
The number of rows of the matrix: the length of the powder of one frame.
Integrator.bins_shape (property)
The shape the powder of one frame reshapes to (see the rows of the matrix in the module docstring; "moment": the first nbins // 2 rows, then the q-weighted ones).
Integrator.from_pyfai
from_pyfai(ai, npt, unit='q_nm^-1', npt_azim=None, split='bbox', **integrate)
The matrix and normalisation of a pyFAI AzimuthalIntegrator: npt radial bins (and
npt_azim azimuthal ones for a 2D result). split is pyFAI's pixel splitting ("no", "bbox",
"full"). integrate: keywords for pyFAI's integrate1d/2d on the image of ones, so that
correctSolidAngle, polarization_factor, flat, normalization_factor ... are the ones the
data will be corrected with. The detector mask of ai is in the matrix.
Integrator.ai
ai(self)
The pyFAI AzimuthalIntegrator the poni text describes.
Integrator.intensity
intensity(self, powder, scale=None)
Normalised intensity of summed powder: powder (nbins,) or (nframes, nbins), integers
divided by scale (from the decoder: its power of two) or floats. Empty bins (norm 0) are 0, as
in pyFAI.
Integrator.rings
rings(self, ranges, background=None)
A "rings" Integrator from this 2D one: for each (lower, upper) in ranges (in unit),
every azimuth bin gets the sum over the radial bins whose centre lies in [lower, upper) of
the normalised intensity (summed signal / norm of each bin), minus the sum over them of
`background` (a value per radial bin, default none). The division by each bin's norm is
folded into the weights, so the result of the matrix product already is the intensity: use
a float dot (not exact), norm is 1. This is the sum of the normalised 2D cake over each
peak window, per azimuth, that texture pipelines compute from pyFAI's integrate2d.
Integrator.save
save(self, path, compression=True)
Write the integrator to an HDF5 file (see the module docstring).
Integrator.load
load(path)
Read an integrator file written by save.
Integrator.first_moment
first_moment(self, q=None)
A "moment" Integrator from this 1D one: the same matrix with, below it, the same
matrix with every pixel's weights multiplied by that pixel's q (in unit; default: pyFAI's
value at the pixel centre), so one product gives, per bin, the sum of weight x signal and
the sum of weight x q x signal. Their ratio is the mean q of the signal in the bin. The
weights are no longer 1-like numbers and the second half cannot be normalised by an image
of ones: use a float dot (not exact). Rows: the first nbins, then the moment.
Integrator.mean_q
mean_q(self, powder)
Of a "moment" integrator's powder: the mean q of the signal of each bin, the second
half of the rows over the first (0 where the bin has no signal).
IntegratorStack
class IntegratorStack(blocks)
Several Integrators of one detector as one matrix: everything comes out of one pass over
the pixels, whatever their number (a 1D powder, ring windows, the first moment ...), each pixel
reaching all of their bins. blocks: a dict, name -> Integrator (the order is kept).
csc the blocks' matrices stacked, rows block after block (float: a block of rings or of
moments makes the whole product float, so use the float dot, not exact)
slices name -> the rows of csc, and of every powder, that belong to the block
split(powder, scale=None) name -> the intensity of that block (powder: (nbins_total,) or (nframes, ...))
raw(powder) name -> the rows of the powder of that block, as the decoder gives them
IntegratorStack.raw
raw(self, powder)
name -> the rows of the powder of each block.
IntegratorStack.split
split(self, powder, scale=None)
name -> the intensity of each block (Integrator.intensity of its rows).
IntegratorStack.save
save(self, path, compression=True)
Write the blocks to one HDF5 file, a group each (their matrices are not repeated as one).
IntegratorStack.load
load(path)
Read a stack written by save.
fixed_weights
fixed_weights(csc, mask, dtype, maxval=None)
(W, scale): the integer weights of bslz4_to_sparse.fixed_point_matrix(csc, mask, dtype,
maxval) (what the "csc-int" dot uses), as a scipy CSC matrix of int64, and the scale (integer, divide
the sums by it).
clip_exact_powder
clip_exact_powder(powder, npx, values, indices, wtmax, weights, in_place=False)
The int64 powder (nframes, nbins) as if every pixel had been clipped to wtmax before the
sum. npx, values, indices: the sparse output of the same frames (rows of values and indices
hold the pixels of one frame in their first npx[i] entries); weights: W of fixed_weights.
Every pixel above wtmax must be in the sparse output (see the module docstring).
Generated by tools/generate_docs.py from bslz4_to_sparse 0.0.22a2 (git v0.0.22-13-g610ed27, sources sha256 6929bd75ab7a).