From 8903e228eec9565ee332e2a35067a9dd495dd353 Mon Sep 17 00:00:00 2001 From: Peter Wegmann Date: Thu, 2 Nov 2023 13:57:24 +0100 Subject: [PATCH] fixed some TODOs - added timeouts to backend put methods - removed sleep on connect - robo testing --- jupyter/robo-testing.ipynb | 191 ++++++++++++++++++++++++++++--- secop_ophyd/AsyncFrappyClient.py | 2 - secop_ophyd/SECoPDevices.py | 10 +- secop_ophyd/SECoPSignal.py | 30 +++-- test/test_async_frappy_client.py | 13 ++- 5 files changed, 207 insertions(+), 39 deletions(-) diff --git a/jupyter/robo-testing.ipynb b/jupyter/robo-testing.ipynb index 6db2a0c..9749286 100644 --- a/jupyter/robo-testing.ipynb +++ b/jupyter/robo-testing.ipynb @@ -2,7 +2,7 @@ "cells": [ { "cell_type": "code", - "execution_count": null, + "execution_count": 1, "metadata": {}, "outputs": [], "source": [ @@ -19,6 +19,8 @@ "\n", "\n", "\n", + "\n", + "\n", "# Create a run engine and a temporary file backed database. Send all the documents from the RE into that database\n", "RE = RunEngine({})\n", "db = temp()\n", @@ -115,25 +117,55 @@ }, { "cell_type": "code", - "execution_count": null, + "execution_count": 2, "metadata": {}, - "outputs": [], + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "1\n", + "2\n", + "3\n", + "4\n", + "5\n", + "6\n" + ] + }, + { + "data": { + "text/plain": [ + "()" + ] + }, + "execution_count": 2, + "metadata": {}, + "output_type": "execute_result" + } + ], "source": [ + "def load_samples(storage, lower, upper):\n", + " loadshort:SECoP_CMD_Device = storage.load_short_dev\n", "\n", - "loadshort:SECoP_CMD_Device = storage.load_short_dev\n", + " for samplepos in range(lower,upper):\n", + " #await asyncio.sleep(2)\n", + " yield from bps.abs_set(loadshort.samplepos_arg,samplepos, group='sample')\n", + " yield from bps.abs_set(loadshort.substance_arg,random.randint(0,6), group='sample') \n", "\n", - "for samplepos in range(1,7):\n", - " #await asyncio.sleep(2)\n", - " await loadshort.samplepos_arg.set(samplepos)\n", - " await loadshort.substance_arg.set(random.randint(0,6))\n", + " yield from bps.wait('sample')\n", + " \n", + " \n", + " # Execute load short command\n", + " yield from bps.trigger(loadshort,wait=True)\n", "\n", - " await loadshort.load_short_x.execute()\n", "\n", - " await storage.wait_for_IDLE()\n", - " await robot.wait_for_IDLE()\n", - " #await asyncio.sleep(2)\n", - " print(samplepos)\n", - " \n", + "\n", + " #yield from bps.wait_for([storage.wait_for_IDLE,robot.wait_for_IDLE])\n", + "\n", + " print(samplepos)\n", + "\n", + "\n", + "RE(load_samples(storage,1,7))\n", " " ] }, @@ -229,7 +261,134 @@ " yield from bps.mv(sample,1)\n", " yield from bps.mv(sample,0)\n", " \n", - "RE(dumb(sample))" + "RE(dumb(sample))\n", + "\n" + ] + }, + { + "cell_type": "code", + "execution_count": 2, + "metadata": {}, + "outputs": [ + { + "data": { + "text/plain": [ + "()" + ] + }, + "execution_count": 2, + "metadata": {}, + "output_type": "execute_result" + } + ], + "source": [ + "async def test_except():\n", + " await asyncio.sleep(1)\n", + " raise RuntimeError\n", + " \n", + " \n", + "def except_plan():\n", + " yield from bps.wait_for([test_except])\n", + " \n", + " \n", + "RE(except_plan())" + ] + }, + { + "cell_type": "code", + "execution_count": 6, + "metadata": {}, + "outputs": [ + { + "data": { + "text/plain": [ + "({ exception=RuntimeError()>},\n", + " set())" + ] + }, + "execution_count": 6, + "metadata": {}, + "output_type": "execute_result" + } + ], + "source": [ + "fut = asyncio.ensure_future(test_except())\n", + "\n", + "await asyncio.wait([fut])\n" + ] + }, + { + "cell_type": "code", + "execution_count": 3, + "metadata": {}, + "outputs": [ + { + "ename": "AttributeError", + "evalue": "primary", + "output_type": "error", + "traceback": [ + "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m", + "\u001b[0;31mKeyError\u001b[0m Traceback (most recent call last)", + "File \u001b[0;32m~/git-repos/secop-ophyd/.venv/lib64/python3.11/site-packages/intake/catalog/base.py:350\u001b[0m, in \u001b[0;36mCatalog.__getattr__\u001b[0;34m(self, item)\u001b[0m\n\u001b[1;32m 349\u001b[0m \u001b[39mtry\u001b[39;00m:\n\u001b[0;32m--> 350\u001b[0m \u001b[39mreturn\u001b[39;00m \u001b[39mself\u001b[39;49m[item] \u001b[39m# triggers reload_on_change\u001b[39;00m\n\u001b[1;32m 351\u001b[0m \u001b[39mexcept\u001b[39;00m \u001b[39mKeyError\u001b[39;00m \u001b[39mas\u001b[39;00m e:\n", + "File \u001b[0;32m~/git-repos/secop-ophyd/.venv/lib64/python3.11/site-packages/intake/catalog/base.py:423\u001b[0m, in \u001b[0;36mCatalog.__getitem__\u001b[0;34m(self, key)\u001b[0m\n\u001b[1;32m 422\u001b[0m \u001b[39mreturn\u001b[39;00m out()\n\u001b[0;32m--> 423\u001b[0m \u001b[39mraise\u001b[39;00m \u001b[39mKeyError\u001b[39;00m(key)\n", + "\u001b[0;31mKeyError\u001b[0m: 'primary'", + "\nThe above exception was the direct cause of the following exception:\n", + "\u001b[0;31mAttributeError\u001b[0m Traceback (most recent call last)", + "\u001b[1;32m/home/peter/git-repos/secop-ophyd/jupyter/robo-testing.ipynb Cell 13\u001b[0m line \u001b[0;36m4\n\u001b[1;32m 45\u001b[0m RE(measure(sample,i))\n\u001b[1;32m 48\u001b[0m run\u001b[39m=\u001b[39mdb[\u001b[39m-\u001b[39m\u001b[39m1\u001b[39m]\n\u001b[0;32m---> 49\u001b[0m run\u001b[39m.\u001b[39;49mprimary\u001b[39m.\u001b[39mread()\n", + "File \u001b[0;32m~/git-repos/secop-ophyd/.venv/lib64/python3.11/site-packages/intake/catalog/base.py:352\u001b[0m, in \u001b[0;36mCatalog.__getattr__\u001b[0;34m(self, item)\u001b[0m\n\u001b[1;32m 350\u001b[0m \u001b[39mreturn\u001b[39;00m \u001b[39mself\u001b[39m[item] \u001b[39m# triggers reload_on_change\u001b[39;00m\n\u001b[1;32m 351\u001b[0m \u001b[39mexcept\u001b[39;00m \u001b[39mKeyError\u001b[39;00m \u001b[39mas\u001b[39;00m e:\n\u001b[0;32m--> 352\u001b[0m \u001b[39mraise\u001b[39;00m \u001b[39mAttributeError\u001b[39;00m(item) \u001b[39mfrom\u001b[39;00m \u001b[39me\u001b[39;00m\n\u001b[1;32m 353\u001b[0m \u001b[39mraise\u001b[39;00m \u001b[39mAttributeError\u001b[39;00m(item)\n", + "\u001b[0;31mAttributeError\u001b[0m: primary" + ] + } + ], + "source": [ + "def measure(sample,sample_num):\n", + " \n", + "\n", + " \n", + " reading = yield from bps.read(sample)\n", + " \n", + " \n", + " curr_sample = reading[sample.value.name]['value']\n", + " \n", + " # holding wrong sample --> put it back into storage\n", + " if curr_sample != 0 and curr_sample != sample_num :\n", + " yield from bps.mv(sample,0)\n", + " \n", + " # gripper empty --> grab correct sample\n", + " if curr_sample == 0:\n", + " yield from bps.mv(sample,i)\n", + " \n", + " # Do actual measurement\n", + "\n", + " @bpp.run_decorator()\n", + " def inner_meas(sample):\n", + "\n", + " complete_status = yield from bps.complete(sample.measure_dev, wait=False) #This message doesn't exist yet\n", + " \n", + " # While the device is still executing, read from the detectors in the detectors list\n", + " while not complete_status.done:\n", + "\n", + " yield Msg('checkpoint') # allows us to pause the run \n", + " \n", + " yield Msg('sleep', None, 1) \n", + " \n", + " uid = yield from inner_meas(sample)\n", + "\n", + " \n", + " # put sample back into storage\n", + " yield from bps.mv(sample,0)\n", + "\n", + " return uid\n", + "\n", + "\n", + "\n", + "\n", + "for i in range(1,7):\n", + " #grab sample i and hold in Measurement Pos\n", + " RE(measure(sample,i))\n", + " \n", + "\n", + "run=db[-1]\n" ] } ], @@ -249,7 +408,7 @@ "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", - "version": "3.11.5" + "version": "3.11.6" }, "orig_nbformat": 4 }, diff --git a/secop_ophyd/AsyncFrappyClient.py b/secop_ophyd/AsyncFrappyClient.py index e8fb2bd..7603dc2 100644 --- a/secop_ophyd/AsyncFrappyClient.py +++ b/secop_ophyd/AsyncFrappyClient.py @@ -88,8 +88,6 @@ async def create(cls, host, port, loop, log=Logger): async def connect(self, try_period=0): await asyncio.to_thread(self.client.connect, try_period) self.conn_timestamp = time.time() - # TODO find better solution than sleep,somehow it is needed - await asyncio.sleep(1) async def disconnect(self, shutdown=True): await asyncio.to_thread(self.client.disconnect, shutdown) diff --git a/secop_ophyd/SECoPDevices.py b/secop_ophyd/SECoPDevices.py index baff329..ce60db4 100644 --- a/secop_ophyd/SECoPDevices.py +++ b/secop_ophyd/SECoPDevices.py @@ -17,6 +17,7 @@ Stoppable, SyncOrAsync, Flyable, + Triggerable, ) from ophyd_async.core.standard_readable import StandardReadable @@ -357,7 +358,6 @@ async def _move(self, new_target): async def stop(self, success=True) -> SyncOrAsync[None]: self._success = success - traceback.print_stack() await self._secclient.execCommand(self._module, "stop") self._stopped = True @@ -439,7 +439,7 @@ def __init__(self, path: Path, secclient: AsyncFrappyClient): super().__init__(name=dev_name) -class SECoP_CMD_Device(StandardReadable, Flyable): +class SECoP_CMD_Device(StandardReadable, Flyable, Triggerable): """ Command devices that have Signals for command args, return values and a signal for triggering command execution @@ -578,9 +578,9 @@ def collect(self) -> Iterator[PartialEvent]: async def describe_collect(self) -> SyncOrAsync[Dict[str, Dict[str, Descriptor]]]: return await self.describe() - -class SECoP_ArrayOf_XDevice(StandardReadable): - pass + def trigger(self) -> Status: + coro = asyncio.wait_for(fut=self._exec_cmd(), timeout=None) + return AsyncStatus(awaitable=coro, watchers=None) class SECoP_Node_Device(StandardReadable): diff --git a/secop_ophyd/SECoPSignal.py b/secop_ophyd/SECoPSignal.py index 0de5179..0fbaa1b 100644 --- a/secop_ophyd/SECoPSignal.py +++ b/secop_ophyd/SECoPSignal.py @@ -201,10 +201,14 @@ async def put(self, value: Any | None, wait=True, timeout=None): argument = arg_datatype.export_datatype(await sig.get_value()) # Run SECoP Command - res, qualifiers = await self._secclient.execCommand( - module=self.path._module_name, - command=self.path._accessible_name, - argument=argument, + + res, qualifiers = await asyncio.wait_for( + fut=self._secclient.execCommand( + module=self.path._module_name, + command=self.path._accessible_name, + argument=argument, + ), + timeout=timeout, ) # write return Value to corresponding Backends @@ -294,9 +298,8 @@ async def connect(self): pass async def put(self, value: Any | None, wait=True, timeout=None): - # TODO wait + timeout - # top level nested datatypes (handled as sting Signals) + if self.path._dev_path == []: if self.SECoPdtype == "tuple": value = self.SECoPdtype_obj.from_string(value) @@ -304,7 +307,11 @@ async def put(self, value: Any | None, wait=True, timeout=None): if self.SECoPdtype == "struct": value = self.SECoPdtype_obj.from_string(value) - await self._secclient.setParameter(**self.get_param_path(), value=value) + await asyncio.wait_for( + self._secclient.setParameter(**self.get_param_path(), value=value), + timeout=timeout, + ) + return # signal sub element of SECoP parameter (tuple or struct member) @@ -320,7 +327,10 @@ async def put(self, value: Any | None, wait=True, timeout=None): new_val = self.path.insert_val(curr_val, value) # set new value - await self._secclient.setParameter(**self.get_param_path(), value=new_val) + await asyncio.wait_for( + fut=self._secclient.setParameter(**self.get_param_path(), value=new_val), + timeout=timeout, + ) async def get_descriptor(self) -> Descriptor: res = {} @@ -440,7 +450,7 @@ def _set_dtype(self) -> None: class PropertyBackend(SignalBackend): - """A read/write/monitor backend for a Signals""" + """read backend for a SECoP Properties""" def __init__( self, prop_key: str, propertyDict: Dict[str, T], secclient: AsyncFrappyClient @@ -517,8 +527,6 @@ def __init__(self, prefix, name, module_name, param_desc, secclient, kind) -> No # TODO: Array: shape for now only for the first Dim, later maybe recursive?? -# TODO: status tuple - # Tuple and struct are handled in a special way. They are unfolded into subdevices diff --git a/test/test_async_frappy_client.py b/test/test_async_frappy_client.py index f931297..f40391b 100644 --- a/test/test_async_frappy_client.py +++ b/test/test_async_frappy_client.py @@ -42,17 +42,20 @@ async def test_async_secopclient_reconn( await async_frappy_client.disconnect(False) - assert async_frappy_client.state == "reconnecting" + # for a short period the status is still "connected" (the disconn task finishes and the state is only set to a new value once the reconnect thread starts) + while async_frappy_client.state == "connected": + await asyncio.sleep(0.001) - await asyncio.sleep(1) + while async_frappy_client.state == "reconnecting": + await asyncio.sleep(0.001) assert async_frappy_client.state == "connected" # ensures we are connected and getting fresh data again - reading1 = await async_frappy_client.getParameter("cryo", "value", False) - reading2 = await async_frappy_client.getParameter("cryo", "value", False) + reading3 = await async_frappy_client.getParameter("cryo", "value", False) + reading4 = await async_frappy_client.getParameter("cryo", "value", False) - assert reading1.get_value() != reading2.get_value() + assert reading3.get_value() != reading4.get_value() await async_frappy_client.disconnect(True)