Add new file

json_gateway.py

+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])