<filename>pcdsdevices/tests/test_ccm.py
import logging
import time

import numpy as np
import pytest
from ophyd.sim import fake_device_cache, make_fake_device

from .. import ccm
from ..sim import FastMotor

logger = logging.getLogger(__name__)


SAMPLE_ALIO = 4.575  # Current value as of writing this file
SAMPLE_THETA = 1.2  # Modest angle
SAMPLE_WAVELENGTH = 1.5  # hard xray


# Make sure the calcs are properly inverted
def test_theta_alio_inversion():
    logger.debug('test_theta_alio_inversion')
    theta = ccm.alio_to_theta(SAMPLE_ALIO, ccm.default_theta0, ccm.default_gr,
                              ccm.default_gd)
    alio_calc = ccm.theta_to_alio(theta, ccm.default_theta0, ccm.default_gr,
                                  ccm.default_gd)
    # Unlike the other inversions, this is just an approximation
    assert np.isclose(alio_calc, SAMPLE_ALIO)


def test_wavelength_theta_inversion():
    logger.debug('test_wavelength_theta_inversion')
    wavelength = ccm.theta_to_wavelength(SAMPLE_THETA, ccm.default_dspacing)
    theta = ccm.wavelength_to_theta(wavelength, ccm.default_dspacing)
    logger.debug('%s, %s', wavelength, theta)
    assert np.isclose(theta, SAMPLE_THETA)
    theta = ccm.wavelength_to_theta(SAMPLE_WAVELENGTH, ccm.default_dspacing)
    wavelength = ccm.theta_to_wavelength(theta, ccm.default_dspacing)
    logger.debug('%s, %s', wavelength, theta)
    assert np.isclose(wavelength, SAMPLE_WAVELENGTH)


def test_energy_wavelength_inversion():
    logger.debug('test_energy_wavelength_inversion')
    energy = ccm.wavelength_to_energy(SAMPLE_WAVELENGTH)
    wavelength_calc = ccm.energy_to_wavelength(energy)
    assert wavelength_calc == SAMPLE_WAVELENGTH


@pytest.fixture(scope='function')
def fake_ccm():
    return make_fake_ccm()


class FakeAlio(FastMotor):
    kill = None
    home = None


def make_fake_ccm():
    fake_device_cache[ccm.CCMMotor] = FastMotor
    fake_device_cache[ccm.CCMAlio] = FakeAlio
    FakeCCM = make_fake_device(ccm.CCM)
    fake_ccm = FakeCCM(alio_prefix='ALIO', theta2fine_prefix='THETA',
                       theta2coarse_prefix='THTA', chi2_prefix='CHI',
                       x_down_prefix='X:DOWN', x_up_prefix='X:UP',
                       y_down_prefix='Y:DOWN', y_up_north_prefix='Y:UP:NORTH',
                       y_up_south_prefix='Y:UP:SOUTH', in_pos=8, out_pos=0,
                       name='fake_ccm')

    def init_pos(mot, pos=0):
        mot.user_readback.sim_put(0)
        mot.user_setpoint.sim_put(0)
        mot.user_setpoint.sim_set_limits((0, 0))
        mot.motor_spg.sim_put(2)
        mot.part_number.sim_put('tasdf')

    init_pos(fake_ccm.x.down)
    init_pos(fake_ccm.x.up)
    init_pos(fake_ccm.y.down)
    init_pos(fake_ccm.y.up_north)
    init_pos(fake_ccm.y.up_south)

    fake_ccm.alio.set(SAMPLE_ALIO)
    fake_ccm.energy.alio.set(SAMPLE_ALIO)
    fake_ccm.energy_with_vernier.alio.set(SAMPLE_ALIO)
    fake_ccm.energy_with_vernier.vernier.setpoint.sim_put(0)

    return fake_ccm


def test_fake_ccm(fake_ccm):
    logger.debug('test_fake_ccm')
    fake_ccm.get()


