1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
|
# SPDX-FileCopyrightText: All Contributors to the PyTango project
# SPDX-License-Identifier: LGPL-3.0-or-later
__all__ = ("PipeConfig",)
from tango._tango import (
Pipe,
PipeWriteType,
UserDefaultPipeProp,
CmdArgType,
DevState,
DispLevel,
)
from tango.utils import (
scalar_to_array_type,
TO_TANGO_TYPE,
is_non_str_seq,
is_pure_str,
is_integer,
is_number,
)
from tango.utils import document_method as __document_method
class PipeConfig:
"""
This class represents the python interface for the Tango IDL
object PipeConfig."""
def __init__(self):
self.name = ""
self.description = ""
self.label = ""
self.level = DispLevel.OPERATOR
self.writable = PipeWriteType.PIPE_READ
self.extensions = []
def __get_pipe_type_simple(obj):
if is_non_str_seq(obj):
if (
len(obj) == 2
and is_pure_str(obj[0])
and (is_non_str_seq(obj[1]) or isinstance(obj[1], dict))
):
tg_type = CmdArgType.DevPipeBlob
else:
tg_type = __get_pipe_type(obj[0])
tg_type = scalar_to_array_type(tg_type)
elif is_pure_str(obj):
tg_type = CmdArgType.DevString
elif isinstance(obj, DevState):
tg_type = CmdArgType.DevState
elif isinstance(obj, bool):
tg_type = CmdArgType.DevBoolean
elif is_integer(obj):
tg_type = CmdArgType.DevLong64
elif is_number(obj):
tg_type = CmdArgType.DevDouble
else:
raise ValueError("Cannot determine object tango type")
return tg_type
def __get_tango_type(dtype):
if is_non_str_seq(dtype):
tg_type = dtype[0]
if is_non_str_seq(tg_type):
raise TypeError("Pipe doesn't support 2D data")
tg_type = TO_TANGO_TYPE[tg_type]
tg_type = scalar_to_array_type(tg_type)
else:
tg_type = TO_TANGO_TYPE[dtype]
return tg_type
def __get_pipe_type(obj, dtype=None):
if dtype is not None:
return __get_tango_type(dtype)
try:
ndim, dtype = obj.ndim, str(obj.dtype)
except AttributeError:
return __get_pipe_type_simple(obj)
if ndim > 1:
raise TypeError(
f"cannot translate numpy array with {obj.ndim} " f"dimensions to tango type"
)
tg_type = TO_TANGO_TYPE[dtype]
if ndim > 0:
tg_type = scalar_to_array_type(dtype)
return tg_type
def __sanatize_pipe_element(elem):
if isinstance(elem, dict):
result = dict(elem)
else:
result = dict(name=elem[0], value=elem[1])
result["value"] = value = result.get("value", result.pop("blob", None))
result["dtype"] = dtype = __get_pipe_type(value, dtype=result.get("dtype"))
if dtype == CmdArgType.DevPipeBlob:
result["value"] = value[0], __sanatize_pipe_blob(value[1])
return result
def __sanatize_pipe_blob(blob):
if isinstance(blob, dict):
return [__sanatize_pipe_element((k, v)) for k, v in blob.items()]
else:
return [__sanatize_pipe_element(elem) for elem in blob]
def __Pipe__set_value(self, value):
"""
Set the pipe value. Check ref:`pipe data types <pytango-pipe-data-types>`:
for more information on which values are supported
"""
root_blob_name, blob = value
self.set_root_blob_name(root_blob_name)
value = __sanatize_pipe_blob(blob)
self._set_value(value)
def __Pipe__get_value(self):
return (self.get_root_blob_name(), self._get_value())
def __init_Pipe():
Pipe.set_value = __Pipe__set_value
def __doc_UserDefaultPipeProp():
def document_method(method_name, desc, append=True):
return __document_method(UserDefaultPipeProp, method_name, desc, append)
UserDefaultPipeProp.__doc__ = """
User class to set pipe default properties.
This class is used to set pipe default properties.
Three levels of pipe properties setting are implemented within Tango.
The highest property setting level is the database.
Then the user default (set using this UserDefaultPipeProp class) and finally
a Tango library default value
"""
document_method(
"set_label",
"""
set_label(self, def_label) -> None
Set default label property.
Parameters :
- def_label : (str) the user default label property
Return : None
""",
)
document_method(
"set_description",
"""
set_description(self, def_description) -> None
Set default description property.
Parameters :
- def_description : (str) the user default description property
Return : None
""",
)
def pipe_init(doc=True):
__init_Pipe()
if doc:
__doc_UserDefaultPipeProp()
|
