niess.tof¶

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 chopper_specs(instrument, values: dict | None = None, origin: float = 0.0,
                  skip=(), path_lengths=None) -> tuple[ChopperSpec, ...]:
    """Every disc in ``instrument``, as ``tof`` wants it.

    ``values`` overrides a knob by name -- ``{'chopper_rotation_speed': 20}`` -- which is what
    running the same instrument at a different speed means. The disc names its knobs, so
    the override needs no convention repeated here. A value that carries its own unit, a
    scipp scalar, is converted to the one the disc declares the knob in, so a speed worked
    out in kHz or a delay in ms can be handed over as it comes; a name that is not a knob
    of this instrument is refused rather than quietly changing nothing.

    ``origin`` is where the ``tof.Source`` these are paired with sits, and it belongs to
    the model rather than to the instrument: an ESS source in ``tof`` carries its own
    0.05 m facility offset, while the niess moderator is at the instrument origin. Pass
    ``source.distance`` to have the two agree.
    """
    from ..chopcalc.train import train_from_instrument
    from ..components.chopper import DISC_CHOPPERS
    from ..walk import visits

    values = dict(values or {})
    _check_knobs(instrument, values)
    train = train_from_instrument(instrument, skip=skip, path_lengths=path_lengths)
    paths = {entry.name: float(entry.path) for entry in train.choppers}

    specs = []
    for visit in visits(instrument):
        disc = visit.obj
        if not isinstance(disc, DISC_CHOPPERS) or visit.name not in paths:
            continue
        speed = _setting(disc, 'speed', values)
        # ESS TotDly, formed as `Chopper.delay_seconds` forms it from the three knobs,
        # which are in nanoseconds and degrees as ESS publishes them; tof works in s.
        experiment, mechanical, pulse = (_setting(disc, which, values)
                                         for which in DELAYS)
        delay = 1e-9 * (pulse + experiment)
        if speed:
            delay += mechanical / (360.0 * abs(speed))
        beam = float(disc.beam_angle.to(unit='deg').value)
        windows = tuple((beam - closing, beam - opening)
                        for opening, closing in disc.slits())
        specs.append(spec_from_windows(
            name=visit.name, windows=windows, delay=float(delay), speed=float(speed),
            distance=origin + paths[visit.name]))
    return tuple(sorted(specs, key=lambda spec: spec.distance))

