import re
from enum import Enum
from typing import List, Sequence, Tuple, Union
from .. import __version__
[docs]def version_tag():
"""Tag with the current version of pymalcolm"""
return f"version:pymalcolm:{__version__}"
port_tag_re = re.compile(r"(source|sink)Port:(.*):(.*)")
[docs]class Port(Enum):
"""Enum with all the known flowgraph port tags to appear on Attribute
Metas"""
BOOL = "bool" #: Boolean value. Typically used in PandA
INT32 = "int32" #: 32-bit signed integer. Typically used in PandA
NDARRAY = "NDArray" #: areaDetector NDArray port
MOTOR = "motor" #: Motor record connection to CS or controller
BLOCK = "block" #: Malcolm level connection to another Block
[docs] def sink_port_tag(self, disconnected_value):
"""Add a tag indicating this is a Sink Port of the given type
Args:
disconnected_value: What value should the Attribute be set to
when the port is disconnected
"""
return f"sinkPort:{self.value}:{disconnected_value}"
[docs] def source_port_tag(self, connected_value):
"""Add a tag indicating this is a Source Port of the given type
Args:
connected_value: What value should a Sink Port be set to if
it is connected to this port
"""
return f"sourcePort:{self.value}:{connected_value}"
[docs] def with_source_port_tag(self, tags, connected_value):
"""Add a Source Port tag to the tags list, removing any other Source
Ports"""
new_tags = [t for t in tags if not t.startswith("sourcePort:")]
new_tags.append(self.source_port_tag(connected_value))
return new_tags
[docs] @classmethod
def port_tag_details(
cls, tags: Sequence[str]
) -> Union[Tuple[bool, "Port", str], None]:
"""Search tags for port info, returning it
Args:
tags: A list of tags to check
Returns:
None or (is_source, port, connected_value|disconnected_value)
where port is one of the Enum entries of Port
"""
for tag in tags:
match = port_tag_re.match(tag)
if match:
source_sink, port, extra = match.groups()
return source_sink == "source", cls(port), extra
return None
[docs]def group_tag(group_name: str) -> str:
"""Marks this field as belonging to a group"""
tag = f"group:{group_name}"
return tag
[docs]def linked_value_tag(mri: str, attribute_name: str) -> str:
"""Marks this field as having another attribute in another Block that
should be displayed below it as a linked value"""
tag = f"linkedvalue:{attribute_name}:{mri}"
return tag
[docs]def badge_value_tag(mri: str, attribute_name: str, display: str = "plus") -> str:
"""Marks this Port as having another attribute in another Block that
should be displayed as its badge.
Args:
mri: The mri of the Block that provides the badge value
attribute_name: The attribute name in the Block given by mri that
provides the badge value
display: The type of formatting to apply to the badge value:
"plus": If value < 1 then hide, otherwise display +%d
"""
tag = f"badgevalue:{display}:{attribute_name}:{mri}"
return tag
[docs]def config_tag(iteration: int = 1) -> str:
"""Marks this field as a value that should be saved and loaded at config
Args:
iteration: All iterations are sorted in increasing order and done in
batches of the same iteration number
"""
tag = f"config:{iteration:d}"
return tag
[docs]def get_config_tag(tags: Sequence[str]) -> Union[str, None]:
"""Get the config_tag from tags or return None"""
for tag in tags:
if tag.startswith("config:"):
return tag
return None
[docs]def method_return_unpacked():
"""This method has a single element returns, and when called will return
this single element unpacked rather than in a single element map
E.g.
hello.greet("me") -> "Hello me" not {"return": "Hello me"}
"""
tag = "method:return:unpacked"
return tag
def method_hidden():
"""Hide this method on the GUI"""
tag = "method:hidden"
return tag