"""Tango type definitions for SDP config entity models."""
from __future__ import annotations
import re
from typing import Annotated, Any
from pydantic import (
AnyUrl,
GetCoreSchemaHandler,
GetJsonSchemaHandler,
TypeAdapter,
UrlConstraints,
)
from pydantic.json_schema import JsonSchemaValue
from pydantic_core import Url, core_schema
from ska_sdp_config.entity.common.tango_trl import TRL, TRL_RE, AnyTRL
# flake8: noqa
# pylint: disable=line-too-long
TANGO_URL_RE = re.compile(
r"^([a-zA-Z][a-zA-Z0-9+.-]*):" # protocol:
r"(?://([A-Za-z0-9.-]*)(?::(\d+))?)?" # //[host[:port]]/ optional, host can be empty
r"(/([A-Za-z0-9_-]+)/([A-Za-z0-9_-]+)/([A-Za-z0-9_-]+))" # device-name (required: 3-part)
r"(?:/([A-Za-z0-9_.]+?))?" # /attribute (optional, non-greedy)
r"(?:(?:->|-%3E)([A-Za-z0-9_.-]+))?" # ->property or %3Eproperty (optional)
r"(?:#dbase=(yes|no))?$" # #dbase=xx (optional)
)
TANGO_ATTRIBUTE_URL_RE = re.compile(
r"^([a-zA-Z][a-zA-Z0-9+.-]*):" # protocol:
r"(?://([A-Za-z0-9.-]*)(?::(\d+))?)?" # //[host[:port]]/ optional, host can be empty
r"(/([A-Za-z0-9_-]+)/([A-Za-z0-9_-]+)/([A-Za-z0-9_-]+))" # device-name (required: 3-part)
r"(?:/([A-Za-z0-9_.]+?))" # /attribute (required)
r"(?:(?:->|-%3E)([A-Za-z0-9_.-]+))?" # ->property or %3Eproperty (optional)
r"(?:#dbase=(yes|no))?$" # #dbase=xx (optional)
)
[docs]
class TangoUrl(AnyUrl):
"""Tango URL compliant to RFC 3986 parsers.
Supported input expressions are of the form:
``tango://<host>[:<port>]/<device_domain>/<device_family>/<device_member>/[<device_attribute>][#dbase=yes|no]``
Defaults:
- port=10000
- dbase=yes
Pre-conditions:
- host required if dbase=yes
Additionally supports construction from a TRL.
For more info see:
https://tango-controls.readthedocs.io/en/9.2.5/manual/C-naming.html
"""
# pylint: disable=protected-access
_constraints = UrlConstraints(allowed_schemes=["tango"])
@property
def domain_name(self) -> str:
"""The device family part of the URL."""
matches = TANGO_URL_RE.match(str(self))
assert matches is not None
return matches[5]
@property
def family_name(self) -> str:
"""The device family part of the URL."""
matches = TANGO_URL_RE.match(str(self))
assert matches is not None
return matches[6]
@property
def member_name(self) -> str:
"""The device member part of the URL."""
matches = TANGO_URL_RE.match(str(self))
assert matches is not None
return matches[7]
@property
def attribute_name(self) -> str:
"""The device name part of the URL."""
matches = TANGO_URL_RE.match(str(self))
assert matches is not None
return matches[8]
@property
def property_name(self) -> str:
"""The Tango attribute name part of the TRL, or `None`."""
matches = TANGO_URL_RE.match(str(self))
assert matches is not None
return matches[9]
@property
def dbase(self) -> bool:
"""The using database flag. `True` for 'dbase=yes'."""
matches = TANGO_URL_RE.match(str(self))
assert matches is not None
return matches[10] != "no"
@property
def device_name(self) -> str:
"""The Tango device name part of the TRL."""
matches = TANGO_URL_RE.match(str(self))
assert matches is not None
return f"{matches[5]}/{matches[6]}/{matches[7]}"
@property
def device_url(self) -> TangoUrl:
"""The device URL."""
matches = TANGO_URL_RE.match(str(self))
assert matches is not None
assert matches[1] is not None
return TangoUrl.build(
scheme=self.scheme,
host=self.host,
port=self.port,
path=self.device_name,
fragment=self.fragment,
)
[docs]
@classmethod
def validate_url(
cls,
value: AnyTRL | AnyUrl | str,
handler: core_schema.ValidatorFunctionWrapHandler,
) -> TangoUrl:
"""
Wrap validate a URL string or TRL instance to a Tango URL.
"""
# handle TRL by explicitly converting to URL format
if isinstance(value, (AnyTRL, TRL)):
value = value.url
instance = cls.__new__(cls)
if matches := TANGO_URL_RE.match(str(value)):
instance._url = Url(handler(str(value)))
dbase_required = matches[10] != "no"
elif matches := TRL_RE.match(str(value)):
instance._url = TRL(handler(str(value))).url
dbase_required = matches[7] != "no"
else:
raise AssertionError(
f"No TangoUrl pattern match {TANGO_URL_RE.pattern}"
)
# if using str_schema instead of url_schema, must seperately
# validate url constraints
if isinstance(value, str):
# network resource authority are usually never optional (RFC 3986 3.2).
# either required (http:, ftp: etc.), or always None (file:, mailto:)
RuntimeTangoUrl = TypeAdapter( # pylint: disable=invalid-name
Annotated[
AnyUrl,
UrlConstraints(
allowed_schemes=["tango"], host_required=dbase_required
),
]
)
RuntimeTangoUrl.validate_python(instance._url)
return instance
@classmethod
def __get_pydantic_core_schema__(
cls, _source: Any, handler: GetCoreSchemaHandler
) -> core_schema.CoreSchema:
schema = core_schema.no_info_wrap_validator_function(
cls.validate_url,
schema=core_schema.union_schema(
[
core_schema.url_schema(
**cls._constraints.defined_constraints,
),
core_schema.str_schema(),
]
),
serialization=core_schema.plain_serializer_function_ser_schema(
lambda u: u._url
),
)
return handler(schema)
@classmethod
def __get_pydantic_json_schema__(
cls,
_core_schema: core_schema.CoreSchema,
_handler: GetJsonSchemaHandler,
) -> JsonSchemaValue:
return {
"format": "uri",
"minLength": 1,
"type": "string",
"pattern": TANGO_URL_RE.pattern,
}
[docs]
class TangoAttributeUrl(TangoUrl):
"""
Tango URL to an attribute on a device instance.
Supported input expressions are of the form:
``tango://<host>[:<port>]/<device_domain>/<device_family>/<device_member>/<device_attribute>[#dbase=yes|no]``
Defaults:
- port=10000
- dbase=yes
Pre-conditions:
- host required if dbase=yes
For more information see:
https://tango-controls.readthedocs.io/en/9.2.5/manual/C-naming.html
"""
# noqa: E501 # pylint: disable=line-too-long
# pylint: disable=protected-access
@classmethod
def __get_pydantic_core_schema__(
cls, _source: Any, handler: GetCoreSchemaHandler
) -> core_schema.CoreSchema:
return core_schema.no_info_wrap_validator_function(
cls.validate_url, schema=handler.generate_schema(TangoUrl)
)
[docs]
@classmethod
def validate_url(
cls,
value: AnyUrl | TRL | str,
handler: core_schema.ValidatorFunctionWrapHandler,
) -> TangoAttributeUrl:
"""
Wrap validate a URL string or TRL instance to a Tango attribute URL.
"""
if isinstance(value, (AnyTRL, TRL)):
value = value.url
url = TangoUrl(handler(str(value)))
if url.attribute_name is None:
raise ValueError("Attribute missing")
if url.property_name is not None:
raise ValueError("Property not supported")
instance = cls.__new__(cls)
instance._url = url._url
return instance
@classmethod
def __get_pydantic_json_schema__(
cls,
_core_schema: core_schema.CoreSchema,
_handler: GetJsonSchemaHandler,
) -> JsonSchemaValue:
return {
"format": "uri",
"minLength": 1,
"type": "string",
"pattern": TANGO_ATTRIBUTE_URL_RE.pattern,
}