"""Defines tango gql queries to use on rest based calls to the tango gql service."""
import json
from typing import Any, Dict, List, NamedTuple, Union
# ws messages and queries
[docs]def subscribe_device(device: str, attribute: str) -> str:
"""Return a subscription message for subscribing to a tango device attribute.
:param device: The device to subscribe to
:type device: str
:param attribute: The attribute to which must be subscribed to
:type attribute: str
:returns: The subscription message for subscribing to a tango device attribute.
"""
fullname = f"{device}/{attribute}"
query = f"""
subscription {{
attributes(fullNames: ["{fullname}"]) {{
device
attribute
value
quality
timestamp
}}
}}
"""
return query
# rest messages
[docs]def rest_info():
"""Return a message to query health status of tango gql.
:returns: The message to query health status of tango gql.
"""
return """
query{
info
}
"""
[docs]def load_all_attributes(device_name: str):
"""Return a message to load all attribute for a given device.
:param device_name: The tango device name
:type device_name: str
:returns: The message to load all attribute for a given device.
"""
return f"""
query{{
devices(pattern: "{device_name}") {{
attributes {{
name
datatype
}}
}}
}}
"""
[docs]def get_device_state(device_name: str):
"""Return a message to get state for a particular device.
:param device_name: The tango device name
:type device_name: str
:returns: The message to get state for a particular device.
"""
return f"""
query{{
devices(pattern: "{device_name}") {{
attributes(pattern: "state") {{
value
}}
}}
}}
"""
[docs]class GQLQuery(NamedTuple):
"""Bundles a gql query into a query (str) and its variables as a NamedTuple."""
query: str
variables: Union[Dict[Any, Any], None]
def __hash__(self) -> int:
"""Create a hash from the gql query.
:returns: the hash integer
"""
return hash((self.query, json.dumps(self.variables)))
COMMAND_QUERY = """
mutation ExecuteCommand($device: String!, $command: String!, $argin: ScalarTypes){
executeCommand(device: $device, command: $command, argin: $argin) {
ok
message
output
}
}
"""
WRITE_ATTR_QUERY = """
mutation WriteAttribute($device: String!, $attribute: String!, $value: ScalarTypes!) {
setAttributeValue(device: $device, name: $attribute, value: $value) {
ok
message
valueBefore
attribute {
device
name
value
writevalue
timestamp
quality
}
}
}
"""
FETCH_ATTRIBUTES = """
query FetchAttributeNames($device: String!) {
device(name: $device) {
attributes {
name
label
dataformat
datatype
}
}
}
"""
FETCH_ATTRIBUTES_VALUE = """
query FetchAttributeValues($fullNames: [String]!) {
attributes(fullNames: $fullNames) {
name
label
device
value
writevalue
timestamp
}
}
"""
FETCH_COMMANDS = """
query FetchCommandNames($device: String!) {
device(name: $device) {
commands {
name
intype
}
}
}
"""
[docs]def fetch_commands(device: str) -> GQLQuery:
"""Return a gql query for fetching all the commands from a given device.
:param device: The tango device name
:type device: str
:return: The gql query for fetching all the commands from a given device.
"""
variables = {"device": device}
return GQLQuery(FETCH_COMMANDS, variables)
[docs]def fetch_attributes(device: str) -> GQLQuery:
"""Return a gql query for fetching all the attributes for a given device.
:param device: The tango device name
:type device: str
:return: The gql query for fetching all the attributes for a given device.
"""
variables = {"device": device}
return GQLQuery(FETCH_ATTRIBUTES, variables)
[docs]def read_attribute(device: str, attr_name: str) -> GQLQuery:
"""Return a gql query for reading the value from a given attribute from a given device.
:param device: The tango device name
:type device: str
:param attr_name: The device attribute
:type attr_name: str
:return: The gql query for reading the value from a given attribute from a given device.
"""
fullnames = [f"{device}/{attr_name}"]
variables = {"fullNames": fullnames}
return GQLQuery(FETCH_ATTRIBUTES_VALUE, variables)
[docs]def command(device: str, command_name: str, argin: Any = None) -> GQLQuery:
"""Return a gql command for commanding a device with given input arguments.
:param device: The tango device name
:type device: str
:param command_name: The command name
:type command_name: str
:param argin: The argument to use for the command, defaults to None
:type argin: Any
:return: The gql command for commanding a device with given input arguments
"""
variables = {
"device": device,
"command": command_name,
"argin": argin,
}
return GQLQuery(COMMAND_QUERY, variables)
[docs]def write_attribute(device: str, attribute: str, value: Any):
"""Return a gql query for writing a given attribute value to a device.
:param device: The tango device name
:param attribute: The device attribute
:param value: The value to be written to the device attribute
:return: The gql query for setting an attribute
"""
variables = {
"device": device,
"attribute": attribute,
"value": value,
}
return GQLQuery(WRITE_ATTR_QUERY, variables)
[docs]def read_attributes_from_multiple_devices(
device_list: List[str], attr: Union[str, List[str]]
) -> GQLQuery:
"""Return a gql query for reading attribute/s from a given list of devices.
If the given attribute is in the form of a list then the attributes and devices,
are paired (zipped) as respective reads for each device.
:param device_list: The list of devices to be queried.
:type device_list: List[str]
:param attr: The attribute/s to be read for each device (a list of attributes is assumed to be
interpreted pair wise)
:type attr: Union[str, List[str]]
:return: The gql query for reading attribute/s from a given list of devices.
"""
if isinstance(attr, str):
fullnames = [f"{device_name}/{attr}" for device_name in device_list]
else:
attr_list = attr
fullnames = [
f"{device_name}/{device_attr}"
for device_name, device_attr in zip(device_list, attr_list)
]
variables = {"fullNames": fullnames}
return GQLQuery(FETCH_ATTRIBUTES_VALUE, variables)