Skip to content
Snippets Groups Projects
Commit 580d8d07 authored by Corné Lukken's avatar Corné Lukken
Browse files

Restructure code for pep8 and black linting

parent ca58c4b6
No related branches found
No related tags found
No related merge requests found
......@@ -80,7 +80,8 @@ class AttributeIO(object):
if value_shape != self.attribute_wrapper.shape:
raise ValueError(
f"Tried writing an array of shape {value_shape} into an attribute of shape {self.attribute_wrapper.shape}"
f"Tried writing an array of shape {value_shape} into an attribute "
f"of shape {self.attribute_wrapper.shape}"
)
self.write_function(value)
......@@ -88,9 +89,7 @@ class AttributeIO(object):
class AttributeWrapper(attribute):
"""
Wraps all the attributes in a wrapper class to manage most of the redundant code behind the scenes
"""
"""Wraps attribute to generate function creation"""
def __init__(
self,
......@@ -102,22 +101,28 @@ class AttributeWrapper(attribute):
**kwargs,
):
"""
wraps around the tango Attribute class. Provides an easier interface for 1d or 2d arrays. Also provides a way to abstract
managing the communications interface.
wraps around the tango Attribute class.
Provides an easier interface for 1d or 2d arrays. Also provides a way to
abstract managing the communications interface.
comms_id: user-supplied identifier that is attached to this object, to identify which communication class will need to be attached
comms_annotation: data passed along to the attribute. can be given any form of data. handling is up to client implementation
comms_id: user-supplied identifier that is attached to this object, to identify
which communication class will need to be attached
comms_annotation: data passed along to the attribute. can be given any form of
data. handling is up to client implementation
datatype: any numpy datatype
dims: dimensions of the attribute as a tuple, or (1,) for a scalar.
init_value: value
"""
# ensure the type is a numpy array.
# see also https://pytango.readthedocs.io/en/stable/server_api/server.html?highlight=devlong#module-tango.server for
# more details about type conversion Python/numpy -> PyTango
# see also
# https://pytango.readthedocs.io/en/stable/server_api/server.html?highlight=devlong#module-tango.server
# for more details about type conversion Python/numpy -> PyTango
if "numpy" not in str(datatype) and datatype != str and datatype != bool:
raise ValueError(
f"Attribute needs to be a Tango-supported numpy, str or bool type, but has type {datatype}"
f"Attribute needs to be a Tango-supported numpy, str or bool type, but "
f"has type {datatype}"
)
# store data that can be used to identify the comms interface to use. not
......@@ -150,8 +155,8 @@ class AttributeWrapper(attribute):
dtype = ((datatype,),)
shape = (max_dim_y, max_dim_x)
else:
# higher dimensional
# >2D arrays collapse into the X and Y dimensions. The Y (major) dimension mirrors the first dimension given, the
# higher dimensional, >2D arrays collapse into the X and Y dimensions.
# The Y (major) dimension mirrors the first dimension given, the
# rest collapses into the X (minor) dimension.
max_dim_x = reduce(mul, dims[1:])
max_dim_y = dims[0]
......@@ -162,7 +167,8 @@ class AttributeWrapper(attribute):
self.shape = shape
if access == AttrWriteType.READ_WRITE:
"""If the attribute is of READ_WRITE type, assign the write and read functions to it"""
"""If the attribute is of READ_WRITE type, assign the write and read
functions to it"""
# we return the last written value, as we are the only ones in control,
# and the hardware does not necessarily return what we've written
......@@ -170,9 +176,7 @@ class AttributeWrapper(attribute):
@fault_on_error()
def write_func_wrapper(device, value):
"""
write_func_wrapper writes a value to this attribute
"""
"""write_func_wrapper writes a value to this attribute"""
try:
io = self.get_attribute_io(device)
......@@ -230,11 +234,14 @@ class AttributeWrapper(attribute):
self.fget = read_func_wrapper
# "fisallowed" is called to ask us whether an attribute can be accessed. If not, the attribute won't be accessed,
# and the cache not updated. This forces Tango to also force a read the moment an attribute does become accessible.
# The provided function will be used with the call signature "(device: Device, req_type: AttReqType) -> bool".
# "fisallowed" is called to ask us whether an attribute can be accessed. If not,
# the attribute won't be accessed, and the cache not updated. This forces Tango
# to also force a read the moment an attribute does become accessible.
# The provided function will be used with the call signature:
# "(device: Device, req_type: AttReqType) -> bool".
#
# NOTE: fisallowed=<callable> does not work: https://gitlab.com/tango-controls/pytango/-/issues/435
# NOTE: fisallowed=<callable> does not work:
# https://gitlab.com/tango-controls/pytango/-/issues/435
# So we have to use fisallowed=<str> here, which causes the function
# device.<str> to be called.
super().__init__(
......@@ -248,8 +255,7 @@ class AttributeWrapper(attribute):
)
def get_attribute_io(self, device):
"""returns the attribute I/O functions from a certain device, or registers it if not present
"""
"""Returns the attribute I/O functions from device"""
try:
return device._attribute_wrapper_io[self]
......@@ -258,9 +264,11 @@ class AttributeWrapper(attribute):
return device._attribute_wrapper_io[self]
def set_comm_client(self, device, client):
"""
takes a communications client as input arguments This client should be of a class containing a "get_mapping" function
and return a read and write function that the wrapper will use to get/set data.
"""Takes communication client with 'get_mapping' function
takes a communications client as input arguments This client should be of a
class containing a "get_mapping" function and return a read and write function
that the wrapper will use to get/set data.
"""
try:
read_attr_func, write_attr_func = client.setup_attribute(
......@@ -272,13 +280,12 @@ class AttributeWrapper(attribute):
io.write_function = write_attr_func
except Exception as e:
raise Exception(
f"Exception while setting {client.__class__.__name__} attribute with annotation: '{self.comms_annotation}'"
f"Exception while setting {client.__class__.__name__} attribute with "
f"annotation: '{self.comms_annotation}'"
) from e
async def async_set_comm_client(self, device, client):
"""
Asynchronous version of set_comm_client.
"""
"""Asynchronous version of set_comm_client."""
try:
read_attr_func, write_attr_func = await client.setup_attribute(
self.comms_annotation, self
......@@ -289,7 +296,8 @@ class AttributeWrapper(attribute):
io.write_function = write_attr_func
except Exception as e:
raise Exception(
f"Exception while setting {client.__class__.__name__} attribute with annotation: '{self.comms_annotation}'"
f"Exception while setting {client.__class__.__name__} attribute with "
f"annotation: '{self.comms_annotation}'"
) from e
def set_pass_func(self, device):
......
......@@ -7,8 +7,8 @@ logger = logging.getLogger()
__all__ = ["AttributeWrapperInterface"]
class AttributeWrapperInterface:
class AttributeWrapperInterface:
def __init__(self):
"""prepare the caches for attribute wrapper objects"""
self._attribute_wrapper_io = {}
......@@ -12,15 +12,14 @@ from unittest import TestCase
import testscenarios
# External imports
from tango import DevState, DevFailed, AttrWriteType, DeviceProxy
from tango.server import attribute, command, Device, DeviceMeta
from tango import DevState, DevFailed, AttrWriteType
from tango.server import command, Device, DeviceMeta
# Test imports
from tango.test_context import DeviceTestContext
# Internal imports
from attribute_wrapper.attribute_wrapper import AttributeWrapper
from attribute_wrapper.interface import AttributeWrapperInterface
from attribute_wrapper.states import INITIALISED_STATES
from test_client import TestClient
......@@ -42,7 +41,6 @@ STR_IMAGE_VAL = [["1", "1"], ["1", "1"], ["1", "1"]]
class DeviceWrapper(Device, metaclass=DeviceMeta):
def __init__(self, cl, name):
super().__init__(cl, name)
......@@ -523,27 +521,24 @@ class TestAttributeTypes(testscenarios.WithScenarios, unittest.TestCase):
self.assertEqual(
1,
2,
" {} is not a valid test_type. please use either scalar, spectrum or image".format(
test_type
),
"{} is not a valid test_type. please use either scalar,"
"spectrum or image".format(test_type),
)
if test_type == "scalar":
comparison = expected == val
self.assertTrue(
comparison,
" Value could not be read or was not what was expected. Expected: {}, got {}".format(
expected, val
),
"Value could not be read or was not what was expected. Expected:"
"{}, got {}".format(expected, val),
)
else:
comparison = expected == val
equal_arrays = comparison.all()
self.assertTrue(
equal_arrays,
" Value could not be read or was not what was expected. Expected: {}, got {}".format(
expected, val
),
" Value could not be read or was not what was expected. Expected:"
"{}, got {}".format(expected, val),
)
print(
......@@ -581,16 +576,10 @@ class TestAttributeTypes(testscenarios.WithScenarios, unittest.TestCase):
self.assertEqual(
1,
2,
" {} is not a valid test_type. please use either scalar, spectrum or image".format(
test_type
),
" {} is not a valid test_type. please use either scalar,"
"spectrum or image".format(test_type),
)
# can't really test anything here except that the writing didnt cause an error.
# reading back happens in readback_test
print(" Test passed! Managed to write: ".format(val))
def read_RW_test(self, dev, dtype, test_type):
"""Test device"""
expected = None
......@@ -619,18 +608,16 @@ class TestAttributeTypes(testscenarios.WithScenarios, unittest.TestCase):
self.assertEqual(
val.shape,
expected.shape,
" image R array dimensions got mangled. Expected {}, got {}".format(
expected.shape, val.shape
),
"image R array dimensions got mangled. Expected {},"
"got {}".format(expected.shape, val.shape),
)
val.reshape(-1)
else:
self.assertEqual(
1,
2,
" {} is not a valid test_type. please use either scalar, spectrum or image".format(
test_type
),
" {} is not a valid test_type. please use either scalar,"
"spectrum or image".format(test_type),
)
if test_type != "scalar":
......@@ -639,13 +626,15 @@ class TestAttributeTypes(testscenarios.WithScenarios, unittest.TestCase):
equal_arrays = comparison.all()
self.assertTrue(
equal_arrays,
" Value could not be handled by the atrribute_wrappers internal RW storer",
"Value could not be handled by the atrribute_wrappers internal"
"RW storer",
)
else:
comparison = expected == val
self.assertTrue(
comparison,
" Value could not be handled by the atrribute_wrappers internal RW storer",
"Value could not be handled by the atrribute_wrappers internal"
"RW storer",
)
print(
......@@ -701,9 +690,8 @@ class TestAttributeTypes(testscenarios.WithScenarios, unittest.TestCase):
self.assertEqual(
1,
2,
" {} is not a valid test_type. please use either scalar, spectrum or image".format(
test_type
),
"{} is not a valid test_type. please use either scalar,"
"spectrum or image".format(test_type),
)
return result_R, result_RW, val
......@@ -725,47 +713,45 @@ class TestAttributeTypes(testscenarios.WithScenarios, unittest.TestCase):
comparison = result_RW == val
self.assertTrue(
comparison,
" Value could not be handled by the atrribute_wrappers internal RW storer. attempted to write: {}".format(
val
),
"Value could not be handled by the atrribute_wrappers internal "
"RW storer. attempted to write: {}".format(val),
)
comparison = result_R == val
self.assertTrue(
comparison,
" value in the clients R attribute not equal to what was written. read: {}, wrote {}".format(
result_R, val
),
"value in the clients R attribute not equal to what was"
"written. read: {}, wrote {}".format(result_R, val),
)
elif dtype != str:
comparison = result_RW == val
equal_arrays = comparison.all()
self.assertTrue(
equal_arrays,
" Value could not be handled by the atrribute_wrappers internal RW storer. attempted to write: {}".format(
val
),
"Value could not be handled by the atrribute_wrappers internal "
"RW storer. attempted to write: {}".format(val),
)
comparison = result_R == val
equal_arrays = comparison.all()
self.assertTrue(
equal_arrays,
" value in the clients R attribute not equal to what was written. read: {}, wrote {}".format(
result_R, val
),
"value in the clients R attribute not equal to what was "
"written. read: {}, wrote {}".format(result_R, val),
)
else:
if test_type == "image":
self.assertEqual(
len(result_RW) * len(result_RW[0]),
6,
"array dimensions do not match the expected dimensions. expected {}, got: {}".format(
"array dimensions do not match the expected dimensions."
"expected {}, got: {}".format(
val, len(result_RW) * len(result_RW[0])
),
)
self.assertEqual(
len(result_RW) * len(result_RW[0]),
6,
"array dimensions do not match the expected dimensions. expected {}, got: {}".format(
"array dimensions do not match the expected dimensions."
"expected {}, got: {}".format(
val, len(result_R) * len([0])
),
)
......@@ -773,16 +759,14 @@ class TestAttributeTypes(testscenarios.WithScenarios, unittest.TestCase):
self.assertEqual(
len(result_RW),
4,
"array dimensions do not match the expected dimensions. expected {}, got: {}".format(
4, len(result_RW)
),
"array dimensions do not match the expected dimensions."
"expected {}, got: {}".format(4, len(result_RW)),
)
self.assertEqual(
len(result_R),
4,
"array dimensions do not match the expected dimensions. expected {}, got: {}".format(
4, len(result_R)
),
"array dimensions do not match the expected dimensions."
"expected {}, got: {}".format(4, len(result_R)),
)
print(
......@@ -790,9 +774,10 @@ class TestAttributeTypes(testscenarios.WithScenarios, unittest.TestCase):
)
except Exception as e:
info = "Test failure in {} {} readback test \n\tW: {} \n\tRW: {} \n\tR: {}".format(
test_type, dtype, val, result_RW, result_R
)
info = {
f"Test failure in {test_type} {dtype} readback test "
f"\n\tW: {val} \n\tRW: {result_RW} \n\tR: {result_R}"
}
raise Exception(info) from e
"""
......
......@@ -10,9 +10,10 @@ logger = logging.getLogger()
class TestClient:
"""
this class provides an example implementation of a comms_client.
During initialisation it creates a correctly shaped zero filled value. on read that value is returned and on write its modified.
"""Example comms_client implementation
During initialisation it creates a correctly shaped zero filled value. on read that
value is returned and on write its modified.
"""
def start(self):
......@@ -37,9 +38,7 @@ class TestClient:
self.connect()
def connect(self):
"""
this function provides a location for the code neccecary to connect to the client
"""
"""Connect to client"""
self.connected = True # set connected to true
def disconnect(self):
......@@ -49,12 +48,14 @@ class TestClient:
logger.debug("disconnected from the 'client' ")
def _setup_annotation(self, annotation):
"""
this function gives the client access to the comm client annotation data given to the attribute wrapper.
The annotation data can be used to provide whatever extra data is necessary in order to find/access the monitor/control point.
"""Provide additional data to monitor/control point
the annotation can be in whatever format may be required. it is up to the user to handle its content
example annotation may include:
this function gives the client access to the comm client annotation data given
to the attribute wrapper. The annotation data can be used to provide whatever
extra data is necessary in order to find/access the monitor/control point.
the annotation can be in whatever format may be required. it is up to the user
to handle its content, example annotations may include:
- a file path and file line/location
- server address
- IDs
......@@ -80,12 +81,10 @@ class TestClient:
return dims, dtype
def _setup_mapping(self, annotation, dims, dtype):
"""
takes all gathered data to configure and return the correct read and write functions
"""
"""Take gathered data to configure and return read and write functions"""
# we emulate that values written to annotations ending in _RW show up in their corresponding _R
# point as well
# we emulate that values written to annotations ending in _RW show up in their
# corresponding _R point as well
if annotation.endswith("_RW"):
annotation = annotation[:-1]
......@@ -114,14 +113,16 @@ class TestClient:
self.values[annotation] = write_value
logger.debug(
"created and bound example_client read/write functions to AttributeWrapper object"
"created and bound example_client read/write functions to AttributeWrapper "
"object"
)
return read_function, write_function
async def setup_attribute(self, annotation=None, attribute=None):
"""
MANDATORY function: is used by the attribute wrapper to get read/write functions.
must return the read and write functions
"""Return tuple with read and write functions
MANDATORY function: is used by the attribute wrapper to get read/write
functions. must return the read and write functions
"""
# process the comms_annotation
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment