Skip to content

niess.nexus

McCode to ESS NeXus Structure JSON. See Produce NeXus Structure JSON for the task-oriented guide and Write NeXus translators for extending it.

nexus

McCode to ESS NeXus Structure JSON conversion.

Dispatches translators over the assembled mccode_antlr Instance tree, the same handoff point :mod:niess.brep uses, and builds NeXus Structure JSON dicts directly -- there is no intermediate NeXus object model.

from niess.nexus import to_nexus_structure
structure = to_nexus_structure(assembler.instrument, origin='sample_origin')

Instrument-specific translators are opt-in per conversion, through a registry that extends the default one:

from niess.nexus.bifrost import BIFROST_REGISTRY
structure = to_nexus_structure(instr, origin='sample_origin',
                               registry=BIFROST_REGISTRY)

Modules:

  • bifrost

    BIFROST-specific NeXus translators.

  • cli

    Command-line conversion of an instrument to NeXus Structure JSON.

  • expression

    Decide whether a component parameter is a literal or a runtime link.

  • instrument

    Walk an assembled instrument and emit its NeXus Structure JSON.

  • nodes

    NeXus Structure JSON node constructors.

  • off

    NXoff_geometry construction.

  • orientation

    Turn mccode-antlr's orientation algebra into a NeXus transformation chain.

  • registry

    Translator registry for the NeXus target.

  • streams

    Filewriter stream and link module directives.

  • translators

    Default per-component-type NeXus translators.

  • variables

    Recover instrument-scope variables for constant folding.

Classes:

  • NexusContext

    Instrument-level state shared by every translator.

  • Translation

    Everything a translator needs about one component instance.

  • NXoff

    Object File Format geometry: a vertex list and polygonal faces.

  • NiessNexusRegistry

    Three-tier translator lookup: niess source type, niess role, McCode type.

Functions:

  • component_body

    What a translator returns: the class and contents of one component's group.

  • to_nexus_structure

    Convert an assembled instrument into ESS NeXus Structure JSON.

  • load_instr

    Load an Instr from a McCode .instr file or a serialized instrument.

  • node_name

    The name of a group, dataset, or named stream node.

  • stream

    A filewriter module directive -- ev44, da00, f144, link, ...

  • resolve_stream

    The stream group for a component, or None if it publishes nothing.

NexusContext dataclass

NexusContext(instr: Any, nxlog_root: str = DEFAULT_NXLOG_ROOT, origin_name: str | None = None, registry: Any = None, declared: dict = dict(), orientations: dict = dict(), origin: Any = None, nodes: dict = dict(), suppressed: set = set(), graph: Any = None)

Instrument-level state shared by every translator.

Methods:

  • literal

    Reduce a value -- or an iterable of them -- to plain JSON-able data.

  • resolve_target

    The absolute path a relative placement should depend on.

  • frame_rotation

    The turn a component's emitted frame carries that the object's does not.

  • frame_offset

    The displacement a component's emitted origin carries that the object's does not.

literal

literal(value)

Reduce a value -- or an iterable of them -- to plain JSON-able data.

Used for attribute values such as transformation vectors, which have no node of their own to carry a link and so must fold to constants.

Source code in src/niess/nexus/instrument.py
def literal(self, value):
    """Reduce a value -- or an iterable of them -- to plain JSON-able data.

    Used for attribute values such as transformation vectors, which have no
    node of their own to carry a link and so must fold to constants.
    """
    from mccode_antlr.common import Expr

    if isinstance(value, Expr):
        resolved = self.resolve(value)
        return expression.literal_value(resolved, str(value))
    if isinstance(value, str):
        return value
    if hasattr(value, '__iter__'):
        return [self.literal(item) for item in value]
    return value

resolve_target

resolve_target(rel) -> str | None

The absolute path a relative placement should depend on.

Source code in src/niess/nexus/instrument.py
def resolve_target(self, rel) -> str | None:
    """The absolute path a relative placement should depend on."""
    node = self.nodes.get(rel.name)
    if node is None:
        raise RuntimeError(f'transformations for {rel.name} defined out of order')
    # The path uses the name the group was *written* under, which a translator may
    # have overridden; self.nodes stays keyed by the McStas instance name, because
    # that is what a placement refers to.
    target = f'{INSTRUMENT_PATH}/{node_name(node)}' 
    if rel.name in self.suppressed:
        # The chain still resolves, but nothing will be written at that path
        logger.warning(
            f'{rel.name} is placed relative to a suppressed component; the emitted '
            'depends_on path will not exist in the written file'
        )

    depends_on = None
    for child in node.get('children') or []:
        if node_name(child) == 'depends_on' and child.get('module') == 'dataset':
            depends_on = child['config'].get('values')
            break
    if depends_on is None:
        return None
    if depends_on.startswith('/'):
        return depends_on
    return None if depends_on == '.' else f'{target}/{depends_on}'

