niess.tof¶

def to_tof_model(obj, *, source=None, values=None, neutrons: int = 1_000_000,
                 pulses: int | None = None, seed: int | None = None,
                 sample: str | None = None, source_name: str | None = None,
                 skip=(), path_lengths=None, graph=None, registry=None) -> TofSetup:
    """Build a ready-to-run ``tof.Model`` from an assembled instrument.

    Parameters
    ----------
    obj:
        An ``Assembler`` or an ``Instr``. A **top-level** assembler: a child from
        ``assembler.included(...)`` merges into its parent only when the block exits, so
        its components -- and every later section's -- are not visible yet.
    source:
        A ``tof.Source`` to use instead of building one. Building one downloads the
        facility's pulse profile on first use, so pass your own to stay offline.
    values:
        Instrument parameter values to use instead of the instrument's own defaults.
        A scipp scalar is converted to whatever unit the instrument declares, so a speed
        worked out in kHz or a delay in ms can be handed over as it comes.
    neutrons:
        How many to sample from each pulse.
    pulses:
        How many source pulses to simulate. More than one is what shows a chopper turning
        at a fraction of the source frequency doing its job: a disc at half of 14 Hz opens
        for every other pulse and absorbs the rest, which a single pulse cannot show.
        Taken from the source when omitted.
    seed:
        Fixes the sampling, so two runs can be compared rather than merely resembling
        each other.
    sample:
        The component to put a detector on at the end of the beam. Found from the beam
        path when omitted.
    graph:
        The particle flow through the instrument, as a ``networkx`` DiGraph. Every
        distance here is walked along it, and McCode has no way to say that a beam
        branches -- so an instrument whose flow is not the order its components are
        declared in, BIFROST after the sample among them, has to be handed the real one.
        Built from the instrument when omitted.
    """
    from ..chopcalc.discovery import build_train

    tof = _tof()
    import scipp as sc

    instrument = obj.instrument if hasattr(obj, 'instrument') else obj
    if getattr(obj, 'parent', None) is not None:
        raise ValueError(
            'to_tof_model needs the top-level Assembler, after every section has been '
            'added. A section\'s child Assembler is merged into its parent only on '
            'leaving the included() block, so its components are not visible yet.'
        )

    parameters = ParameterValues(instrument, values)
    graph = instrument.build_flow_graph() if graph is None else graph
    places = positions(instrument)
    source_instance = find_source(instrument, graph, source_name)

    notes = []
    if source is None:
        source, facility = _build_source(instrument, source_instance, parameters, tof,
                                         neutrons=neutrons, pulses=pulses, seed=seed)
        notes.append(f'source pulse from the {facility!r} profile')
    else:
        notes.append('source supplied by the caller')
    origin = float(source.distance.to(unit='m').value)

    excluded = ()
    specs: dict[str, ChopperSpec] = {}
    try:
        train = build_train(instrument, source=source_name, skip=skip,
                            path_lengths=path_lengths, graph=graph)
    except ChopcalcError as error:
        notes.append(f'no chopper train: {error}')
    else:
        excluded = train.excluded
        for entry in train.choppers:
            speed = parameters.evaluate_text(entry.speed, used_by=f'{entry.name}.speed')
            delay = parameters.evaluate_text(entry.delay, used_by=f'{entry.name}.delay')
            windows = [(parameters.evaluate_text(low), parameters.evaluate_text(high))
                       for low, high in entry.windows]
            path = parameters.evaluate_text(entry.path)
            if None in (speed, delay, path) or any(v is None for w in windows for v in w):
                notes.append(f'{entry.name}: left out, its description did not reduce to '
                             f'numbers')
                continue
            specs[entry.name] = spec_from_windows(
                name=entry.name, windows=windows, delay=delay, speed=speed,
                distance=origin + path)

    registry = DEFAULT_TOF_REGISTRY if registry is None else registry
    components, detector_names = [], []
    for instance in instrument.components:
        if instance.name == source_instance.name:
            continue
        builder = registry.resolve_builder(instance)
        if builder is None:
            continue
        provenance = NiessProvenance.from_instance(instance)
        spec = specs.get(instance.name)
        if spec is None and provenance is not None:
            spec = specs.get(provenance.extra.get('nexus_group_id'))
        try:
            distance = origin + beam_path_length(graph, places, source_instance.name,
                                                 instance.name)
        except ChopcalcError as error:
            notes.append(f'{instance.name}: left out, {error}')
            continue
        built = builder(Conversion(instrument=instrument, instance=instance,
                                   values=parameters, distance=distance,
                                   provenance=provenance, spec=spec))
        if built is None:
            continue
        components.append(built)
        if getattr(built, 'kind', None) == 'detector':
            detector_names.append(built.name)

    sample_at = _furthest_measurable(graph, places, source_instance.name,
                                     instrument, sample)
    if sample_at is not None and sample_at not in detector_names:
        try:
            distance = origin + beam_path_length(graph, places, source_instance.name,
                                                 sample_at)
        except ChopcalcError as error:
            notes.append(f'{sample_at}: no sample detector, {error}')
        else:
            components.append(tof.Detector(distance=sc.scalar(distance, unit='m'),
                                           name=sample_at))
            detector_names.append(sample_at)

    model = tof.Model(source=source, components=components)

    def rebuild(overrides):
        merged = {**(values or {}), **overrides}
        return to_tof_model(obj, source=None, values=merged, neutrons=neutrons,
                            pulses=pulses, seed=seed, sample=sample,
                            source_name=source_name, skip=skip,
                            path_lengths=path_lengths, graph=graph, registry=registry)

    return TofSetup(
        model=model, source=source,
        choppers=tuple(specs[name] for name in sorted(specs, key=lambda n: specs[n].distance)),
        detectors=tuple(detector_names),
        parameters=parameters.uses(),
        excluded=tuple(excluded),
        notes=tuple(notes),
        _rebuild=rebuild,
    )