# Make sure we set up the forward/inverse to use the right methods
def test_ccm_calc(fake_ccm):
    logger.debug('test_ccm_calc')
    calc = fake_ccm.energy

    logger.debug('physics pos is %s', calc.position)
    logger.debug('real pos is %s', calc.real_position)
    logger.debug('sample alio is %s', SAMPLE_ALIO)

    theta_func = ccm.alio_to_theta(
        SAMPLE_ALIO,
        calc.theta0_rad_val,
        calc.gr_val,
        calc.gd_val,
    )
    wavelength_func = ccm.theta_to_wavelength(theta_func, calc.dspacing_val)
    energy_func = ccm.wavelength_to_energy(wavelength_func)
    energy = calc.energy.position
    assert energy == energy_func

    calc.alio.move(0)
    calc.move(energy, wait=False)
    assert np.isclose(calc.alio.position, SAMPLE_ALIO)

    calc.alio.move(calc.alio.position)
    calc.move(energy=calc.energy.position, wait=False)
    assert np.isclose(calc.alio.position, SAMPLE_ALIO)


# Make sure sync'd axes work and that unk/in/out states work
@pytest.mark.timeout(5)
def test_ccm_main(fake_ccm):
    logger.debug('test_ccm_main')
    fake_ccm.y.move(5, wait=False)
    assert fake_ccm.y.down.user_setpoint.get() == 5
    assert fake_ccm.y.up_north.user_setpoint.get() == 5
    assert fake_ccm.y.up_south.user_setpoint.get() == 5

    assert fake_ccm.removed
    assert not fake_ccm.inserted

    fake_ccm.x.down.user_readback.sim_put(8)
    fake_ccm.x.up.user_readback.sim_put(8)
    assert not fake_ccm.removed
    assert fake_ccm.inserted

    fake_ccm.x.down.user_readback.sim_put(4)
    fake_ccm.x.up.user_readback.sim_put(4)
    assert not fake_ccm.removed
    assert not fake_ccm.inserted

    fake_ccm.insert(wait=False)
    assert fake_ccm.x.down.user_setpoint.get() == 8
    assert fake_ccm.x.up.user_setpoint.get() == 8

    fake_ccm.remove(wait=False)
    assert fake_ccm.x.down.user_setpoint.get() == 0
    assert fake_ccm.x.up.user_setpoint.get() == 0


@pytest.mark.timeout(5)
def test_vernier(fake_ccm):
    logger.debug('test_vernier')

    pseudopos = fake_ccm.energy_with_vernier

    # Moving with vernier should move the energy request motor too
    pseudopos.move(7, wait=False)
    assert np.isclose(pseudopos.energy.position, 7)
    assert pseudopos.vernier.position == 7000

    pseudopos.move(8, wait=False)
    assert np.isclose(pseudopos.energy.position, 8)
    assert pseudopos.vernier.position == 8000

    pseudopos.move(9, wait=False)
    assert np.isclose(pseudopos.energy.position, 9)
    assert pseudopos.vernier.position == 9000

    # Small moves (less than 30eV) should be skipped on the energy request
    pseudopos.move(9.001, wait=False)
    assert np.isclose(pseudopos.energy.position, 9.001)
    assert pseudopos.vernier.position == 9000

    # Unless we set the option for not skipping them
    pseudopos.vernier.skip_small_moves = False
    pseudopos.move(9.002, wait=False)
    assert np.isclose(pseudopos.energy.position, 9.002)
    assert pseudopos.vernier.position == 9002


@pytest.mark.timeout(5)
def test_set_current_position(fake_ccm):
    logger.debug('test_set_current_position')
    mot = fake_ccm.energy.energy
    for energy in range(6, 14):
        mot.set_current_position(energy)
        assert np.isclose(mot.position, energy)


