GUI: Manual sample panel (WIP)

daq/src/aaredaq/daq.py

@@ -177,6 +177,11 @@ class AareDAQ:
             raise
 
     def create_sample(self, target: SampleShortInfo):
+        curr_sample = self.__cfg.current_sample
+
+        if curr_sample is not None and curr_sample.location is not None:
+            raise Exception("Sample from TELL is loaded")
+
         self.__aare.create_manual_sample(target)
         self.__cfg.current_sample = target
 
@@ -839,3 +844,10 @@ class AareDAQ:
         if self.__cfg.state == BeamlineStateEnum.DataCollection:
             self.__devs.aerotech.stop()
             self.__jfjoch.cancel()
+
+    def anneal(self, time_s: float):
+        self.__cfg.set_busy(BeamlineStateEnum.SampleAlignment)
+        try:
+            self.__devs.anneal(time_s)
+        finally:
+            self.__cfg.state_busy = False

daq/src/aaredaq/devices.py

@@ -1,3 +1,5 @@
+import time
+
 from epics import PV
 
 from aaredaqlib.beamline import MXBeamline
@@ -47,6 +49,7 @@ class BeamlineDevices:
         self.bsy = MyMotor(f"{BEAMLINE}-ES-BS:TRY1")
 
         self.__bpm = PV(f"{BEAMLINE}-OP-XBPM1:SumAll:MeanValue_RBV")
+        self.__anneal = PV(f"{BEAMLINE}-ES-ANN:SET_POS")
 
         self.aerotech = aerotech.Abr(beamline)
 
@@ -339,4 +342,9 @@ class BeamlineDevices:
         # 2 nA -> 4e11 ph/s
         if self.transmission.get() is None:
             return 0.0
-        return self.transmission.get() * self.__bpm.value / 2.0 * 4e11
+        return self.transmission.get() * self.__bpm.value / 2.0 * 4e11
+
+    def anneal(self, value: float):
+        self.__anneal.put(1)
+        time.sleep(value)
+        self.__anneal.put(0)

daq/src/aaredaq/server.py

@@ -98,6 +98,12 @@ async def save_abr_meas_pos(token: str = Depends(oauth2_scheme)):
     daq.save_abr_meas_pos()
     return "OK"
 
+@app.post("/beamline/anneal")
+async def anneal(time_s: float, token: str = Depends(oauth2_scheme)):
+    auth.check_jwt_rw(cfg, auth.parse_token(token))
+    daq.anneal(time_s)
+    return "OK"
+
 @app.post("/beamline/goto_abr_meas_pos")
 async def goto_abr_meas_pos(token: str = Depends(oauth2_scheme)):
     auth.check_jwt_rw(cfg, auth.parse_token(token))