frame_rotation

frame_rotation(instance)

The turn a component's emitted frame carries that the object's does not.

None when there is none, which is everything but a disc chopper today.

Source code in src/niess/nexus/instrument.py
def frame_rotation(self, instance):
    """The turn a component's emitted frame carries that the object's does not.

    ``None`` when there is none, which is everything but a disc chopper today.
    """
    provenance = NiessProvenance.from_instance(instance)
    if provenance is None:
        return None
    rotvec = provenance.extra.get('mccode_frame_rotation')
    if not rotvec or not any(abs(float(a)) > 0 for a in rotvec):
        return None
    return [float(a) for a in rotvec]

frame_offset

frame_offset(instance)

The displacement a component's emitted origin carries that the object's does not.

None when there is none, which is everything but a disc chopper today.

Source code in src/niess/nexus/instrument.py
def frame_offset(self, instance):
    """The displacement a component's emitted origin carries that the object's does not.

    ``None`` when there is none, which is everything but a disc chopper today.
    """
    provenance = NiessProvenance.from_instance(instance)
    if provenance is None:
        return None
    offset = provenance.extra.get('mccode_frame_offset')
    if not offset or not any(abs(float(v)) > 0 for v in offset):
        return None
    return [float(v) for v in offset]

Translation dataclass

Translation(context: 'NexusContext', instance: Any, index: int, provenance: NiessProvenance | None = None)

Everything a translator needs about one component instance.

Methods:

  • resolve

    Resolve a named instance parameter to a literal or a link description.

  • parameter

    The literal value of a parameter, or default if it is not constant.

  • parameter_node

    A node for a parameter: dataset when constant, link group when not.

  • siblings_in_group

    Instances sharing this one's nexus_group_id provenance tag, in order.

Attributes:

  • instr

    The whole instrument, for translators that must inspect sibling instances.

instr property

instr

The whole instrument, for translators that must inspect sibling instances.

resolve

resolve(name: str, default=None)

Resolve a named instance parameter to a literal or a link description.

Source code in src/niess/nexus/instrument.py
def resolve(self, name: str, default=None):
    """Resolve a named instance parameter to a literal or a link description."""
    parameter = self.instance.get_parameter(name)
    if parameter is None:
        logger.debug(f'{self.type_name} does not define the parameter {name}')
        return expression.Literal(default)
    return self.context.resolve(parameter.value)

parameter

parameter(name: str, default=None, dtype=None)

The literal value of a parameter, or default if it is not constant.

Source code in src/niess/nexus/instrument.py
def parameter(self, name: str, default=None, dtype=None):
    """The literal value of a parameter, or ``default`` if it is not constant."""
    resolved = self.resolve(name, default=default)
    value = expression.literal_value(resolved, default)
    if dtype is not None and value is not None:
        try:
            return dtype(value)
        except (TypeError, ValueError):
            return default
    return value

parameter_node

parameter_node(name: str, source: str | None = None, dtype=None, attrs: dict | None = None)

A node for a parameter: dataset when constant, link group when not.

dtype coerces a constant value only -- a runtime-linked parameter has no value here to coerce.

Source code in src/niess/nexus/instrument.py
def parameter_node(
        self,
        name: str,
        source: str | None = None,
        dtype=None,
        attrs: dict | None = None,
):
    """A node for a parameter: dataset when constant, link group when not.

    ``dtype`` coerces a constant value only -- a runtime-linked parameter has
    no value here to coerce.
    """
    resolved = self.resolve(source or name)
    if dtype is not None and isinstance(resolved, expression.Literal):
        try:
            resolved = expression.Literal(dtype(resolved.value))
        except (TypeError, ValueError):
            pass
    return expression.parameter_node(name, resolved, attrs=attrs)

siblings_in_group

siblings_in_group() -> list

Instances sharing this one's nexus_group_id provenance tag, in order.

Source code in src/niess/nexus/instrument.py
def siblings_in_group(self) -> list:
    """Instances sharing this one's ``nexus_group_id`` provenance tag, in order."""
    if self.provenance is None:
        return []
    group_id = self.provenance.extra.get('nexus_group_id')
    if group_id is None:
        return []

    found = []
    for instance in self.instr.components:
        other = NiessProvenance.from_instance(instance)
        if other is not None and other.extra.get('nexus_group_id') == group_id:
            found.append((other.extra.get('nexus_group_index', 0), instance))
    return [instance for _, instance in sorted(found, key=lambda pair: pair[0])]

NXoff

NXoff(vertices, faces)

Object File Format geometry: a vertex list and polygonal faces.