@pytest.mark.timeout(5)
def test_check_valid_constant(fake_ccm):
    logger.debug('test_check_valid_constant')

    # First call to make_valid sends the first monitor update
    def make_valid(sig, valid):
        if valid:
            sig.put(1)
        else:
            sig.put(0)

    def make_conn(sig, conn):
        sig._metadata['connected'] = conn

    def output(sig):
        return fake_ccm._check_valid_constant(sig, sig.get())

    test_sig = fake_ccm.dspacing

    # Can we get to all the enum values?
    make_conn(test_sig, False)
    assert output(test_sig) == ccm.CCMConstantWarning.ALWAYS_DISCONNECT
    make_conn(test_sig, True)
    make_valid(test_sig, False)
    assert output(test_sig) == ccm.CCMConstantWarning.INVALID_CONNECT
    make_conn(test_sig, False)
    assert output(test_sig) == ccm.CCMConstantWarning.INVALID_DISCONNECT
    make_conn(test_sig, True)
    make_valid(test_sig, True)
    assert output(test_sig) == ccm.CCMConstantWarning.NO_WARNING
    make_conn(test_sig, False)
    assert output(test_sig) == ccm.CCMConstantWarning.VALID_DISCONNECT

    # theta0_deg is allowed to be zero, unlike the others
    test_sig2 = fake_ccm.theta0_deg
    make_conn(test_sig2, True)
    make_valid(test_sig2, False)
    assert output(test_sig2) == ccm.CCMConstantWarning.NO_WARNING


@pytest.mark.timeout(5)
def test_show_constant_warning(fake_ccm, caplog):
    logger.debug('test_show_constant_warning')
    for warning in (
        ccm.CCMConstantWarning.NO_WARNING,
        ccm.CCMConstantWarning.ALWAYS_DISCONNECT,
        ccm.CCMConstantWarning.VALID_DISCONNECT,
        ccm.CCMConstantWarning.INVALID_DISCONNECT,
        ccm.CCMConstantWarning.INVALID_CONNECT,
    ):
        caplog.clear()
        with caplog.at_level(logging.WARNING):
            fake_ccm._show_constant_warning(
                warning,
                fake_ccm.dspacing,
                0.111111,
                0.222222,
            )
        if warning == ccm.CCMConstantWarning.NO_WARNING:
            assert len(caplog.records) == 0
        else:
            assert len(caplog.records) == 1


@pytest.mark.timeout(5)
def test_warn_invalid_constants(fake_ccm, caplog):
    logger.debug('test_warn_invalid_constants')
    # Trick the warning into thinking we've be initialized for a while
    fake_ccm._init_time = time.monotonic() - 1000
    fake_ccm.theta0_deg.put(0)
    fake_ccm.dspacing.put(0)
    fake_ccm.gr.put(0)
    fake_ccm.gd.put(0)
    # We expect three warnings from the fake PVs that start at 0
    caplog.clear()
    with caplog.at_level(logging.WARNING):
        fake_ccm.warn_invalid_constants(only_new=False)
        assert len(caplog.records) == 3
        # We expect the warnings to not repeat
        caplog.clear()
        fake_ccm.warn_invalid_constants(only_new=True)
        assert len(caplog.records) == 0
        # Unless we ask them to
        caplog.clear()
        fake_ccm.warn_invalid_constants(only_new=False)
        assert len(caplog.records) == 3
        # Let's fix the issue and make sure no warnings are shown
        fake_ccm.reset_calc_constant_defaults(confirm=False)
        caplog.clear()
        fake_ccm.warn_invalid_constants(only_new=False)
        assert len(caplog.records) == 0


@pytest.mark.timeout(5)
def test_disconnected_ccm():
    ccm.CCM(alio_prefix='ALIO', theta2fine_prefix='THETA',
            theta2coarse_prefix='THTA', chi2_prefix='CHI',
            x_down_prefix='X:DOWN', x_up_prefix='X:UP',
            y_down_prefix='Y:DOWN', y_up_north_prefix='Y:UP:NORTH',
            y_up_south_prefix='Y:UP:SOUTH', in_pos=8, out_pos=0,
            name='ccm')

<filename>pcdsdevices/tests/test_ccm.py
import logging
import time

import numpy as np
import pytest
from ophyd.sim import fake_device_cache, make_fake_device

from .. import ccm
from ..sim import FastMotor

logger = logging.getLogger(__name__)


SAMPLE_ALIO = 4.575  # Current value as of writing this file
SAMPLE_THETA = 1.2  # Modest angle
SAMPLE_WAVELENGTH = 1.5  # hard xray


