from __future__ import annotations
import warnings
from datetime import datetime, timedelta
from typing import Any, Generic, Optional, TypeVar, Union
from uuid import UUID, uuid4
from pydantic.dataclasses import dataclass
from pydantic.fields import Field
from typing_extensions import Annotated, Self, deprecated, override
from kelvin.krn import KRN, KRNAppVersion, KRNAsset, KRNAssetDataStream, KRNAssetParameter, KRNDatastream
from kelvin.message.base_messages import (
ControlChangeAck,
ControlChangeAckPayload,
ControlChangeMsg,
ControlChangePayload,
ControlChangeQueueing,
CustomActionFailureReason,
CustomActionMsg,
CustomActionPayload,
CustomActionReportMsg,
CustomActionReportPayload,
CustomActionResultMsg,
CustomActionResultPayload,
CustomActionState,
CustomActionStatusLog,
DataTagMsg,
DataTagPayload,
EdgeParameter,
InsightAcceptance,
ParametersMsg,
ParametersPayload,
ParameterType,
QueueingMode,
RecommendationActions,
RecommendationControlChange,
RecommendationCustomAction,
RecommendationMsg,
RecommendationPayload,
ReportedValues,
ResourceParameters,
StateEnum,
ValuePoint,
)
from kelvin.message.evidences import BaseEvidence, Evidence
from kelvin.message.message import Message
from kelvin.message.msg_type import KMessageTypeAction
TMessage = TypeVar("TMessage", bound=Message)
TBuilder = TypeVar("TBuilder", bound="MessageBuilder[Any]")
class MessageBuilder(Generic[TMessage]):
_msg: Optional[TMessage] = None
def to_message(self) -> TMessage:
return NotImplemented
@classmethod
def from_message(cls: type[TBuilder], msg: TMessage) -> TBuilder: # pyright: ignore[reportUnusedParameter]
return NotImplemented
[docs]
@dataclass
class ControlChange(MessageBuilder[ControlChangeMsg]):
"""Control change builder.
Use this helper class to build a Kelvin Control change.
Exactly one of ``expiration_date`` or ``expiration_delta`` must be provided
when the control change is used standalone. When used inside a
:class:`Recommendation`, both may be omitted — the control change will
inherit the recommendation's ``expiration_date``.
Args:
resource (KRN): The kelvin resource targeted by the control change, represented by a KRN (usually
KRNAssetDataStream)
payload (bool, int, float, str): The desired target value for the control change
expiration_date (Optional[datetime | timedelta]): The absolute time in the future when the Control Change
expires. Provide either an absolute datetime or a timedelta from now. Mutually exclusive with
expiration_delta.
expiration_delta (Optional[timedelta]): Expiration delta interval. Mutually exclusive with expiration_date.
When used in a standalone control change, it is converted to an absolute datetime. When used inside a
recommendation, it is passed through as seconds to the backend.
retries (int): Optional number of retries
retry_interval (int): Optional timeout time (for retries)
control_change_id (UUID): Optional UUID to set an specific ID for the control change
from_value (ValuePoint): Optional initial value for the control change
trace_id (str): Optional trace ID for the control change
queueing (ControlChangeQueueing): Optional queueing behaviour for the control change
"""
resource: KRN
payload: Any
expiration_date: Optional[Union[datetime, timedelta]] = None
expiration_delta: Optional[timedelta] = Field(
default=None, description="Expiration delta in seconds", gt=timedelta(seconds=0)
)
retries: Optional[int] = None
timeout: Annotated[Optional[int], Field(deprecated="Field 'timeout' is deprecated")] = None
retry_interval: Optional[int] = None
control_change_id: Optional[UUID] = None
from_value: Optional[ValuePoint] = None
trace_id: Optional[str] = None
queueing: Optional[ControlChangeQueueing] = None
def __post_init__(self) -> None:
if self.expiration_date is not None and self.expiration_delta is not None:
raise ValueError("Cannot set both 'expiration_date' and 'expiration_delta'")
if self.timeout is not None and self.retry_interval is not None:
raise ValueError("Cannot set both 'timeout' and 'retry_interval'")
[docs]
@override
def to_message(self) -> ControlChangeMsg:
if self.expiration_delta is not None:
expiration = datetime.now() + self.expiration_delta
elif isinstance(self.expiration_date, datetime):
expiration = self.expiration_date
elif isinstance(self.expiration_date, timedelta):
expiration = datetime.now() + self.expiration_date
else:
raise ValueError(
"Either 'expiration_date' or 'expiration_delta' must be set for a standalone control change"
)
return ControlChangeMsg(
id=self.control_change_id if self.control_change_id is not None else uuid4(),
trace_id=self.trace_id,
resource=self.resource,
payload=ControlChangePayload(
timeout=self.timeout if self.timeout is not None else self.retry_interval,
retries=self.retries,
expiration_date=expiration,
payload=self.payload,
from_value=self.from_value, # type: ignore[assignment]
queueing=QueueingMode(mode=self.queueing) if self.queueing is not None else None,
),
)
[docs]
@classmethod
@override
def from_message(cls, msg: ControlChangeMsg) -> ControlChange:
obj = cls(
resource=msg.resource, # type: ignore[arg-type]
expiration_date=msg.payload.expiration_date,
payload=msg.payload.payload,
retries=msg.payload.retries,
retry_interval=msg.payload.timeout,
control_change_id=msg.id,
from_value=msg.payload.from_value,
trace_id=msg.trace_id,
queueing=msg.payload.queueing.mode if msg.payload.queueing is not None else None,
)
obj._msg = msg
return obj
[docs]
@dataclass
class Recommendation(MessageBuilder[RecommendationMsg]):
"""Recommendation Builder. Use this helper class to build a Kelvin Recommendation.
Args:
resource (KRNAsset): The kelvin asset resource targeted by the recommendation
type (str): the type of the recommendation, chose one from the available on the kelvin platform (eg generic,
speed_inc, speed_dec, ...)
expiration_date (datetime | timedelta): The absolute time in the future when the recommendation expires. Provide
either a absolute datetime or a timedelta from now
description (str): An optional description for the recommendation
confidence (int): Optional confidence of the recommendation (from 1 to 4)
control_changes (list[ControlChanges]): the list of ControlChanges associated with the recommendation
metadata: (dict[str, Any]): Optional metadata for the recommendation
auto_accepted (bool): Sets the Recommendation as auto accepted. Default is False
evidences (list[Evidence]): List of evidences associated with the recommendation
custom_identifier (str): Optional custom identifier for the recommendation
actions (list[CustomAction]): List of custom actions associated with the recommendation
trace_id (str): Optional trace ID for the recommendation
"""
resource: KRNAsset
type: str
expiration_date: Optional[Union[datetime, timedelta]] = None
description: Optional[str] = None
confidence: Optional[int] = Field(default=None, ge=1, le=4)
control_changes: list[ControlChange] = Field(default_factory=list)
metadata: Optional[dict[str, Any]] = None
auto_accepted: bool = False
evidences: list[Evidence] = Field(default_factory=list)
custom_identifier: Optional[str] = None
actions: list[CustomAction] = Field(default_factory=list)
trace_id: Optional[str] = None
[docs]
@override
def to_message(self) -> RecommendationMsg:
now = datetime.now()
if self.expiration_date is None:
rec_expiration_date = None
elif isinstance(self.expiration_date, datetime):
rec_expiration_date = self.expiration_date
else:
rec_expiration_date = now + self.expiration_date
ccs: list[RecommendationControlChange] = []
for cc in self.control_changes:
if cc.expiration_delta is not None:
cc_expiration_date = None
cc_expiration_delta = cc.expiration_delta.total_seconds()
elif cc.expiration_date is not None:
cc_expiration_date = (
cc.expiration_date if isinstance(cc.expiration_date, datetime) else now + cc.expiration_date
)
cc_expiration_delta = None
elif rec_expiration_date is not None:
cc_expiration_date = rec_expiration_date
cc_expiration_delta = None
else:
raise ValueError(
"Control change has no expiration and the recommendation has no expiration_date to inherit from"
)
ccs.append(
RecommendationControlChange(
retry=cc.retries,
timeout=cc.timeout if cc.timeout is not None else cc.retry_interval,
expiration_date=cc_expiration_date,
expiration_delta=cc_expiration_delta,
payload=cc.payload,
resource=cc.resource,
control_change_id=cc.control_change_id,
from_value=cc.from_value, # type: ignore[assignment]
trace_id=cc.trace_id,
queueing=QueueingMode(mode=cc.queueing) if cc.queueing is not None else None,
)
)
actions = [
RecommendationCustomAction(
type=action.type,
resource=action.resource,
custom_action_id=action.custom_action_id,
trace_id=action.trace_id,
title=action.title,
description=action.description,
expiration_date=(
action.expiration_date
if isinstance(action.expiration_date, datetime)
else now + action.expiration_date
),
payload=action.payload,
)
for action in self.actions
]
return RecommendationMsg(
resource=self.resource,
trace_id=self.trace_id,
payload=RecommendationPayload(
resource=self.resource,
type=self.type,
description=self.description,
expiration_date=rec_expiration_date,
confidence=self.confidence,
actions=RecommendationActions(control_changes=ccs, custom_actions=actions),
metadata=self.metadata,
state="auto_accepted" if self.auto_accepted is True else None,
evidences=[BaseEvidence(type=ev._TYPE, payload=ev) for ev in self.evidences], # pyright: ignore[reportPrivateUsage]
custom_identifier=self.custom_identifier,
trace_id=self.trace_id,
),
)
[docs]
@dataclass
class AppParameter:
"""App Parameter Helper.
Args:
resource (KRNAssetParameter): Kelvin Resource name for the target Asset Parameter
value (Union[bool, int, float, string]): parameter value
comment (Optional[str]): optional comment for parameter change
"""
resource: KRNAssetParameter
value: ParameterType
comment: Optional[str] = None
[docs]
class AssetParameter(AppParameter):
"""[Deprecated] Use AppParameter instead
Asset Parameter Helper.
Args:
resource (KRNAssetParameter): Kelvin Resource name for the target Asset Parameter
value (Union[bool, int, float, string]): parameter value
comment (Optional[str]): optional comment for parameter change
"""
[docs]
@dataclass
class AppParameters(MessageBuilder[ParametersMsg]):
"""Parameters Builder. Set application parameters in bulk.
Args:
resource (Optional[KRNAppVersion]): Optional Kelvin Resource name for the target App Version.
Defaults to current app.
parameters (list[AssetParameters]): list of single asset parameters
"""
parameters: list[AppParameter]
resource: Optional[KRNAppVersion] = None
[docs]
@override
def to_message(self) -> ParametersMsg:
asset_params: dict[str, list[EdgeParameter]] = {}
for asset_param in self.parameters:
asset_params.setdefault(asset_param.resource.asset, []).append(
EdgeParameter(name=asset_param.resource.parameter, value=asset_param.value, comment=asset_param.comment)
)
param_models = [
ResourceParameters(resource=KRNAsset(asset), parameters=params) for asset, params in asset_params.items()
]
return ParametersMsg(resource=self.resource, payload=ParametersPayload(resource_parameters=param_models))
[docs]
class AssetParameters(AppParameters):
"""[Deprecated] Use AppParameters instead
Parameters Builder. Set application parameters in bulk.
Args:
resource (Optional[KRNAppVersion]): Optional Kelvin Resource name for the target App Version.
Defaults to current app.
parameters (list[AssetParameters]): list of single asset parameters
"""
[docs]
@dataclass
class Insight(MessageBuilder[DataTagMsg]):
"""Insight (formerly Data Tag).
Args:
start_date (datetime): The start date of the insight.
tag_name (str): The name of the insight.
resource (KRNAsset): The asset resource associated with the insight.
end_date (Optional[datetime]): The end date of the insight. If not specified, the insight is
considered one point in time, the start_date.
contexts (Optional[list[KRNDatastream]]): The list of datastream contexts associated with the insight.
description (Optional[str]): The description of the insight. Truncated to 256 characters.
Attributes:
acceptance (Optional[InsightAcceptance]): Set by the platform on received insights, ``SNOOZED`` while the
tag is snoozed for the asset. Not a constructor argument.
"""
start_date: datetime
tag_name: str
resource: KRNAsset
end_date: Optional[datetime] = None
contexts: Optional[list[KRNDatastream]] = None
description: Optional[str] = None
acceptance: Optional[InsightAcceptance] = Field(default=None, init=False)
[docs]
@override
def to_message(self) -> DataTagMsg:
end_date = self.end_date or self.start_date
description = self.description[:256] if self.description else None
return DataTagMsg(
resource=self.resource,
payload=DataTagPayload(
start_date=self.start_date,
end_date=end_date,
tag_name=self.tag_name,
resource=self.resource,
description=description,
contexts=self.contexts,
),
)
[docs]
@classmethod
@override
def from_message(cls, msg: DataTagMsg) -> Self:
obj = cls(
start_date=msg.payload.start_date,
tag_name=msg.payload.tag_name,
resource=msg.payload.resource,
end_date=msg.payload.end_date,
contexts=msg.payload.contexts,
description=msg.payload.description,
)
obj.acceptance = msg.payload.acceptance
obj._msg = msg
return obj
[docs]
@deprecated("DataTag is deprecated, use Insight instead")
class DataTag(Insight):
"""[Deprecated] Use Insight instead."""
[docs]
@dataclass
class ControlAck(MessageBuilder[ControlChangeAck]):
"""Control Change Ack
Args:
resource (KRNAssetDataStream): The resource associated with the control change
state (StateEnum): The state of the control change
message (Optional[str]): Optional message
before (Optional[ValuePoint]): Optional value point before the control change
after (Optional[ValuePoint]): Optional value point after the control change
metadata (Optional[dict]): Optional metadata for the control change ack
"""
resource: KRNAssetDataStream
state: StateEnum
message: Optional[str] = None
before: Optional[ValuePoint] = None
after: Optional[ValuePoint] = None
metadata: Optional[dict[str, Any]] = None
[docs]
@override
def to_message(self) -> ControlChangeAck:
return ControlChangeAck(
resource=self.resource,
payload=ControlChangeAckPayload(
state=self.state,
message=self.message,
reported=ReportedValues(before=self.before, after=self.after) if self.before or self.after else None,
metadata=self.metadata,
),
)
[docs]
@dataclass
class CustomAction(MessageBuilder[CustomActionMsg]):
"""Custom Action Builder. Use this helper class to build a Kelvin custom action.
Args:
resource (KRN): The kelvin resource targeted by the action, represented by a KRN (usually KRNAssetData)
type (str): The type of the action
title (str): The title of the action
description (Optional[str]): Optional description for the action
expiration_date (Union[datetime, timedelta]): Expiration date for the action.
Provide either a absolute datetime or a timedelta from now
payload (Dict): Optional payload for the action
trace_id (Optional[str]): Optional trace ID for the action
custom_action_id (UUID): Optional UUID to set an specific ID for the control change
"""
resource: KRN
type: str
title: str
expiration_date: Union[datetime, timedelta]
description: Optional[str] = None
payload: dict[str, Any] = Field(default_factory=dict)
trace_id: Optional[str] = None
custom_action_id: Optional[UUID] = None
[docs]
@override
def to_message(self) -> CustomActionMsg:
if isinstance(self.expiration_date, timedelta):
expiration = datetime.now() + self.expiration_date
else:
expiration = self.expiration_date
return CustomActionMsg(
id=self.custom_action_id if self.custom_action_id is not None else uuid4(),
type=KMessageTypeAction(action_type=self.type),
trace_id=self.trace_id,
resource=self.resource,
payload=CustomActionPayload(
title=self.title,
description=self.description,
expiration_date=expiration,
payload=self.payload,
),
)
[docs]
@classmethod
@override
def from_message(cls, msg: CustomActionMsg) -> CustomAction:
if not isinstance(msg.type, KMessageTypeAction): # pyright: ignore[reportUnnecessaryIsInstance]
raise ValueError(f"invalid message type: {msg.type}")
obj = cls(
resource=msg.resource, # type: ignore[arg-type]
type=msg.type.type,
title=msg.payload.title,
description=msg.payload.description,
expiration_date=msg.payload.expiration_date,
payload=msg.payload.payload,
trace_id=msg.trace_id,
custom_action_id=msg.id,
)
obj._msg = msg
return obj
[docs]
def result(
self,
success: bool,
message: Optional[str] = None,
response: Optional[dict[str, Any]] = None,
metadata: Optional[dict[str, Any]] = None,
) -> CustomActionResult:
"""Create a result for the custom action.
Args:
success (bool): Whether the action was successful
message (Optional[str]): Optional message
response (Optional[dict]): Optional response body for the action result.
It is delivered back to the app that created the action, in its
custom action report.
metadata (Optional[dict]): [Deprecated] Use ``response`` instead. When
only ``metadata`` is given its value is used as the ``response``.
Returns:
CustomActionResult: The custom action result
"""
if self._msg is None:
raise ValueError("cannot build CustomActionAck, internal _msg not set")
if metadata is not None:
warnings.warn(
"Argument 'metadata' is deprecated, use 'response' instead",
DeprecationWarning,
stacklevel=2,
)
# folded in here rather than passed down, so the caller gets exactly
# one warning for one deprecated argument
if response is None:
response = metadata
return CustomActionResult(
resource=self.resource,
action_id=self._msg.id,
success=success,
message=message,
response=response,
trace_id=self.trace_id,
)
[docs]
@dataclass
class CustomActionResult(MessageBuilder[CustomActionResultMsg]):
"""Custom Action Result Builder. Use this helper class to build a Kelvin custom action ack.
Args:
resource (KRN): The kelvin resource targeted by the action, represented by a KRN (usually
KRNAssetDataStream)
action_id (UUID): The ID of the action
success (bool): Whether the action was successful
message (Optional[str]): Optional message
response (Optional[dict[str, Any]]): Optional response body for the action result.
It is delivered back to the app that created the action, in its custom action report.
metadata (Optional[dict[str, Any]]): [Deprecated] Use ``response`` instead. When only
``metadata`` is given its value is used as the ``response``.
trace_id (Optional[str]): Optional trace ID for the action
"""
resource: KRN
action_id: UUID
success: bool
message: Optional[str] = None
# `response` sits before `metadata` so existing positional calls
# `CustomActionResult(resource, action_id, success, message)` keep working.
response: Optional[dict[str, Any]] = None
metadata: Annotated[
Optional[dict[str, Any]],
Field(deprecated="Argument 'metadata' is deprecated, use 'response' instead"),
] = None
trace_id: Optional[str] = None
def __post_init__(self) -> None:
# backfill `response` from the deprecated `metadata` argument
if self.metadata is not None and self.response is None:
self.response = self.metadata
[docs]
@override
def to_message(self) -> CustomActionResultMsg:
return CustomActionResultMsg(
resource=self.resource,
trace_id=self.trace_id,
payload=CustomActionResultPayload(
id=self.action_id,
success=self.success,
message=self.message,
response=self.response,
),
)
[docs]
@classmethod
@override
def from_message(cls, msg: CustomActionResultMsg) -> CustomActionResult:
response = msg.payload.response
if response is None:
# Older producers still populate `metadata`. Read it out of the model
# dict so the field's deprecation descriptor does not warn the app for
# something the SDK did.
response = msg.payload.__dict__.get("metadata")
obj = cls(
resource=msg.resource, # type: ignore[arg-type]
action_id=msg.payload.id,
success=msg.payload.success,
message=msg.payload.message,
response=response,
trace_id=msg.trace_id,
)
obj._msg = msg
return obj
[docs]
@dataclass
class CustomActionReport(MessageBuilder[CustomActionReportMsg]):
"""Custom Action Report. Received by the app that created a custom action once
that action reaches a terminal state (``completed``, ``failed`` or ``cancelled``).
Reports are automatic for every custom action type the app declares as an output,
there is nothing to opt into.
```python
from kelvin.message import CustomActionReport, CustomActionState
@app.on_custom_action_report
async def handle_report(report: CustomActionReport) -> None:
if report.state == CustomActionState.COMPLETED:
print(f"{report.action_id} completed: {report.response}")
else:
print(f"{report.action_id} ended as {report.state}: {report.message}")
```
Args:
resource (KRN): The kelvin resource the action targeted
action_id (UUID): The ID of the custom action being reported
type (str): The type of the action
title (str): The title of the action
state (CustomActionState): The terminal state the action reached
expiration_date (datetime): Expiration date of the action
created (datetime): When the action was created
created_by (str): KRN of the creator of the action, as a string
updated (datetime): When the action was last updated
updated_by (str): KRN of the last updater of the action, as a string
description (Optional[str]): Description of the action
timestamp (Optional[datetime]): Timestamp of the action
payload (Optional[dict[str, Any]]): The payload the action was created with
response (Optional[dict[str, Any]]): The response reported by the executor.
Absent when the executor reported none, and never set on a cancelled action.
failure_reason (Optional[CustomActionFailureReason]): Why the action failed
message (Optional[str]): Message reported with the terminal state
status_logs (list[CustomActionStatusLog]): The action's status history
trace_id (Optional[str]): Trace ID the action was created with
"""
resource: KRN
action_id: UUID
type: str
title: str
state: CustomActionState
expiration_date: datetime
created: datetime
created_by: str
updated: datetime
updated_by: str
description: Optional[str] = None
timestamp: Optional[datetime] = None
payload: Optional[dict[str, Any]] = None
response: Optional[dict[str, Any]] = None
failure_reason: Optional[CustomActionFailureReason] = None
message: Optional[str] = None
status_logs: list[CustomActionStatusLog] = Field(default_factory=list)
trace_id: Optional[str] = None
[docs]
@override
def to_message(self) -> CustomActionReportMsg:
return CustomActionReportMsg(
resource=self.resource,
# The envelope trace id carries the custom action id; the app's own
# trace id travels in the payload. Matches what the edge emits.
trace_id=str(self.action_id),
payload=CustomActionReportPayload(
id=self.action_id,
type=self.type,
title=self.title,
state=self.state,
expiration_date=self.expiration_date,
created=self.created,
created_by=self.created_by,
updated=self.updated,
updated_by=self.updated_by,
trace_id=self.trace_id,
description=self.description,
timestamp=self.timestamp,
payload=self.payload,
response=self.response,
failure_reason=self.failure_reason,
message=self.message,
status_logs=self.status_logs,
),
)
[docs]
@classmethod
@override
def from_message(cls, msg: CustomActionReportMsg) -> CustomActionReport:
obj = cls(
resource=msg.resource, # type: ignore[arg-type]
action_id=msg.payload.id,
type=msg.payload.type,
title=msg.payload.title,
state=msg.payload.state,
expiration_date=msg.payload.expiration_date,
created=msg.payload.created,
created_by=msg.payload.created_by,
updated=msg.payload.updated,
updated_by=msg.payload.updated_by,
description=msg.payload.description,
timestamp=msg.payload.timestamp,
payload=msg.payload.payload,
response=msg.payload.response,
failure_reason=msg.payload.failure_reason,
message=msg.payload.message,
status_logs=msg.payload.status_logs,
trace_id=msg.payload.trace_id,
)
obj._msg = msg
return obj
BUILDER_REGISTRY: dict[type[Message], type[MessageBuilder[Any]]] = {
ControlChangeMsg: ControlChange,
CustomActionMsg: CustomAction,
CustomActionResultMsg: CustomActionResult,
CustomActionReportMsg: CustomActionReport,
DataTagMsg: Insight,
}
def convert_message(msg: Message) -> Optional[MessageBuilder[Any]]:
"""Convert a message to its builder class.
Args:
msg (Message): The message to convert
Returns:
Optional[MessageBuilder]: The converted message builder
"""
if msg.__class__ in BUILDER_REGISTRY:
return BUILDER_REGISTRY[msg.__class__].from_message(msg)
return None