test_calcat_utils.py

import pathlib
import timeit

from calng import calcat_utils
from calng.utils import Stopwatch
from karabo.bound import Hash, Schema

# TODO: secrets management
_test_dir = pathlib.Path(__file__).absolute().parent
_test_calcat_secrets_fn = _test_dir / "calibration-client-secrets.json"


class DummyLogger:
    DEBUG = print
    INFO = print
    WARN = print


class DummyBaseDevice:
    log = DummyLogger()

    def log_status_info(self, msg):
        self.log.INFO(msg)

    def log_status_warn(self, msg):
        self.log.WARN(msg)

    def get(self, key):
        return self.schema.get(key)

    def set(self, key, value):
        print(f'Would set "{key}" = {value}')


# TODO: consider testing by attaching to real karabo.bound.PythonDevice
class DummyAgipdDevice(DummyBaseDevice):
    device_class_schema = Schema()
    managed_keys = set()

    @staticmethod
    def expectedParameters(expected):
        calcat_utils.AgipdCalcatFriend.add_schema(
            expected, DummyAgipdDevice.managed_keys
        )

    def __init__(self, config):
        self.schema = config
        self.calibration_constant_manager = calcat_utils.AgipdCalcatFriend(
            self,
            _test_calcat_secrets_fn,
        )
        print(self.managed_keys)


DummyAgipdDevice.expectedParameters(DummyAgipdDevice.device_class_schema)
print(DummyAgipdDevice.device_class_schema)
print(DummyAgipdDevice.managed_keys)


class DummyDsscDevice(DummyBaseDevice):
    device_class_schema = Schema()
    managed_keys = set()

    @staticmethod
    def expectedParameters(expected):
        # super(DummyDsscDevice, DummyDsscDevice).expectedParameters(expected)
        calcat_utils.DsscCalcatFriend.add_schema(expected, DummyDsscDevice.managed_keys)

    def __init__(self, config):
        # TODO: check config against schema (as Karabo would)
        self.schema = config
        self.calibration_constant_manager = calcat_utils.DsscCalcatFriend(
            self,
            _test_calcat_secrets_fn,
        )


DummyDsscDevice.expectedParameters(DummyDsscDevice.device_class_schema)


def test_agipd_constants_and_caching_and_async():
    # def test_agipd_constants():
    conf = Hash()
    conf["constantParameters.detectorType"] = "AGIPD-Type"
    conf["constantParameters.detectorName"] = "SPB_DET_AGIPD1M-1"
    conf["constantParameters.karaboDa"] = "AGIPD00"
    conf["constantParameters.pixelsX"] = 512
    conf["constantParameters.pixelsY"] = 128
    conf["constantParameters.memoryCells"] = 352
    conf["constantParameters.acquisitionRate"] = 1.1
    conf["constantParameters.biasVoltage"] = 300
    conf["constantParameters.gainSetting"] = 0
    conf["constantParameters.photonEnergy"] = 9.2
    device = DummyAgipdDevice(conf)

    def backcall(constant_name, metadata_and_data):
        # TODO: think of something reasonable to check
        data = metadata_and_data
        assert data.nbytes > 1000

    with Stopwatch() as timer_async_cold:
        # TODO: put this sort of thing in BaseCalcatFriend
        threads = []
        for constant in calcat_utils.AgipdConstants:
            thread = device.calibration_constant_manager.get_constant_version_and_call_me_back(
                constant, backcall
            )
            threads.append(thread)
        for thread in threads:
            thread.join()

    with Stopwatch() as timer_async_warm:
        threads = []
        for constant in calcat_utils.AgipdConstants:
            thread = device.calibration_constant_manager.get_constant_version_and_call_me_back(
                constant, backcall
            )
            threads.append(thread)
        for thread in threads:
            thread.join()

    with Stopwatch() as timer_sync_warm:
        for constant in calcat_utils.AgipdConstants:
            data = device.calibration_constant_manager.get_constant_version(
                constant,
            )
            assert data.nbytes > 1000, "Should find some constant data"

    print(f"Cold async took {timer_async_cold.elapsed} s")
    print(f"Warm async took {timer_async_warm.elapsed} s")
    print(f"Warm sync took {timer_sync_warm.elapsed} s")
    assert (
        timer_async_cold.elapsed > timer_async_warm.elapsed
    ), "Caching should make second go faster"
    assert timer_sync_warm.elapsed > timer_async_warm.elapsed, "Async should be faster"


def test_dssc_constants():
    conf = Hash()
    conf["constantParameters.detectorType"] = "DSSC-Type"
    conf["constantParameters.detectorName"] = "SCS_DET_DSSC1M-1"
    conf["constantParameters.karaboDa"] = "DSSC00"
    conf["constantParameters.memoryCells"] = 400
    conf["constantParameters.biasVoltage"] = 100
    conf["constantParameters.pixelsX"] = 512
    conf["constantParameters.pixelsY"] = 128
    # conf["constantParameters.pulseIdChecksum"] = 2.8866323107820637e-36
    # conf["constantParameters.acquisitionRate"] = 4.5
    # conf["constantParameters.encodedGain"] = 67328
    device = DummyDsscDevice(conf)
    offset_map = device.calibration_constant_manager.get_constant_version("Offset")

    assert offset_map is not None