# Make sure the calcs are properly inverted
def test_theta_alio_inversion():
    logger.debug('test_theta_alio_inversion')
    theta = ccm.alio_to_theta(SAMPLE_ALIO, ccm.default_theta0, ccm.default_gr,
                              ccm.default_gd)
    alio_calc = ccm.theta_to_alio(theta, ccm.default_theta0, ccm.default_gr,
                                  ccm.default_gd)
    # Unlike the other inversions, this is just an approximation
    assert np.isclose(alio_calc, SAMPLE_ALIO)


def test_wavelength_theta_inversion():
    logger.debug('test_wavelength_theta_inversion')
    wavelength = ccm.theta_to_wavelength(SAMPLE_THETA, ccm.default_dspacing)
    theta = ccm.wavelength_to_theta(wavelength, ccm.default_dspacing)
    logger.debug('%s, %s', wavelength, theta)
    assert np.isclose(theta, SAMPLE_THETA)
    theta = ccm.wavelength_to_theta(SAMPLE_WAVELENGTH, ccm.default_dspacing)
    wavelength = ccm.theta_to_wavelength(theta, ccm.default_dspacing)
    logger.debug('%s, %s', wavelength, theta)
    assert np.isclose(wavelength, SAMPLE_WAVELENGTH)


def test_energy_wavelength_inversion():
    logger.debug('test_energy_wavelength_inversion')
    energy = ccm.wavelength_to_energy(SAMPLE_WAVELENGTH)
    wavelength_calc = ccm.energy_to_wavelength(energy)
    assert wavelength_calc == SAMPLE_WAVELENGTH


@pytest.fixture(scope='function')
def fake_ccm():
    return make_fake_ccm()


class FakeAlio(FastMotor):
    kill = None
    home = None


def make_fake_ccm():
    fake_device_cache[ccm.CCMMotor] = FastMotor
    fake_device_cache[ccm.CCMAlio] = FakeAlio
    FakeCCM = make_fake_device(ccm.CCM)
    fake_ccm = FakeCCM(alio_prefix='ALIO', theta2fine_prefix='THETA',
                       theta2coarse_prefix='THTA', chi2_prefix='CHI',
                       x_down_prefix='X:DOWN', x_up_prefix='X:UP',
                       y_down_prefix='Y:DOWN', y_up_north_prefix='Y:UP:NORTH',
                       y_up_south_prefix='Y:UP:SOUTH', in_pos=8, out_pos=0,
                       name='fake_ccm')

    def init_pos(mot, pos=0):
        mot.user_readback.sim_put(0)
        mot.user_setpoint.sim_put(0)
        mot.user_setpoint.sim_set_limits((0, 0))
        mot.motor_spg.sim_put(2)
        mot.part_number.sim_put('tasdf')

    init_pos(fake_ccm.x.down)
    init_pos(fake_ccm.x.up)
    init_pos(fake_ccm.y.down)
    init_pos(fake_ccm.y.up_north)
    init_pos(fake_ccm.y.up_south)

    fake_ccm.alio.set(SAMPLE_ALIO)
    fake_ccm.energy.alio.set(SAMPLE_ALIO)
    fake_ccm.energy_with_vernier.alio.set(SAMPLE_ALIO)
    fake_ccm.energy_with_vernier.vernier.setpoint.sim_put(0)

    return fake_ccm


def test_fake_ccm(fake_ccm):
    logger.debug('test_fake_ccm')
    fake_ccm.get()


