Attributes In Expressions

qtoggleserver/core/expressions/__init__.py

@@ -11,6 +11,7 @@
     Expression,
     Role,
 )
+from .exceptions import EmptyExpression
 from .literalvalues import LiteralValue
 
 
@@ -52,14 +53,20 @@ def parse(self_port_id: str | None, sexpression: str, role: Role, pos: int = 1)
     pos += len(sexpression) - len(stripped)
     sexpression = stripped.rstrip()
 
-    if sexpression and sexpression[0] in ("$", "@"):
+    if not sexpression:
+        raise EmptyExpression()
+
+    if sexpression[0] in ("$", "@"):
         return PortExpression.parse(self_port_id, sexpression, role, pos)
+    elif sexpression[0] == "#":
+        return DeviceExpression.parse(self_port_id, sexpression, role, pos)
     elif "(" in sexpression or ")" in sexpression:
         return Function.parse(self_port_id, sexpression, role, pos)
     else:
         return LiteralValue.parse(self_port_id, sexpression, role, pos)
 
 
+from .devices import DeviceExpression  # noqa: E402
 from .functions import (  # noqa: E402
     Function,  # noqa: E402
     aggregation,

qtoggleserver/core/expressions/base.py

@@ -53,6 +53,18 @@ async def _eval(self, context: EvalContext) -> EvalResult:
         raise NotImplementedError()
 
     def get_deps(self) -> set[str]:
+        """
+        Return a set with all dependencies of this expression.
+        Each dependency is a string and, depending on its format denotes a different type of dependency:
+         - If string starts with `$`, the dep is a port value or port attribute dependency; going further, we have:
+            - `$port_id` - dependency on the corresponding port's value
+            - `$port_id:` - dependency on the corresponding port's attributes
+         - If string starts with `#`, the dep is a device attribute dependency; going further, we have:
+            - `#:` - dependency on (main) device attributes
+            - `#slave_name:` - dependency on the corresponding slave's attributes
+         - One of the `DEP_*` constants (e.g. `DEP_YEAR` or `"year"`) indicates a time dependency
+        """
+
         if self._cached_deps is None:
             self._cached_deps = self._get_deps()
         return self._cached_deps

qtoggleserver/core/expressions/devices.py

@@ -0,0 +1,75 @@
+from __future__ import annotations
+
+import abc
+import re
+
+from .base import EvalContext, EvalResult, Expression, Role
+from .exceptions import DeviceAttrUnavailable, MissingAttrPrefix, TransformNotSupported, UnexpectedCharacter
+
+
+_TRANSFORM_ROLES = (Role.TRANSFORM_READ, Role.TRANSFORM_WRITE)
+
+
+class DeviceExpression(Expression, metaclass=abc.ABCMeta):
+    def __init__(self, device_name: str | None, prefix: str, role: Role, attr_name: str | None = None) -> None:
+        super().__init__(role)
+
+        self.device_name: str | None = device_name
+        self.attr_name: str | None = attr_name
+        self.prefix: str = prefix
+
+    @staticmethod
+    def parse(self_port_id: str | None, sexpression: str, role: Role, pos: int) -> Expression:
+        stripped = sexpression.lstrip()
+        pos += len(sexpression) - len(stripped)
+        sexpression = stripped.rstrip()
+
+        prefix = sexpression[0]
+        sub_sexpression = sexpression[1:]
+
+        if prefix == "#":
+            if role in _TRANSFORM_ROLES:
+                raise TransformNotSupported(sexpression, pos)
+            parts = sub_sexpression.split(":", 1)
+            if len(parts) != 2:
+                raise MissingAttrPrefix(pos + len(sub_sexpression) + 1)
+
+            device_name, attr_name = parts
+            if device_name:
+                m = re.search(r"[^a-zA-Z0-9_-]", device_name)
+                if m:
+                    p = m.start()
+                    raise UnexpectedCharacter(device_name[p], p + pos + 3)
+
+                return SlaveDeviceAttr(device_name, prefix, role, attr_name)
+            else:
+                return MainDeviceAttr(prefix, role, attr_name)
+        else:
+            raise UnexpectedCharacter(prefix, pos)
+
+
+class DeviceAttr(DeviceExpression):
+    def __str__(self) -> str:
+        return f"{self.prefix}{self.device_name or ''}:{self.attr_name}"
+
+    def _get_deps(self) -> set[str]:
+        return {f"#{self.device_name or ''}:"}
+
+    async def _eval(self, context: EvalContext) -> EvalResult:
+        key = f"{self.device_name}:{self.attr_name or ''}" if self.device_name else self.attr_name
+        value = context.device_attrs.get(key)
+        if value is None:
+            raise DeviceAttrUnavailable(self.device_name or "", self.attr_name or "")
+        if not isinstance(value, (int, float)):  # this includes `bool`
+            value = int(bool(value))
+
+        return value
+
+
+class SlaveDeviceAttr(DeviceAttr):
+    pass
+
+
+class MainDeviceAttr(DeviceAttr):
+    def __init__(self, prefix: str, role: Role, attr_name: str | None = None) -> None:
+        super().__init__(None, prefix, role, attr_name)

qtoggleserver/core/expressions/exceptions.py

@@ -63,17 +63,6 @@ def to_json(self) -> GenericJSONDict:
         return {"reason": "unexpected-end"}
 
 
-class NonSelfDependency(ExpressionParseError):
-    def __init__(self, port_id: str, pos: int) -> None:
-        self.port_id = port_id
-        self.pos = pos
-
-        super().__init__(f'Non-self dependency on port "{port_id}"')
-
-    def to_json(self) -> GenericJSONDict:
-        return {"reason": "non-self-dependency", "token": self.port_id, "pos": self.pos}
-
-
 class UnexpectedCharacter(ExpressionParseError):
     def __init__(self, c: str, pos: int) -> None:
         self.c = c
@@ -93,6 +82,27 @@ def to_json(self) -> GenericJSONDict:
         return {"reason": "empty"}
 
 
+class MissingAttrPrefix(ExpressionParseError):
+    def __init__(self, pos: int) -> None:
+        self.pos: int = pos
+
+        super().__init__("Missing attribute prefix")
+
+    def to_json(self) -> GenericJSONDict:
+        return {"reason": "missing-attr-prefix", "pos": self.pos}
+
+
+class TransformNotSupported(ExpressionParseError):
+    def __init__(self, token: str, pos: int) -> None:
+        self.token: str = token
+        self.pos: int = pos
+
+        super().__init__(f'Expression "{token}" is not supported in transform expressions')
+
+    def to_json(self) -> GenericJSONDict:
+        return {"reason": "transform-not-supported", "token": self.token, "pos": self.pos}
+
+
 class ExpressionEvalException(ExpressionException):
     pass
 
@@ -127,6 +137,26 @@ class DisabledPort(PortValueUnavailable):
     MSG = 'Port "%s" is disabled'
 
 
+class PortAttrUnavailable(ValueUnavailable):
+    MSG = 'Port attribute "%s:%s" is unavailable'
+
+    def __init__(self, port_id: str, attr_name: str) -> None:
+        self.port_id = port_id
+        self.attr_name = attr_name
+
+        super().__init__(self.MSG % (port_id, attr_name))
+
+
+class DeviceAttrUnavailable(ValueUnavailable):
+    MSG = 'Device attribute "%s:%s" is unavailable'
+
+    def __init__(self, device_name: str, attr_name: str) -> None:
+        self.device_name = device_name
+        self.attr_name = attr_name
+
+        super().__init__(self.MSG % (device_name, attr_name))
+
+
 class ExpressionArithmeticError(ExpressionEvalException):
     def __init__(self) -> None:
         super().__init__("Expression arithmetic error")

qtoggleserver/core/expressions/functions/__init__.py

@@ -6,8 +6,9 @@
 
 from .. import DEP_ASAP, exceptions, parse
 from ..base import EvalContext, EvalResult, Expression, Role
+from ..devices import DeviceAttr
 from ..literalvalues import LiteralValue
-from ..ports import PortValue
+from ..ports import PortAttr, PortValue
 
 
 FUNCTIONS = {}
@@ -73,7 +74,7 @@ def validate_arg_kinds(cls, args: list[Expression], pos_list: list[int]) -> None
             try:
                 kind = cls.ARG_KINDS[i]
             except IndexError:
-                kind = (LiteralValue, PortValue, Function)
+                kind = (LiteralValue, PortValue, Function, PortAttr, DeviceAttr)
 
             if not isinstance(arg, kind):
                 raise exceptions.InvalidArgumentKind(cls.NAME, pos_list[i], i + 1)

qtoggleserver/core/expressions/ports.py

@@ -5,18 +5,30 @@
 from qtoggleserver.core.typing import NullablePortValue
 
 from .base import EvalContext, EvalResult, Expression, Role
-from .exceptions import DisabledPort, PortValueUnavailable, UnexpectedCharacter, UnknownPortId
+from .exceptions import (
+    DisabledPort,
+    PortAttrUnavailable,
+    PortValueUnavailable,
+    TransformNotSupported,
+    UnexpectedCharacter,
+    UnknownPortId,
+)
+
+
+_TRANSFORM_ROLES = (Role.TRANSFORM_READ, Role.TRANSFORM_WRITE)
 
 
 class PortExpression(Expression, metaclass=abc.ABCMeta):
-    def __init__(self, port_id: str, prefix: str, role: Role) -> None:
+    def __init__(self, port_id: str, prefix: str, role: Role, attr_name: str | None = None) -> None:
         super().__init__(role)
 
         self.port_id: str = port_id
+        self.attr_name: str | None = attr_name
         self.prefix: str = prefix
         self._cached_port: core_ports.BasePort | None = None
 
     def get_port(self) -> core_ports.BasePort | None:
+        # TODO: test what happens if a port is removed; do we need this `is_removed` check?
         port = self._cached_port
         if port is None or port.is_removed():
             port = core_ports.get(self.port_id)
@@ -30,23 +42,56 @@ def parse(self_port_id: str | None, sexpression: str, role: Role, pos: int) -> E
         sexpression = stripped.rstrip()
 
         prefix = sexpression[0]
-        port_id = sexpression[1:]
-
-        if port_id:
-            m = re.search(r"[^a-zA-Z0-9_.-]", port_id)
-            if m:
-                p = m.start()
-                raise UnexpectedCharacter(port_id[p], p + pos + 2)
-
-            if prefix == "$":
-                return PortValue(port_id, prefix, role)
-            else:  # assuming prefix == '@'
-                return PortRef(port_id, prefix, role)
+        sub_sexpression = sexpression[1:]
+
+        if sub_sexpression:
+            parts = sub_sexpression.split(":", 1)
+            if len(parts) == 2:  # port attribute
+                port_id, attr_name = parts
+                m = re.search(r"[^a-zA-Z0-9_.-]", port_id)
+                if m:
+                    p = m.start()
+                    raise UnexpectedCharacter(port_id[p], p + pos + 2)
+                m = re.search(r"[^a-zA-Z0-9_-]", attr_name)
+                if m:
+                    p = m.start()
+                    raise UnexpectedCharacter(attr_name[p], p + pos + len(port_id) + 2)
+
+                if prefix == "$":
+                    if role in _TRANSFORM_ROLES:
+                        raise TransformNotSupported(sexpression, pos)
+                    if port_id:
+                        return PortAttr(port_id, prefix, role, attr_name)
+                    else:
+                        return SelfPortAttr(self_port_id, prefix, role, attr_name)
+                else:
+                    raise UnexpectedCharacter(prefix, pos)
+            else:
+                port_id = sub_sexpression
+                m = re.search(r"[^a-zA-Z0-9_.-]", port_id)
+                if m:
+                    p = m.start()
+                    raise UnexpectedCharacter(sub_sexpression[p], p + pos + 2)
+
+                if prefix == "$":
+                    if role in _TRANSFORM_ROLES and port_id != self_port_id:
+                        raise TransformNotSupported(sexpression, pos=pos)
+                    return PortValue(port_id, prefix, role)
+                elif prefix == "@":
+                    if role in _TRANSFORM_ROLES:
+                        raise TransformNotSupported(sexpression, pos)
+                    return PortRef(port_id, prefix, role)
+                else:
+                    raise UnexpectedCharacter(prefix, pos)
         else:
             if prefix == "$":
                 return SelfPortValue(self_port_id, prefix, role)
-            else:  # assuming prefix == '@'
+            elif prefix == "@":
+                if role in _TRANSFORM_ROLES:
+                    raise TransformNotSupported(sexpression, pos)
                 return SelfPortRef(self_port_id, prefix, role)
+            else:
+                raise UnexpectedCharacter(prefix, pos)
 
 
 class PortValue(PortExpression):
@@ -94,3 +139,29 @@ async def _eval(self, context: EvalContext) -> EvalResult:
 class SelfPortRef(PortRef):
     def __str__(self) -> str:
         return self.prefix
+
+
+class PortAttr(PortExpression):
+    def __str__(self) -> str:
+        return f"{self.prefix}{self.port_id}:{self.attr_name}"
+
+    def _get_deps(self) -> set[str]:
+        return {f"${self.port_id}:"}
+
+    async def _eval(self, context: EvalContext) -> EvalResult:
+        port = self.get_port()
+        if not port:
+            raise UnknownPortId(self.port_id)
+
+        value = context.port_attrs.get(self.port_id, {}).get(self.attr_name)
+        if value is None:
+            raise PortAttrUnavailable(self.port_id, self.attr_name or "")
+        if not isinstance(value, (int, float)):  # this includes `bool`
+            value = int(bool(value))
+
+        return value
+
+
+class SelfPortAttr(PortAttr):
+    def __str__(self) -> str:
+        return f"{self.prefix}:{self.attr_name}"