Methods:

  • from_wedge

    A trapezoidal prism, origin at the centre of the entry face, +z downbeam.

Source code in src/niess/nexus/off.py
def __init__(self, vertices, faces):
    self.vertices = vertices
    self.faces = faces

from_wedge classmethod

from_wedge(l, w1, h1, w2=None, h2=None)

A trapezoidal prism, origin at the centre of the entry face, +z downbeam.

Source code in src/niess/nexus/off.py
@classmethod
def from_wedge(cls, l, w1, h1, w2=None, h2=None):
    """A trapezoidal prism, origin at the centre of the entry face, +z downbeam."""
    if w2 is None:
        w2 = w1
    if h2 is None:
        h2 = h1
    x1, y1, x2, y2 = (float(v) / 2 for v in (w1, h1, w2, h2))
    vertices = [
        [-x1, -y1, 0], [-x1, y1, 0], [x1, y1, 0], [x1, -y1, 0],
        [-x2, -y2, l], [-x2, y2, l], [x2, y2, l], [x2, -y2, l],
    ]
    # Clockwise winding, facing out
    faces = [
        [0, 1, 2, 3], [1, 5, 6, 2], [5, 4, 7, 6],
        [6, 7, 3, 2], [7, 4, 0, 3], [1, 0, 4, 5],
    ]
    return cls(vertices, faces)

NiessNexusRegistry

NiessNexusRegistry(parent: 'NiessRegistry[B] | None' = None)

Bases: NiessRegistry[NexusTranslator]

Three-tier translator lookup: niess source type, niess role, McCode type.

Source code in src/niess/dispatch.py
def __init__(self, parent: 'NiessRegistry[B] | None' = None) -> None:
    self.parent = parent
    self._source_type_builders: dict[str, B] = {}
    self._role_builders: dict[str, B] = {}
    self._component_type_builders: dict[str, B] = {}

component_body

component_body(nx_class: str, children: list | None = None, attrs: dict | None = None, name: str | None = None) -> dict

What a translator returns: the class and contents of one component's group.

name overrides the group's name, which defaults to the McStas instance's. Use it where the instance name is an artefact of how the instrument was built rather than something a reader of the file should see -- a composite emitted as several instances, say, whose NeXus group should carry the name of the thing itself.

Source code in src/niess/nexus/instrument.py
def component_body(
        nx_class: str,
        children: list | None = None,
        attrs: dict | None = None,
        name: str | None = None,
) -> dict:
    """What a translator returns: the class and contents of one component's group.

    ``name`` overrides the group's name, which defaults to the McStas instance's. Use
    it where the instance name is an artefact of how the instrument was built rather
    than something a reader of the file should see -- a composite emitted as several
    instances, say, whose NeXus group should carry the name of the thing itself.
    """
    return {
        'nx_class': nx_class,
        'children': list(children or []),
        'attrs': dict(attrs or {}),
        'name': name,
    }

to_nexus_structure

to_nexus_structure(instr, origin: str | None = None, nxlog_root: str | None = None, absolute_depends_on: bool = False, registry=None, graph=None) -> dict

Convert an assembled instrument into ESS NeXus Structure JSON.

Parameters:

  • instr

    The mccode_antlr Instr an Assembler produced.

  • origin (str | None, default: None ) –

    Name of the component to treat as the coordinate origin. Defaults to the instrument's sample-category component.

  • nxlog_root (str | None, default: None ) –

    Where runtime parameter values are published, for link directives.

  • absolute_depends_on (bool, default: False ) –

    Rewrite relative depends_on values as absolute NeXus paths.

  • registry

    Translator registry; defaults to :data:DEFAULT_NEXUS_REGISTRY, which holds only the generic per-component-type translators. Pass an instrument-specific registry -- niess.nexus.bifrost.BIFROST_REGISTRY, say -- to add its translators to this conversion alone.

  • graph

    A networkx DiGraph representing the possible particle path(s) through the instrument. A standard linear path will be constructed for @inputs and @outputs group attributes if this is not provided.