# Make sure we set up the forward/inverse to use the right methods
def test_ccm_calc(fake_ccm):
    logger.debug('test_ccm_calc')
    calc = fake_ccm.energy

    logger.debug('physics pos is %s', calc.position)
    logger.debug('real pos is %s', calc.real_position)
    logger.debug('sample alio is %s', SAMPLE_ALIO)

    theta_func = ccm.alio_to_theta(
        SAMPLE_ALIO,
        calc.theta0_rad_val,
        calc.gr_val,
        calc.gd_val,
    )
    wavelength_func = ccm.theta_to_wavelength(theta_func, calc.dspacing_val)
    energy_func = ccm.wavelength_to_energy(wavelength_func)
    energy = calc.energy.position
    assert energy == energy_func

    calc.alio.move(0)
    calc.move(energy, wait=False)
    assert np.isclose(calc.alio.position, SAMPLE_ALIO)

    calc.alio.move(calc.alio.position)
    calc.move(energy=calc.energy.position, wait=False)
    assert np.isclose(calc.alio.position, SAMPLE_ALIO)


# Make sure sync'd axes work and that unk/in/out states work
@pytest.mark.timeout(5)
def test_ccm_main(fake_ccm):
    logger.debug('test_ccm_main')
    fake_ccm.y.move(5, wait=False)
    assert fake_ccm.y.down.user_setpoint.get() == 5
    assert fake_ccm.y.up_north.user_setpoint.get() == 5
    assert fake_ccm.y.up_south.user_setpoint.get() == 5

    assert fake_ccm.removed
    assert not fake_ccm.inserted

    fake_ccm.x.down.user_readback.sim_put(8)
    fake_ccm.x.up.user_readback.sim_put(8)
    assert not fake_ccm.removed
    assert fake_ccm.inserted

    fake_ccm.x.down.user_readback.sim_put(4)
    fake_ccm.x.up.user_readback.sim_put(4)
    assert not fake_ccm.removed
    assert not fake_ccm.inserted

    fake_ccm.insert(wait=False)
    assert fake_ccm.x.down.user_setpoint.get() == 8
    assert fake_ccm.x.up.user_setpoint.get() == 8

    fake_ccm.remove(wait=False)
    assert fake_ccm.x.down.user_setpoint.get() == 0
    assert fake_ccm.x.up.user_setpoint.get() == 0


@pytest.mark.timeout(5)
def test_vernier(fake_ccm):
    logger.debug('test_vernier')

    pseudopos = fake_ccm.energy_with_vernier

    # Moving with vernier should move the energy request motor too
    pseudopos.move(7, wait=False)
    assert np.isclose(pseudopos.energy.position, 7)
    assert pseudopos.vernier.position == 7000

    pseudopos.move(8, wait=False)
    assert np.isclose(pseudopos.energy.position, 8)
    assert pseudopos.vernier.position == 8000

    pseudopos.move(9, wait=False)
    assert np.isclose(pseudopos.energy.position, 9)
    assert pseudopos.vernier.position == 9000

    # Small moves (less than 30eV) should be skipped on the energy request
    pseudopos.move(9.001, wait=False)
    assert np.isclose(pseudopos.energy.position, 9.001)
    assert pseudopos.vernier.position == 9000

    # Unless we set the option for not skipping them
    pseudopos.vernier.skip_small_moves = False
    pseudopos.move(9.002, wait=False)
    assert np.isclose(pseudopos.energy.position, 9.002)
    assert pseudopos.vernier.position == 9002


@pytest.mark.timeout(5)
def test_set_current_position(fake_ccm):
    logger.debug('test_set_current_position')
    mot = fake_ccm.energy.energy
    for energy in range(6, 14):
        mot.set_current_position(energy)
        assert np.isclose(mot.position, energy)


@pytest.mark.timeout(5)
def test_check_valid_constant(fake_ccm):
    logger.debug('test_check_valid_constant')

    # First call to make_valid sends the first monitor update
    def make_valid(sig, valid):
        if valid:
            sig.put(1)
        else:
            sig.put(0)

    def make_conn(sig, conn):
        sig._metadata['connected'] = conn

    def output(sig):
        return fake_ccm._check_valid_constant(sig, sig.get())

    test_sig = fake_ccm.dspacing

    # Can we get to all the enum values?
    make_conn(test_sig, False)
    assert output(test_sig) == ccm.CCMConstantWarning.ALWAYS_DISCONNECT
    make_conn(test_sig, True)
    make_valid(test_sig, False)
    assert output(test_sig) == ccm.CCMConstantWarning.INVALID_CONNECT
    make_conn(test_sig, False)
    assert output(test_sig) == ccm.CCMConstantWarning.INVALID_DISCONNECT
    make_conn(test_sig, True)
    make_valid(test_sig, True)
    assert output(test_sig) == ccm.CCMConstantWarning.NO_WARNING
    make_conn(test_sig, False)
    assert output(test_sig) == ccm.CCMConstantWarning.VALID_DISCONNECT

    # theta0_deg is allowed to be zero, unlike the others
    test_sig2 = fake_ccm.theta0_deg
    make_conn(test_sig2, True)
    make_valid(test_sig2, False)
    assert output(test_sig2) == ccm.CCMConstantWarning.NO_WARNING


