Newer
Older
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
import json
import tango
from tango.server import Device, attribute, command, run, device_property
from tango import Attr, AttrWriteType, AttrQuality, DevState
from tango.server import attribute
# Type mapping for Tango data types
TANGO_TYPE_MAPPING = {
"DEV_BOOLEAN": tango.DevBoolean,
"DEV_SHORT": tango.DevShort,
"DEV_LONG": tango.DevLong,
"DEV_FLOAT": tango.DevFloat,
"DEV_DOUBLE": tango.DevDouble,
"DEV_STRING": tango.DevString,
}
class JSONGateway(Device):
# JSON property to define dynamic attributes
dynamic_attributes_config = device_property(dtype=(str,), default_value=("[]"))
dynamic_attributes = {}
attributes_config = {}
def init_device(self):
print("init_device")
"""Initialize the device and create dynamic attributes."""
super().init_device()
try:
# Parse the JSON configuration for dynamic attributes
self.attributes_config = json.loads(' '.join(self.dynamic_attributes_config))
except Exception as e:
print("ERROR: not valid JSON string in property dynamic_attributes_config.\nMust be:\n[\n {\"name\": \"name1\", \"type\": \"DEV_STRING\", \"writable\": true},\n {\"name\": \"name2\", \"type\": \"DEV_STRING\", \"writable\": false, \"value\": {\"device\":\"a/b/c\", \"operation\":\"command\", \"name\":\"Status\"}}\n]")
self.error_stream("Failed to initialize dynamic attributes: {e}")
try:
# Iterate through attribute definitions and create attributes
for attr_def in self.attributes_config:
print("attr_def: ", attr_def)
name = attr_def["name"]
dtype_str = attr_def["type"]
writable = attr_def.get("writable", False)
self.add_dynamic_attribute(name, dtype_str, writable)
except Exception as e:
print("Failed to initialize dynamic attributes:", e)
self.error_stream("Failed to initialize dynamic attributes: {e}")
# Define getter method
def read_callback(self, attr_obj):
name = attr_obj.get_name()
# print(name)
for attr_def in self.attributes_config:
if attr_def["name"]==name:
request = attr_def["value"]
target_device = request.get("device", None)
operation = request["operation"] # "command", "attribute", or "dynamic"
name = request["name"] # Command or attribute name
value = request.get("value") # Optional value for write operations
# Other operations: Command or Attribute
target_device = target_device or self.default_tango_device
device = tango.DeviceProxy(target_device)
if operation == "command":
result = device.command_inout(name, json.loads(value)) if value else device.command_inout(name)
elif operation == "attribute":
if value is not None:
# Write attribute
device.write_attribute(name, json.loads(value))
result = {"status": "success", "message": "Attribute {name} written successfully"}
else:
# Read attribute
result = device.read_attribute(name).value
else:
raise ValueError("Invalid operation type. Use 'command', 'attribute', or 'dynamic'.")
self.info_stream("Reading attribute '{name}': {value}")
# print(result)
attr_obj.set_value(json.dumps(result.tolist()))
# Define setter method (only if writable)
def write_callback(self, attr_obj):
# e.g. "{\"device\":\"g/gun/guntiming-p1.1\", \"operation\":\"command\", \"name\":\"Status\"}"
# "{\"device\":\"sr/sim/srElettra2_high_betax_long_straights\", \"operation\":\"command\", \"name\":\"GetHistExtendedOpticsData\", \"value\": \"[[0,100],[\\\"SR__DIAGNOSTICS__BPM_S04.09__horpos\\\"]]\"}"
value = attr_obj.get_write_value()
name = attr_obj.get_name()
for attr_def in self.attributes_config:
if attr_def["name"]==name:
try:
attr_def["value"] = json.loads(value)
except Exception as e:
raise ValueError("Invalid JSON string, please use syntax: {\"device\":\"a/b/c\", \"operation\":\"command\", \"name\":\"Status\"}")
if not "device" in attr_def["value"]:
raise ValueError("Missing device, please use syntax: {\"device\":\"a/b/c\", \"operation\":\"command\", \"name\":\"Status\"}")
if not "operation" in attr_def["value"]:
raise ValueError("Missing operation, please use syntax: {\"device\":\"a/b/c\", \"operation\":\"command\", \"name\":\"Status\"}")
if not "name" in attr_def["value"]:
raise ValueError("Missing name, please use syntax: {\"device\":\"a/b/c\", \"operation\":\"command\", \"name\":\"Status\"}")
# self.dynamic_attributes[name] = value
# self.info_stream("Writing attribute '{name}': {value}")
print("Writing attribute ", name, " : ", value)
def add_dynamic_attribute(self, name, dtype_str, writable):
"""Dynamically add an attribute to the device."""
dtype = TANGO_TYPE_MAPPING.get(dtype_str)
if dtype is None:
self.error_stream(f"Unsupported Tango data type: {dtype_str}")
return
# Define the attribute using the Attr class
attr = Attr(name, dtype, AttrWriteType.READ_WRITE if writable else AttrWriteType.READ)
# Register the attribute with the device
self.add_attribute(attr, self.read_callback, self.write_callback if writable else None)
# Initialize storage for the attribute value
self.dynamic_attributes[name] = None
# Command to handle JSON requests
@command(dtype_in=str, dtype_out=str)
def TransformJSON(self, json_request):
"""
Transforms a JSON request into a Tango command or attribute operation.
"""
try:
# Parse the JSON request
request = json.loads(json_request)
# Extract details
target_device = request.get("device", None)
operation = request["operation"] # "command", "attribute", or "dynamic"
name = request["name"] # Command or attribute name
value = request.get("value") # Optional value for write operations
if operation == "dynamic":
# Handle dynamic attributes
if name in self.dynamic_attributes:
if value is not None:
self.dynamic_attributes[name] = value
result = {"status": "success", "message": "Dynamic attribute {name} set to {value}"}
else:
result = self.dynamic_attributes[name]
else:
raise KeyError("Dynamic attribute '{name}' not found")
else:
# Other operations: Command or Attribute
target_device = target_device or self.default_tango_device
device = tango.DeviceProxy(target_device)
if operation == "command":
result = device.command_inout(name, value) if value else device.command_inout(name)
elif operation == "attribute":
if value is not None:
# Write attribute
device.write_attribute(name, value)
result = {"status": "success", "message": "Attribute {name} written successfully"}
else:
# Read attribute
result = device.read_attribute(name).value
else:
raise ValueError("Invalid operation type. Use 'command', 'attribute', or 'dynamic'.")
# Format response
response = {"status": "success", "result": result}
except Exception as e:
response = {"status": "error", "message": str(e)}
return json.dumps(response)
# Entry point for the device server
if __name__ == "__main__":
run([JSONGateway])