Skip to content
This repository has been archived by the owner on Oct 11, 2023. It is now read-only.

Commit

Permalink
Changes to work with a real PandA
Browse files Browse the repository at this point in the history
- Block names can have numbers and underscores in
- Type hints seem to work better on the class, not instance
- Added demo
- Added type hints function which doesn't look through parent classes
  • Loading branch information
coretl authored and olliesilvester committed Aug 3, 2023
1 parent c5efe8b commit 6a20c46
Show file tree
Hide file tree
Showing 2 changed files with 38 additions and 5 deletions.
14 changes: 14 additions & 0 deletions docs/examples/panda_demo.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,14 @@
from bluesky import RunEngine

# these three lines just let you use await statements
# #in ipython terminal with the Run Engine event loop.
from IPython import get_ipython
from ophyd.v2.core import DeviceCollector

from ophyd_epics_devices.panda import PandA

get_ipython().run_line_magic("autoawait", "call_in_bluesky_event_loop")
RE = RunEngine()

with DeviceCollector():
my_panda = PandA("TS-PANDA")
29 changes: 24 additions & 5 deletions src/ophyd_epics_devices/panda.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
from __future__ import annotations

import re
from enum import Enum
from typing import (
Expand Down Expand Up @@ -34,6 +36,19 @@
ctxt = Context("pva")


# Better as utility funcion?
def get_type_hints_no_inheritance(cls):
hints = get_type_hints(cls)
base_classes = cls.__bases__
own_hints = hints
for base_cls in base_classes:
base_hints = get_type_hints(base_cls)
for base_hint_names in base_hints.keys():
if base_hint_names in own_hints:
del own_hints[base_hint_names]
return own_hints


class PulseBlock(Device):
delay: SignalRW[float]
width: SignalRW[float]
Expand Down Expand Up @@ -92,7 +107,7 @@ class PVIEntry(TypedDict, total=False):


def block_name_number(block_name: str) -> Tuple[str, int]:
m = re.match("^([a-z]+)([0-9]*)$", block_name)
m = re.match("^([0-9_a-z]+)([0-9]*)$", block_name)
assert m, f"Expected '<block_name><block_num>', got '{block_name}'"
name, num = m.groups()
return name, int(num or 1)
Expand Down Expand Up @@ -134,7 +149,7 @@ def __init__(self, prefix: str, name: str = "") -> None:

def verify_block(self, name: str, num: int):
"""Given a block name and number, return information about a block."""
anno = get_type_hints(self).get(name)
anno = get_type_hints(type(self)).get(name)

block: Device = Device()

Expand All @@ -155,11 +170,15 @@ async def _make_block(self, name: str, num: int, block_pv: str, sim: bool = Fals
"""
block = self.verify_block(name, num)

field_annos = get_type_hints(block)
field_annos = get_type_hints_no_inheritance(type(block))
block_pvi = await pvi_get(block_pv) if not sim else None

# finds which fields this class actually has, e.g. delay, width...
for sig_name, sig_type in field_annos.items():
# arm PV not currently implemented
if sig_name == "arm":
continue

origin = get_origin(sig_type)
args = get_args(sig_type)

Expand Down Expand Up @@ -219,7 +238,7 @@ async def _make_untyped_block(self, block_pv: str):
return block

def set_attribute(self, name, num, block):
anno = get_type_hints(self).get(name)
anno = get_type_hints(type(self)).get(name)

# get_origin to see if it's a device vector.
if (anno == DeviceVector[PulseBlock]) or (anno == DeviceVector[SeqBlock]):
Expand All @@ -237,7 +256,7 @@ async def connect(self, sim=False) -> None:
makes all required blocks.
"""
pvi = await pvi_get(self._init_prefix + ":PVI") if not sim else None
hints = get_type_hints(self)
hints = get_type_hints_no_inheritance(type(self))

# create all the blocks pvi says it should have,
if pvi:
Expand Down

0 comments on commit 6a20c46

Please sign in to comment.