Source code in src/niess/nexus/instrument.py
def to_nexus_structure(
        instr,
        origin: str | None = None,
        nxlog_root: str | None = None,
        absolute_depends_on: bool = False,
        registry=None,
        graph=None,
) -> dict:
    """Convert an assembled instrument into ESS NeXus Structure JSON.

    Parameters
    ----------
    instr:
        The ``mccode_antlr`` ``Instr`` an ``Assembler`` produced.
    origin:
        Name of the component to treat as the coordinate origin. Defaults to the
        instrument's sample-category component.
    nxlog_root:
        Where runtime parameter values are published, for link directives.
    absolute_depends_on:
        Rewrite relative ``depends_on`` values as absolute NeXus paths.
    registry:
        Translator registry; defaults to :data:`DEFAULT_NEXUS_REGISTRY`, which holds
        only the generic per-component-type translators. Pass an instrument-specific
        registry -- ``niess.nexus.bifrost.BIFROST_REGISTRY``, say -- to add its
        translators to this conversion alone.
    graph:
        A networkx DiGraph representing the possible particle path(s) through the
        instrument. A standard linear path will be constructed for @inputs and @outputs
        group attributes if this is not provided.
    """
    context = NexusContext(
        instr,
        nxlog_root=nxlog_root or DEFAULT_NXLOG_ROOT,
        origin_name=origin,
        registry=registry,
        graph=graph,
    )
    instrument = group('instrument', 'NXinstrument', children=instrument_children(context))
    entry = group('entry', 'NXentry', children=[instrument])

    if absolute_depends_on:
        absolutize_depends_on(entry, '')

    return {'children': [entry]}

load_instr

load_instr(filepath: str | Path)

Load an Instr from a McCode .instr file or a serialized instrument.

Source code in src/niess/nexus/cli.py
def load_instr(filepath: str | Path):
    """Load an ``Instr`` from a McCode ``.instr`` file or a serialized instrument."""
    filepath = Path(filepath)
    if not filepath.is_file():
        raise ValueError(f'{filepath} does not exist or is not a file')

    suffix = filepath.suffix.lower()
    if suffix == '.instr':
        from mccode_antlr.loader import load_mcstas_instr
        return load_mcstas_instr(filepath)
    if suffix == '.json':
        from mccode_antlr.io.json import load_json
        return load_json(filepath)
    if suffix in ('.msgpack', '.mpk'):
        from mccode_antlr.io.msgpack import load_msgpack
        return load_msgpack(filepath)

    raise ValueError(f'Cannot load an instrument from {filepath.suffix} files')

node_name

node_name(node) -> str | None

The name of a group, dataset, or named stream node.

Source code in src/niess/nexus/nodes.py
def node_name(node) -> str | None:
    """The name of a group, dataset, or named stream node."""
    if not isinstance(node, dict):
        return None
    if 'name' in node:
        return node['name']
    return (node.get('config') or {}).get('name')

stream

stream(module: str, config: dict, attrs: dict[str, Any] | None = None) -> dict

A filewriter module directive -- ev44, da00, f144, link, ...

Source code in src/niess/nexus/nodes.py
def stream(module: str, config: dict, attrs: dict[str, Any] | None = None) -> dict:
    """A filewriter module directive -- ``ev44``, ``da00``, ``f144``, ``link``, ..."""
    node = {'module': module, 'config': config}
    attributes = _attributes(None, attrs)
    if attributes:
        node['attributes'] = attributes
    return node

resolve_stream

resolve_stream(translation, default: dict | None = None, name: str = 'data') -> dict | None

The stream group for a component, or None if it publishes nothing.

The protocol is never chosen here. Some monitors belong on da00 histograms and some on ev44 events; which one is a property of the instrument setup, so it is read from the instrument in priority order:

  1. a METADATA "nexus_structure_stream_data" block on the component -- the escape hatch for instruments authored outside niess, emitted verbatim;
  2. a nexus_stream entry in the component's niess provenance extra, which is how a niess component records the choice made when the instrument was built;
  3. default -- the component type's established behaviour, used only when the instrument expressed no preference at all.

A component with no selection and no default gets no stream group rather than a guessed one.

Source code in src/niess/nexus/streams.py
def resolve_stream(translation, default: dict | None = None, name: str = 'data') -> dict | None:
    """The stream group for a component, or ``None`` if it publishes nothing.

    The protocol is never chosen here. Some monitors belong on ``da00`` histograms
    and some on ``ev44`` events; which one is a property of the instrument setup, so
    it is read from the instrument in priority order:

    1. a ``METADATA "nexus_structure_stream_data"`` block on the component -- the
       escape hatch for instruments authored outside niess, emitted verbatim;
    2. a ``nexus_stream`` entry in the component's niess provenance ``extra``,
       which is how a niess component records the choice made when the instrument
       was built;
    3. ``default`` -- the component type's established behaviour, used only when
       the instrument expressed no preference at all.

    A component with no selection and no default gets no stream group rather than a
    guessed one.
    """
    from json import JSONDecodeError, loads

    for metadata in translation.instance.metadata:
        if metadata.mimetype != 'application/json':
            continue
        if metadata.name != 'nexus_structure_stream_data':
            continue
        try:
            return stream_group_from_config(name, loads(metadata.value))
        except JSONDecodeError:
            continue

    selection = None
    if translation.provenance is not None:
        selection = translation.provenance.extra.get('nexus_stream')

    if selection is None:
        selection = default
    if selection is None:
        return None

    return stream_group_from_selection(name, selection)