cristallina/cristallina.py

#!/usr/bin/env python
import sys
import os
from loguru import logger

# at the moment this allows us to group the subdirectories as modules easily
# TODO: a more general way would be to have this cristallina as a installed package
os.chdir("/sf/cristallina/applications/slic/cristallina")


def setup_general_logging():
    """ Setup logging to console and files in both the snapshots and 
        the respective pgroup.
    """

    logger.remove()
    logger.add(
        sys.stderr,
        format="{time:YYYY-MM-DD at HH:mm:ss} | {level} | {message}",
        level="INFO",
    )
    logger.info("Loading started.")

    # create file handler which logs
    try:
        logger.add(
            "/sf/cristallina/applications/beamline/snapshots/slic_logs/cristallina.log",
            format="{time:YYYY-MM-DD at HH:mm:ss} | {level} | {message}",
            level="DEBUG",
            rotation="1 week",
        )
        logger.info("Logging to snapshots.")
    except PermissionError as e:
        logger.warning("Cannot write log file to snapshots.")
        logger.warning(e)

def setup_logging_pgroup(pgroup, level="INFO"):
    try:
        logger.add(
            f"/sf/cristallina/data/{pgroup}/scratch/slic.log",
            format="{time:YYYY-MM-DD at HH:mm:ss} | {level} | {message}",
            level=level,
            rotation="1 week",
        )
        logger.info(f"Logging to pgroup {pgroup}.")
    except PermissionError as e:
        logger.warning(f"Cannot write log file to pgroup {pgroup}.")


# We setup the logging before going further so that
# other modules can write startup messages into the log file.
setup_general_logging()

from beamline.alignment_laser import AlignmentLaser

from slic.gui import GUI
from slic.core.adjustable import Adjustable, PVAdjustable, DummyAdjustable
from slic.core.acquisition import SFAcquisition, PVAcquisition
from slic.core.condition import PVCondition
from slic.core.scanner import Scanner


from slic.devices.general.motor import Motor

from slic.utils import devices, Marker, as_shortcut
from slic.utils import Channels, Config, Elog, Screenshot, PV
from slic.core.acquisition.fakeacquisition import FakeAcquisition


from channels.bs_channels import (
    detectors,
    detectors,
    detectors_MX,
    bs_channels,
    camera_channels,
)

from channels.pv_channels import pvs
from spreadsheet import overview

# from channels_minimal import detectors_min, channels_min, pvs_min


################# DEVICES #################
dummy = DummyAdjustable(units="au")


#from devices.knife_edge import KnifeEdge
#from devices.standa import standa
#from devices.newport import newport


from beamline.components import upstream_attenuator, attenuator, shutter, pulsepicker, front_end_attenuator

from beamline.components import kbHor, kbVer

from systems.components import cta

from gp_exp.components import Newport_large, OWIS



def test_attenuator():
    tfundamental = attenuator.get_transmission()
    try:
        assert tfundamental > 0
    except TypeError:
        logger.warning("No transmission value reported from {attenuator.ID}")


test_attenuator()

# Undulators
from beamline import undulator

undulators = undulator.Undulators()
logger.info(
    f"Using undulator (Aramis) offset to PSSS energy of {undulator.energy_offset} eV."
)


## Slits
from slic.devices.xoptics import slits

## Smaract & attocube stages
from gp_exp.smaract_device_def import smaract_Juraj, smaract_mini_XYZ
from gp_exp.attocube_device_def import attocube
from gp_exp.jj_device_def import jjslits

# Bernina monochromator
from beamline.bernina_mono import BerninaMono

BerninaDCM = BerninaMono("SAROP21-ODCM098")

# Diffractometer
from crq_exp.diffractometer import Diffractometer

diffractometer = Diffractometer("diffractometer")


from crq_exp.dilsc import Dilution
dilution = Dilution()


################# Stand setup ##################

# TODO: requires the stand client, need small howto how to start and configure or let it run all the time
from slic.core.acquisition.spreadsheet import Spreadsheet

# setup spreadsheet for transmission to stand
spreadsheet = Spreadsheet(
   {
       "Transmission" : attenuator.trans1st,
       "Upstream Transmission": upstream_attenuator.trans1st,
       "Energy_setpoint" : undulators,
       "Energy_offset": undulator.energy_offset,
       "TD": diffractometer.td,
       "TRX": diffractometer.tr_x,
       "TRY": diffractometer.tr_y,
       "TRXBASE": diffractometer.trx_base,
       "TRYBASE": diffractometer.try_base,
       "THETA": diffractometer.theta,
       "TWOTHETA": diffractometer.twotheta,      
       "Magnet_X": dilution.x,
       "Magnet_Y": dilution.y,
       "Magnet_Z": dilution.z,
       "DilSc_T_plato": dilution.T_plato,
       "DilSc_T_pucksensor": dilution.T_pucksensor,
   },
   
   placeholders=("comment", "sample", "run_usable"),
   host="saresc-vcons-02.psi.ch",
   port=9090,
)

try:
    from stand.client import Client

    stand_host = "saresc-vcons-02.psi.ch"
    stand_client = Client(host=stand_host, port=9090)
    response = stand_client.get()
    logger.info("Connected to stand server")
except Exception as error:
    # catching excpetions with a broad net because different connection errors can occur.
    logger.warning(f"Cannot connect to stand server on {stand_host}. Disabling spreadsheet.")
    spreadsheet = None 

# spreadsheet = Spreadsheet(adjs, placeholders=PLACEHOLDERS, host="127.0.0.1", port=8080))

################# DAQ Setup #################
instrument = "cristallina"

from pgroups import pgroup, pgroup_scratch


# setup pgroup specific logger
setup_logging_pgroup(pgroup)

daq = SFAcquisition(
    instrument,
    pgroup,
    default_channels=bs_channels,
    default_pvs=pvs,
    default_detectors=detectors,
    rate_multiplicator=1,
    spreadsheet=spreadsheet,
)


# There is a new EPICS buffer, so the archiver is no longer used. This makes sure we are taking PVs from the right place.
daq.update_config_pvs()

from acquisition import multiple_daqs

DAQS = multiple_daqs.generate_DAQS(instrument, pgroup,bs_channels, pvs, detectors, spreadsheet)

# daq_fake = FakeAcquisition(instrument, pgroup)
# daq_PV_only = PVAcquisition(instrument, pgroup, default_channels=channels_ks) # workaround for KS not going to DB

# Run the scan only when gas monitor value larger than 10uJ (and smaller than 2000uJ):
# required fraction defines ammount of data recorded to save the step and move on to the next one
check_intensity_gas_monitor = PVCondition(
    "SARFE10-PBPG050:PHOTON-ENERGY-PER-PULSE-US",
    vmin=4,
    vmax=2000,
    wait_time=0.5,
    required_fraction=0.8,
)

scan = Scanner(default_acquisitions=[daq], condition=check_intensity_gas_monitor)
gui = GUI(scan, show_goto=True, show_spec=True)

logger.info(f"Running at {instrument} with pgroup {pgroup}.")
logger.info("Loading finished.")