@pytest.mark.timeout(5)
def test_show_constant_warning(fake_ccm, caplog):
    logger.debug('test_show_constant_warning')
    for warning in (
        ccm.CCMConstantWarning.NO_WARNING,
        ccm.CCMConstantWarning.ALWAYS_DISCONNECT,
        ccm.CCMConstantWarning.VALID_DISCONNECT,
        ccm.CCMConstantWarning.INVALID_DISCONNECT,
        ccm.CCMConstantWarning.INVALID_CONNECT,
    ):
        caplog.clear()
        with caplog.at_level(logging.WARNING):
            fake_ccm._show_constant_warning(
                warning,
                fake_ccm.dspacing,
                0.111111,
                0.222222,
            )
        if warning == ccm.CCMConstantWarning.NO_WARNING:
            assert len(caplog.records) == 0
        else:
            assert len(caplog.records) == 1


@pytest.mark.timeout(5)
def test_warn_invalid_constants(fake_ccm, caplog):
    logger.debug('test_warn_invalid_constants')
    # Trick the warning into thinking we've be initialized for a while
    fake_ccm._init_time = time.monotonic() - 1000
    fake_ccm.theta0_deg.put(0)
    fake_ccm.dspacing.put(0)
    fake_ccm.gr.put(0)
    fake_ccm.gd.put(0)
    # We expect three warnings from the fake PVs that start at 0
    caplog.clear()
    with caplog.at_level(logging.WARNING):
        fake_ccm.warn_invalid_constants(only_new=False)
        assert len(caplog.records) == 3
        # We expect the warnings to not repeat
        caplog.clear()
        fake_ccm.warn_invalid_constants(only_new=True)
        assert len(caplog.records) == 0
        # Unless we ask them to
        caplog.clear()
        fake_ccm.warn_invalid_constants(only_new=False)
        assert len(caplog.records) == 3
        # Let's fix the issue and make sure no warnings are shown
        fake_ccm.reset_calc_constant_defaults(confirm=False)
        caplog.clear()
        fake_ccm.warn_invalid_constants(only_new=False)
        assert len(caplog.records) == 0


@pytest.mark.timeout(5)
def test_disconnected_ccm():
    ccm.CCM(alio_prefix='ALIO', theta2fine_prefix='THETA',
            theta2coarse_prefix='THTA', chi2_prefix='CHI',
            x_down_prefix='X:DOWN', x_up_prefix='X:UP',
            y_down_prefix='Y:DOWN', y_up_north_prefix='Y:UP:NORTH',
            y_up_south_prefix='Y:UP:SOUTH', in_pos=8, out_pos=0,
            name='ccm')

# Current value as of writing this file # Modest angle # hard xray # Make sure the calcs are properly inverted # Unlike the other inversions, this is just an approximation # Make sure we set up the forward/inverse to use the right methods # Make sure sync'd axes work and that unk/in/out states work # Moving with vernier should move the energy request motor too # Small moves (less than 30eV) should be skipped on the energy request # Unless we set the option for not skipping them # First call to make_valid sends the first monitor update # Can we get to all the enum values? # theta0_deg is allowed to be zero, unlike the others # Trick the warning into thinking we've be initialized for a while # We expect three warnings from the fake PVs that start at 0 # We expect the warnings to not repeat # Unless we ask them to # Let's fix the issue and make sure no warnings are shown