Skip to content

Instantly share code, notes, and snippets.

@klauer
Created January 21, 2021 19:50
Show Gist options
  • Save klauer/7193ad2989b43a3030b124b6d907ec7e to your computer and use it in GitHub Desktop.
Save klauer/7193ad2989b43a3030b124b6d907ec7e to your computer and use it in GitHub Desktop.
Reusing the async caproto context
import caproto
import asyncio
from caproto.asyncio.client import Context
caproto.config_caproto_logging(level='DEBUG')
class Poller():
def __init__(self):
self.val = 0
self.ctx = None
async def poll(self, pvName):
pv = await self.ctx.get_pvs(pvName)
read = await pv[0].read()
val = read.data[0]
print(pvName, " == ", val)
self.val = val
return val
async def main(self):
async with Context() as self.ctx:
await self.poll("A")
await self.poll("B")
await self.poll("C")
await self.poll("D")
if __name__ == "__main__":
p = Poller()
asyncio.run(p.main())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment