Skip to content

Update OpticsCorr drivers: change how "write" functions handle "command" PVs #330

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
VitorSouzaLNLS wants to merge 6 commits into
base: master
Choose a base branch
from
Open

Update OpticsCorr drivers: change how "write" functions handle "command" PVs #330

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
bf7db9d
MNT (OPTICSCORR.DRIVER): Update drivers' "write" function
VitorSouzaLNLS Aug 13, 2026
77b8bee
MNT (OPTICSCORR.DRIVER): Adjust logging on "write"
VitorSouzaLNLS Aug 14, 2026
71c9fb4
MNT (OPTICSCORR.DRIVER): Adjust logging on "write" (chrom.py)
VitorSouzaLNLS Aug 14, 2026
53ff82b
MNT, STY (OPTICSCORR.DRIVER): Format driver log messages in write method
VitorSouzaLNLS Aug 19, 2026
690c193
MNT (OPTICSCORR.DRIVER): define old_val before the app writing
VitorSouzaLNLS Aug 19, 2026
dd26789
STY (OPTICSCORR): Format with Ruff
VitorSouzaLNLS Aug 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 36 additions & 7 deletions as-ap-opticscorr/as_ap_opticscorr/chrom/chrom.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,6 @@ def _attribute_access_security_group(server, dbase):


class _PCASDriver(_pcaspy.Driver):

def __init__(self, app):
"""Initialize driver."""
super().__init__()
Expand All @@ -49,9 +48,38 @@ def read(self, reason):

def write(self, reason, value):
"""Write IOC pvs according to main application."""
if self.app.write(reason, value):
return super().write(reason, value)
return False
if not self._is_valid(reason, value):
return False
old_val = self.getParam(reason)
ret = self.app.write(reason, value)
if reason.endswith('-Cmd'):
value = old_val + 1
if ret:
msg = f'YES write {reason}: {str(value)}'
_log.info(msg)
else:
msg = f'NO write {reason}: {str(value)} '
msg += f'(current value is {str(old_val)})'
_log.warning(msg)
value = old_val
self.setParam(reason, value)
self.updatePV(reason)
return True

def _is_valid(self, reason, val):
if reason.endswith(('-Sts', '-RB', '-Mon', '-Cte')):
strf = f'PV {reason:s} is read only.'
_log.debug(strf)
return False
if val is None:
msg = 'client tried to set None value. refusing...'
_log.error(msg)
return False
enums = self.getParamInfo(reason, info_keys=('enums',))['enums']
if enums and isinstance(val, int) and val >= len(enums):
_log.warning('value %d too large for enum type PV %s', val, reason)
return False
return True

def update_pv(self, pvname, value=None, info=None, field='value', **kws):
"""Update PV."""
Expand Down Expand Up @@ -87,11 +115,12 @@ def run(acc):

# check if another IOC is running
_util.print_ioc_banner(
ioc_name=acc.upper()+'-AP-ChromCorr',
ioc_name=acc.upper() + '-AP-ChromCorr',
db=dbase,
description=acc.upper()+'-AP-ChromCorr Soft IOC',
description=acc.upper() + '-AP-ChromCorr Soft IOC',
version=_version,
prefix=_ioc_prefix)
prefix=_ioc_prefix,
)

# create a new simple pcaspy server and driver to respond client's requests
server = _pcaspy.SimpleServer()
Expand Down
43 changes: 36 additions & 7 deletions as-ap-opticscorr/as_ap_opticscorr/tune/tune.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,6 @@ def _attribute_access_security_group(server, dbase):


class _PCASDriver(_pcaspy.Driver):

def __init__(self, app):
"""Initialize driver."""
super().__init__()
Expand All @@ -49,9 +48,38 @@ def read(self, reason):

def write(self, reason, value):
"""Write IOC pvs according to main application."""
if self.app.write(reason, value):
return super().write(reason, value)
return False
if not self._is_valid(reason, value):
return False
old_val = self.getParam(reason)
ret = self.app.write(reason, value)
if reason.endswith('-Cmd'):
value = old_val + 1
if ret:
msg = f'YES write {reason}: {str(value)}'
_log.info(msg)
else:
msg = f'NO write {reason}: {str(value)} '
msg += f'(current value is {str(old_val)})'
_log.warning(msg)
value = old_val
self.setParam(reason, value)
self.updatePV(reason)
return True

def _is_valid(self, reason, val):
if reason.endswith(('-Sts', '-RB', '-Mon', '-Cte')):
strf = f'PV {reason:s} is read only.'
_log.debug(strf)
return False
if val is None:
msg = 'client tried to set None value. refusing...'
_log.error(msg)
return False
enums = self.getParamInfo(reason, info_keys=('enums',))['enums']
if enums and isinstance(val, int) and val >= len(enums):
_log.warning('value %d too large for enum type PV %s', val, reason)
return False
return True

def update_pv(self, pvname, value, **kwargs):
"""Update PV."""
Expand Down Expand Up @@ -84,11 +112,12 @@ def run(acc):

# check if another IOC is running
_util.print_ioc_banner(
ioc_name=acc.upper()+'-AP-TuneCorr',
ioc_name=acc.upper() + '-AP-TuneCorr',
db=dbase,
description=acc.upper()+'-AP-TuneCorr Soft IOC',
description=acc.upper() + '-AP-TuneCorr Soft IOC',
version=_version,
prefix=_ioc_prefix)
prefix=_ioc_prefix,
)

# create a new simple pcaspy server and driver to respond client's requests
server = _pcaspy.SimpleServer()
Expand Down