def to_tof_model(instrument, *, source=None, values: dict | None = None,
                 neutrons: int = 1_000_000, pulses: int | None = None,
                 seed: int | None = None, sample: str | None = None, skip=(),
                 path_lengths=None):
    """A ready-to-run ``tof.Model`` for ``instrument``, and what went into it.

    The instrument-reading version of this asks `niess.chopcalc` for a chopper train,
    which is C text, and parses the numbers back out. Here every number is on an object:
    a disc's speed and delay, a monitor's position, the source's wavelength band.

    ``values`` overrides a knob by name, which is what running the same instrument at a
    different chopper speed means -- a chopper's speed or delay, or one of the source's
    wavelength bounds. A scipp scalar is converted to the unit that knob is declared in,
    and a name that is not a knob is refused. See `chopper_specs`.

    ``source`` is a ``tof.Source`` to use instead of building one. Building one fetches
    the facility's pulse profile, so passing your own is how this runs offline -- which
    is what the test suite does, and the reason this is an argument rather than
    something read off the instrument.

    ``sample`` names the component to put the last detector on. The instrument's own
    ``origin`` is used when it is not given, which is right for an instrument that
    declares one.
    """
    import scipp as sc

    from ..components.chopper import DISC_CHOPPERS
    from ..components.monitors import FrameMonitor
    from ..components.moderator import Moderator
    from ..walk import visits
    from .setup import _tof, _facility_for
    from .parameters import Use

    tof = _tof()
    values = dict(values or {})
    _check_knobs(instrument, values)
    seen = list(visits(instrument))

    source_visit = next((v for v in seen if isinstance(v.obj, Moderator)), None)
    if source_visit is None:
        raise ValueError('the instrument has no moderator, so there is nothing to model')

    band, band_used = {}, []
    for field, keyword in (('wavelength_minimum', 'wmin'),
                           ('wavelength_maximum', 'wmax')):
        knob = _band_knob(getattr(source_visit.obj, field, None))
        if knob is None:
            continue
        name, default, unit = knob
        held = as_declared(name, values[name], unit) if name in values else default
        if held is None:
            continue
        # the bound is declared in whatever unit the calibration chose; tof wants angstrom
        band[keyword] = sc.scalar(float(held), unit=unit).to(unit='angstrom')
        band_used.append(Use(name=name, value=float(held), default=default, unit=unit,
                             overridden=name in values,
                             used_by=(f'{source_visit.name}.{field}',)))

    if source is None:
        facility = _facility_for(instrument, tof)
        extra = {}
        if pulses is not None:
            extra['pulses'] = int(pulses)
        if seed is not None:
            extra['seed'] = int(seed)
        source = tof.Source(facility=facility, neutrons=int(neutrons), **band, **extra)
    else:
        # a source handed over is used as it is: it already carries its own pulses,
        # neutrons and band, and quietly rebuilding it would discard them
        band_used = []
    origin = float(source.distance.to(unit='m').value)

    specs = chopper_specs(instrument, values=values, origin=origin, skip=skip,
                          path_lengths=path_lengths)
    by_name = {spec.name: spec for spec in specs}

    train = _paths(instrument, skip=skip, path_lengths=path_lengths)
    components, detectors = [], []
    for visit in seen:
        name = visit.name
        if isinstance(visit.obj, DISC_CHOPPERS) and name in by_name:
            components.append(_as_tof_chopper(tof, by_name[name]))
        elif isinstance(visit.obj, (FrameMonitor,)) and name in train:
            components.append(tof.Detector(distance=sc.scalar(origin + train[name],
                                                              unit='m'), name=name))
            detectors.append(name)

    sample = getattr(instrument, 'origin', None) if sample is None else sample
    if sample is not None and sample in train and sample not in detectors:
        components.append(tof.Detector(distance=sc.scalar(origin + train[sample],
                                                          unit='m'), name=sample))
        detectors.append(sample)

    used = list(band_used)
    for visit in seen:
        if not isinstance(visit.obj, DISC_CHOPPERS) or visit.name not in by_name:
            continue
        for which in KNOBS:
            knob, default, unit = _knob(visit.obj, which)
            used.append(Use(name=knob, value=_setting(visit.obj, which, values),
                            default=default, unit=unit,
                            overridden=knob in values,
                            used_by=(f'{visit.name}.{which}',)))

    def rebuild(overrides):
        return to_tof_model(instrument, source=source, values={**values, **overrides},
                            neutrons=neutrons, pulses=pulses, seed=seed, sample=sample,
                            skip=skip, path_lengths=path_lengths)

    from .setup import TofSetup
    return TofSetup(model=tof.Model(source=source, components=components),
                    source=source, choppers=tuple(specs),
                    detectors=tuple(detectors), parameters=tuple(used),
                    _rebuild=rebuild)

def as_declared(name: str, value, unit: str | None) -> float:
    """A supplied value as a plain number, in the unit whatever reads it declares.

    A chopper speed calculated elsewhere arrives as a scipp scalar carrying its own unit
    -- and a calculator is as entitled to hand back a period in milliseconds or a speed in
    rpm as it is to match what the instrument happens to say. The thing being set knows
    what unit it wants, so convert rather than making the caller strip the unit and hope
    the two agree.

    ``unit`` of ``None`` means nothing declared one, so there is nothing to convert to and
    the number is taken as it comes. Shared with the tree-reading route, which knows the
    unit from the disc rather than from a DEFINE line, so that a value carrying its unit
    is accepted the same way whichever way the instrument was read.
    """
    if not hasattr(value, 'unit'):
        return float(value)
    if unit is None:
        return float(value.value)
    try:
        return float(value.to(unit=unit, dtype='float64').value)
    except Exception as error:
        raise ValueError(
            f'{name} is declared in {unit!r} and {value.unit} does not convert to it'
        ) from error

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}

def as_declared(name: str, value, unit: str | None) -> float:
    """A supplied value as a plain number, in the unit whatever reads it declares.

    A chopper speed calculated elsewhere arrives as a scipp scalar carrying its own unit
    -- and a calculator is as entitled to hand back a period in milliseconds or a speed in
    rpm as it is to match what the instrument happens to say. The thing being set knows
    what unit it wants, so convert rather than making the caller strip the unit and hope
    the two agree.

    ``unit`` of ``None`` means nothing declared one, so there is nothing to convert to and
    the number is taken as it comes. Shared with the tree-reading route, which knows the
    unit from the disc rather than from a DEFINE line, so that a value carrying its unit
    is accepted the same way whichever way the instrument was read.
    """
    if not hasattr(value, 'unit'):
        return float(value)
    if unit is None:
        return float(value.value)
    try:
        return float(value.to(unit=unit, dtype='float64').value)
    except Exception as error:
        raise ValueError(
            f'{name} is declared in {unit!r} and {value.unit} does not convert to it'
        ) from error