def delay_to_phase(delay: float, speed: float) -> float:
    """A disc's delay in seconds, as the phase angle in degrees ``tof`` wants.

    ``tof`` has no notion of a delay. An opening is placed by its angle and the whole disc
    is shifted by ``phase``, which ``open_close_times`` adds to every angle before dividing
    by the angular speed -- so a delay of ``d`` seconds is ``360 * |speed| * d`` degrees.

    **The sign does not flip with the direction of rotation**, which is the one thing here
    worth being suspicious of. A NeXus phase is an angle in the disc's own rotating frame,
    so its sign *does* flip: ``tof.Chopper.from_nexus`` writes ``phase = -phase`` for a
    negative rotation speed for exactly that reason. A niess delay is a *time*, and a later
    time is later whichever way the disc turns.

    Checked against ``Chopper.open_close_times`` for both directions on an asymmetric
    three-opening disc: negating this puts every opening somewhere else.
    """
    return 360.0 * abs(speed) * delay

def spec_from_windows(*, name: str, windows, delay: float, speed: float,
                      distance: float) -> ChopperSpec:
    """A spec from the windows ``niess.chopcalc`` extracts.

    Those windows are already in the frame where an edge at angle ``a`` is on the beam at
    ``delay + a / (360 * speed)`` -- that is, ``a`` is measured *from the beam towards the
    mark*, which is the opposite sense to ``tof``, where an opening at angle ``a`` is
    reached after turning through it. So the two swap sign, and a ``(minimum, maximum)``
    pair becomes ``(-maximum, -minimum)``: still increasing, which ``tof`` requires.
    """
    if speed == 0:
        raise ValueError(f'{name}: a chopper that is not turning has no phase')
    pairs = [(-float(high), -float(low)) for low, high in windows]
    return ChopperSpec(
        name=name,
        frequency=abs(float(speed)),
        anticlockwise=float(speed) > 0,
        open=tuple(low for low, _ in pairs),
        close=tuple(high for _, high in pairs),
        phase=delay_to_phase(float(delay), float(speed)),
        distance=float(distance),
    )

def delay_to_phase(delay: float, speed: float) -> float:
    """A disc's delay in seconds, as the phase angle in degrees ``tof`` wants.

    ``tof`` has no notion of a delay. An opening is placed by its angle and the whole disc
    is shifted by ``phase``, which ``open_close_times`` adds to every angle before dividing
    by the angular speed -- so a delay of ``d`` seconds is ``360 * |speed| * d`` degrees.

    **The sign does not flip with the direction of rotation**, which is the one thing here
    worth being suspicious of. A NeXus phase is an angle in the disc's own rotating frame,
    so its sign *does* flip: ``tof.Chopper.from_nexus`` writes ``phase = -phase`` for a
    negative rotation speed for exactly that reason. A niess delay is a *time*, and a later
    time is later whichever way the disc turns.

    Checked against ``Chopper.open_close_times`` for both directions on an asymmetric
    three-opening disc: negating this puts every opening somewhere else.
    """
    return 360.0 * abs(speed) * delay

def spec_from_windows(*, name: str, windows, delay: float, speed: float,
                      distance: float) -> ChopperSpec:
    """A spec from the windows ``niess.chopcalc`` extracts.

    Those windows are already in the frame where an edge at angle ``a`` is on the beam at
    ``delay + a / (360 * speed)`` -- that is, ``a`` is measured *from the beam towards the
    mark*, which is the opposite sense to ``tof``, where an opening at angle ``a`` is
    reached after turning through it. So the two swap sign, and a ``(minimum, maximum)``
    pair becomes ``(-maximum, -minimum)``: still increasing, which ``tof`` requires.
    """
    if speed == 0:
        raise ValueError(f'{name}: a chopper that is not turning has no phase')
    pairs = [(-float(high), -float(low)) for low, high in windows]
    return ChopperSpec(
        name=name,
        frequency=abs(float(speed)),
        anticlockwise=float(speed) > 0,
        open=tuple(low for low, _ in pairs),
        close=tuple(high for _, high in pairs),
        phase=delay_to_phase(float(delay), float(speed)),
        distance=float(distance),
    )

def instrument_defaults(instrument) -> dict[str, object]:
    """Every DEFINE INSTRUMENT parameter that has a value, by name.

    niess writes a chopper's speed and delay with the calibration's own numbers as the
    defaults, so an instrument it built is usually complete on its own -- which is why the
    report below can normally say that nothing needs supplying.
    """
    known = {}
    for parameter in instrument.parameters:
        value = parameter.value
        if value is None or not getattr(value, 'has_value', False):
            continue
        known[parameter.name] = value
    return known

def instrument_units(instrument) -> dict[str, str | None]:
    """Parameter units, unquoted -- a DEFINE line carries them as `name/"Hz" = 14`."""
    return {p.name: (p.unit or '').strip().strip('"') or None
            for p in instrument.parameters}