Message

The message module provides message types and builders for Kelvin applications.

Main Module

Kelvin Messages.

class kelvin.message.Message(**kwargs)

Bases: BaseModel

Parameters:

• type (KMessageType)

• resource (KRN | None)

• id (UUID)

• trace_id (str | None)

• source (KRN | None)

• timestamp (datetime)

• payload (Any)

MESSAGE_TYPES_: ClassVar[Dict[KMessageType, Type[Message]]] = {KMessageTypeAction('custom-action-create', ''): <class 'kelvin.message.base_messages.CustomActionMsg'>, KMessageTypeActionAck('custom-action-result', ''): <class 'kelvin.message.base_messages.CustomActionResultMsg'>, KMessageTypeControl('control', ''): <class 'kelvin.message.base_messages.ControlChangeMsg'>, KMessageTypeControlAck('control-ack', ''): <class 'kelvin.message.base_messages.ControlChangeAck'>, KMessageTypeControlStatus('control-status', ''): <class 'kelvin.message.base_messages.ControlChangeStatus'>, KMessageTypeData('data', 'pt=boolean'): <class 'kelvin.message.primitives.Boolean'>, KMessageTypeData('data', 'pt=number'): <class 'kelvin.message.primitives.Number'>, KMessageTypeData('data', 'pt=object;icd=kelvin.sensor_data'): <class 'kelvin.message.base_messages.SensorDataMsg'>, KMessageTypeData('data', 'pt=string'): <class 'kelvin.message.primitives.String'>, KMessageTypeDataTag('datatag', ''): <class 'kelvin.message.base_messages.DataTagMsg'>, KMessageTypeParameter('parameter', 'pt=boolean'): <class 'kelvin.message.primitives.BooleanParameter'>, KMessageTypeParameter('parameter', 'pt=number'): <class 'kelvin.message.primitives.NumberParameter'>, KMessageTypeParameter('parameter', 'pt=string'): <class 'kelvin.message.primitives.StringParameter'>, KMessageTypeParameters('parameters', ''): <class 'kelvin.message.base_messages.ParametersMsg'>, KMessageTypeRecommendation('recommendation', ''): <class 'kelvin.message.base_messages.RecommendationMsg'>, KMessageTypeRuntimeManifest('runtime_manifest', ''): <class 'kelvin.message.runtime_manifest.RuntimeManifest'>}

TYPE_: ClassVar[KMessageType | None] = None

type: KMessageType

resource: KRN | None

id: UUID

trace_id: str | None

source: KRN | None

timestamp: datetime

payload: Any

static __new__(cls, **kwargs)

Initialise message.

Return type:

Message

Parameters:

kwargs (Any)

__init__(**kwargs)

Create a kelvin Message.

Parameters

idstr, optional

UUID of the message. Optional, auto generated if not provided.

typeKMessageType

Message Type

trace_idstr, optional

Optional trace id. UUID

sourceKRN, optional

Identifies the source of the message.

timestampdatetime, optional

Sets a timestamp for the message. If not provided current time is used.

resourceKRN, optional

Sets a resource that the message relates to.

payloadAny

Payload of the message. Specific for each message sub type.

Parameters:

kwargs (Any)

Return type:

None

dict(by_alias=True, exclude_none=True, exclude_unset=False, **kwargs)

Generate a dictionary representation of the model.

Return type:

Dict[str, Any]

Parameters:

• by_alias (bool)

• exclude_none (bool)

• exclude_unset (bool)

• kwargs (Any)

json(by_alias=True, exclude_none=True, exclude_unset=False, **kwargs)

Generate a dictionary representation of the model.

Return type:

str

Parameters:

• by_alias (bool)

• exclude_none (bool)

• exclude_unset (bool)

• kwargs (Any)

encode()

Encode message

Return type:

bytes

classmethod decode(data)

Return type:

Self

Parameters:

data (bytes)

serialize_timestamp(ts)

Return type:

str

Parameters:

ts (datetime)

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kelvin.message.Boolean(**kwargs)

Bases: Message

Parameters:

• type (KMessageTypeData)

• resource (KRN | None)

• id (UUID)

• trace_id (str | None)

• source (KRN | None)

• timestamp (datetime)

• payload (Annotated[bool, Strict(strict=True)])

TYPE_: ClassVar[KMessageType | None] = KMessageTypeData('data', 'pt=boolean')

type: KMessageTypeData

payload: Annotated[bool, Strict(strict=True)]

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kelvin.message.Number(**kwargs)

Bases: Message

Parameters:

• type (KMessageTypeData)

• resource (KRN | None)

• id (UUID)

• trace_id (str | None)

• source (KRN | None)

• timestamp (datetime)

• payload (Annotated[float, Strict(strict=True)] | Annotated[int, Strict(strict=True)])

TYPE_: ClassVar[KMessageType | None] = KMessageTypeData('data', 'pt=number')

type: KMessageTypeData

payload: Annotated[float, Strict(strict=True)] | Annotated[int, Strict(strict=True)]

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kelvin.message.String(**kwargs)

Bases: Message

Parameters:

• type (KMessageTypeData)

• resource (KRN | None)

• id (UUID)

• trace_id (str | None)

• source (KRN | None)

• timestamp (datetime)

• payload (Annotated[str, Strict(strict=True)])

TYPE_: ClassVar[KMessageType | None] = KMessageTypeData('data', 'pt=string')

type: KMessageTypeData

payload: Annotated[str, Strict(strict=True)]

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kelvin.message.NumberParameter(**kwargs)

Bases: Message

Parameters:

• type (KMessageTypeParameter)

• resource (KRN | None)

• id (UUID)

• trace_id (str | None)

• source (KRN | None)

• timestamp (datetime)

• payload (Annotated[float, Strict(strict=True)] | Annotated[int, Strict(strict=True)])

TYPE_: ClassVar[KMessageType | None] = KMessageTypeParameter('parameter', 'pt=number')

type: KMessageTypeParameter

payload: Annotated[float, Strict(strict=True)] | Annotated[int, Strict(strict=True)]

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kelvin.message.BooleanParameter(**kwargs)

Bases: Message

Parameters:

• type (KMessageTypeParameter)

• resource (KRN | None)

• id (UUID)

• trace_id (str | None)

• source (KRN | None)

• timestamp (datetime)

• payload (Annotated[bool, Strict(strict=True)])

TYPE_: ClassVar[KMessageType | None] = KMessageTypeParameter('parameter', 'pt=boolean')

type: KMessageTypeParameter

payload: Annotated[bool, Strict(strict=True)]

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kelvin.message.StringParameter(**kwargs)

Bases: Message

Parameters:

• type (KMessageTypeParameter)

• resource (KRN | None)

• id (UUID)

• trace_id (str | None)

• source (KRN | None)

• timestamp (datetime)

• payload (Annotated[str, Strict(strict=True)])

TYPE_: ClassVar[KMessageType | None] = KMessageTypeParameter('parameter', 'pt=string')

type: KMessageTypeParameter

payload: Annotated[str, Strict(strict=True)]

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kelvin.message.KMessageType(msg_type=None, components=None)

Bases: object

Kelvin Message Type representation

Parameters:

• msg_type (str)

• components (Dict[str, str])

msg_type: str

components: Dict[str, str]

classmethod validate(v)

Return type:

KMessageType

Parameters:

v (Any)

classmethod from_krn(msg_type, components)

Return type:

KMessageType

Parameters:

• msg_type (str)

• components (Dict[str, str])

classmethod from_string(v)

Return type:

Self

Parameters:

v (str)

encode()

Return type:

str

class kelvin.message.KMessageTypeData(primitive='object', icd=None)

Bases: KMessageTypePrimitive

Parameters:

• primitive (PrimitiveTypes)

• icd (str | None)

class kelvin.message.KMessageTypePrimitive(primitive='object', icd=None)

Bases: KMessageType

Parameters:

• primitive (PrimitiveTypes)

• icd (str | None)

primitive: PrimitiveTypes

icd: str | None = None

classmethod from_krn(msg_type, components)

Return type:

Self

Parameters:

• msg_type (str)

• components (Dict[str, str] | None)

class kelvin.message.KMessageTypeParameter(primitive='object', icd=None)

Bases: KMessageTypePrimitive

Parameters:

• primitive (PrimitiveTypes)

• icd (str | None)

class kelvin.message.KMessageTypeControl(msg_type=None, components=None)

Bases: KMessageType

Parameters:

• msg_type (str)

• components (Dict[str, str])

classmethod from_krn(_, __)

Return type:

Self

Parameters:

• _ (str)

• __ (Dict[str, str] | None)

class kelvin.message.KMessageTypeRecommendation(msg_type=None, components=None)

Bases: KMessageType

Parameters:

• msg_type (str)

• components (Dict[str, str])

classmethod from_krn(_, __)

Return type:

Self

Parameters:

• _ (str)

• __ (Dict[str, str] | None)

class kelvin.message.KMessageTypeControlStatus(msg_type=None, components=None)

Bases: KMessageType

Parameters:

• msg_type (str)

• components (Dict[str, str])

classmethod from_krn(_, __)

Return type:

Self

Parameters:

• _ (str)

• __ (Dict[str, str] | None)

class kelvin.message.RecommendationMsg(**kwargs)

Bases: Message

Parameters:

• type (KMessageTypeRecommendation)

• resource (KRN | None)

• id (UUID)

• trace_id (str | None)

• source (KRN | None)

• timestamp (datetime)

• payload (RecommendationPayload)

TYPE_: ClassVar[KMessageType | None] = KMessageTypeRecommendation('recommendation', '')

type: KMessageTypeRecommendation

payload: RecommendationPayload

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kelvin.message.RecommendationActions(**data)

Bases: BaseModel

Parameters:

• control_changes (List[RecommendationControlChange])

• custom_actions (List[RecommendationCustomAction])

control_changes: List[RecommendationControlChange]

custom_actions: List[RecommendationCustomAction]

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kelvin.message.RecommendationControlChange(**data)

Bases: ControlChangePayload

Parameters:

• timeout (int | None)

• retry (int | None)

• expiration_date (datetime)

• payload (Any)

• from_value (ValuePoint | None)

• state (str | None)

• resource (KRN | None)

• control_change_id (UUID | None)

• trace_id (str | None)

retries: int | None

state: str | None

resource: KRN | None

control_change_id: UUID | None

trace_id: str | None

model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kelvin.message.ControlChangeMsg(**kwargs)

Bases: Message

Generic Control Change Message

Parameters:

• type (KMessageTypeControl)

• resource (KRN | None)

• id (UUID)

• trace_id (str | None)

• source (KRN | None)

• timestamp (datetime)

• payload (ControlChangePayload)

TYPE_: ClassVar[KMessageType | None] = KMessageTypeControl('control', '')

type: KMessageTypeControl

payload: ControlChangePayload

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kelvin.message.ControlChangePayload(**data)

Bases: BaseModel

Parameters:

• timeout (int | None)

• retries (int | None)

• expiration_date (datetime)

• payload (Any)

• from_value (ValuePoint | None)

model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

timeout: int | None

retries: int | None

expiration_date: datetime

payload: Any

from_value: ValuePoint | None

serialize_timestamp(ts)

Return type:

str

Parameters:

ts (datetime)

class kelvin.message.ControlChangeStatus(**kwargs)

Bases: Message

Control Change Status

Parameters:

• type (KMessageTypeControlStatus)

• resource (KRN | None)

• id (UUID)

• trace_id (str | None)

• source (KRN | None)

• timestamp (datetime)

• payload (ControlChangeStatusPayload)

TYPE_: ClassVar[KMessageType | None] = KMessageTypeControlStatus('control-status', '')

type: KMessageTypeControlStatus

payload: ControlChangeStatusPayload

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kelvin.message.ControlChangeAckPayload(**data)

Bases: ControlChangeStatusPayload

Control Change Ack Payload

Parameters:

• state (StateEnum)

• message (str | None)

• reported (ReportedValues | None)

• control_change_id (UUID | None)

• metadata (Dict | None)

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kelvin.message.ControlChangeAck(**kwargs)

Bases: Message

Control Change Ack

Parameters:

• type (KMessageTypeControlAck)

• resource (KRN | None)

• id (UUID)

• trace_id (str | None)

• source (KRN | None)

• timestamp (datetime)

• payload (ControlChangeAckPayload)

TYPE_: ClassVar[KMessageType | None] = KMessageTypeControlAck('control-ack', '')

type: KMessageTypeControlAck

payload: ControlChangeAckPayload

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kelvin.message.ValuePoint(**data)

Bases: BaseModel

Parameters:

• value (Any)

• timestamp (datetime)

• source (str | None)

value: Any

timestamp: datetime

source: str | None

serialize_timestamp(ts)

Return type:

str

Parameters:

ts (datetime)

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kelvin.message.StateEnum(*values)

Bases: str, Enum

ready = 'ready'

sent = 'sent'

failed = 'failed'

processed = 'processed'

applied = 'applied'

rejected = 'rejected'

class kelvin.message.Recommendation(*args, **kwargs)

Bases: MessageBuilder[RecommendationMsg]

Recommendation Builder. Use this helper class to build a Kelvin Recommendation.

Parameters:

• 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 | None) – (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: datetime | timedelta | None = None

description: str | None = None

confidence: int | None = None

control_changes: List[ControlChange] = FieldInfo(annotation=NoneType, required=False, default_factory=list)

metadata: Dict | None = None

auto_accepted: bool = False

evidences: List[Evidence] = FieldInfo(annotation=NoneType, required=False, default_factory=list)

custom_identifier: str | None = None

actions: List[CustomAction] = FieldInfo(annotation=NoneType, required=False, default_factory=list)

trace_id: str | None = None

to_message()

Return type:

RecommendationMsg

class kelvin.message.ControlChange(*args, **kwargs)

Bases: MessageBuilder[ControlChangeMsg]

Control change builder Use this helper class to build a Kelvin Control change.

Parameters:

• resource (KRN) – The kelvin resource targeted by the control change, represented by a KRN (usually KRNAssetDataStream)

• expiration_date (datetime | timedelta) – The absolute time in the future when the Control Change expires. Provide either a absolute datetime or a timedelta from now

• payload (bool, int, float, str) – The desired target value for the control change

• retries (int) – Optional number of retries

• timeout (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

resource: KRN

expiration_date: datetime | timedelta

payload: Any

retries: int | None = None

timeout: int | None = None

control_change_id: UUID | None = None

from_value: ValuePoint | None = None

trace_id: str | None = None

to_message()

Return type:

ControlChangeMsg

classmethod from_message(msg)

Return type:

ControlChange

Parameters:

msg (ControlChangeMsg)

class kelvin.message.AssetParameter(*args, **kwargs)

Bases: AppParameter

[Deprecated] Use AppParameter instead Asset Parameter Helper.

Parameters:

• 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

class kelvin.message.AssetParameters(*args, **kwargs)

Bases: AppParameters

[Deprecated] Use AppParameters instead Parameters Builder. Set application parameters in bulk.

Parameters:

• resource (Optional[KRNAppVersion]) – Optional Kelvin Resource name for the target App Version. Defaults to current app.

• parameters (List[AssetParameters]) – list of single asset parameters

class kelvin.message.AssetDataMessage(**kwargs)

Bases: Message

A message containing data for an asset datastream.

This type represents messages that carry data values (numbers, strings, booleans) for a specific datastream on an asset.

Parameters:

• type (KMessageTypeData)

• resource (KRNAssetDataStream)

• id (UUID)

• trace_id (str | None)

• source (KRN | None)

• timestamp (datetime)

• payload (Any)

type

The data message type (includes primitive type information).

resource

The KRN identifying the asset and datastream.

type: KMessageTypeData

resource: KRNAssetDataStream

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kelvin.message.DataTag(*args, **kwargs)

Bases: MessageBuilder

Data Tag.

Parameters:

• start_date (datetime) – The start date of the data tag.

• tag_name (str) – The name of the data tag.

• resource (KRNAsset) – The asset resource associated with the data tag.

• end_date (Optional[datetime]) – The end date of the data tag. If not specified, the data tag is

• time (considered one point in)

• start_date. (the)

• contexts (Optional[List[KRN]]) – The list of contexts associated with the data tag.

• description (Optional[str]) – The description of the data tag. Truncated to 256 characters.

start_date: datetime

tag_name: str

resource: KRNAsset

end_date: datetime | None = None

contexts: List[KRN] | None = None

description: str | None = None

to_message()

Return type:

DataTagMsg

class kelvin.message.DataTagMsg(**kwargs)

Bases: Message

Parameters:

• type (KMessageTypeDataTag)

• resource (KRNAsset)

• id (UUID)

• trace_id (str | None)

• source (KRN | None)

• timestamp (datetime)

• payload (DataTagPayload)

TYPE_: ClassVar[KMessageType | None] = KMessageTypeDataTag('datatag', '')

type: KMessageTypeDataTag

resource: KRNAsset

payload: DataTagPayload

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kelvin.message.KMessageTypeDataTag(msg_type=None, components=None)

Bases: KMessageType

Parameters:

• msg_type (str)

• components (Dict[str, str])

classmethod from_krn(_, __)

Return type:

Self

Parameters:

• _ (str)

• __ (Dict[str, str] | None)

class kelvin.message.ControlAck(*args, **kwargs)

Bases: MessageBuilder

Control Change Ack

Parameters:

• 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: str | None = None

before: ValuePoint | None = None

after: ValuePoint | None = None

metadata: Dict | None = None

to_message()

Return type:

ControlChangeAck

class kelvin.message.AppParameters(*args, **kwargs)

Bases: MessageBuilder

Parameters Builder. Set application parameters in bulk.

Parameters:

• 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: KRNAppVersion | None = None

to_message()

Return type:

ParametersMsg

class kelvin.message.AppParameter(*args, **kwargs)

Bases: object

App Parameter Helper.

Parameters:

• 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: Annotated[bool, Strict(strict=True)] | Annotated[int, Strict(strict=True)] | Annotated[float, Strict(strict=True)] | Annotated[str, Strict(strict=True)]

comment: str | None = None

class kelvin.message.CustomActionMsg(**kwargs)

Bases: Message

Parameters:

• type (KMessageTypeAction)

• resource (KRN | None)

• id (UUID)

• trace_id (str | None)

• source (KRN | None)

• timestamp (datetime)

• payload (CustomActionPayload)

TYPE_: ClassVar[KMessageType | None] = KMessageTypeAction('custom-action-create', '')

type: KMessageTypeAction

payload: CustomActionPayload

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kelvin.message.CustomActionResultMsg(**kwargs)

Bases: Message

Parameters:

• type (KMessageTypeActionAck)

• resource (KRN | None)

• id (UUID)

• trace_id (str | None)

• source (KRN | None)

• timestamp (datetime)

• payload (CustomActionResultPayload)

TYPE_: ClassVar[KMessageType | None] = KMessageTypeActionAck('custom-action-result', '')

type: KMessageTypeActionAck

payload: CustomActionResultPayload

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kelvin.message.CustomAction(*args, **kwargs)

Bases: MessageBuilder[CustomActionMsg]

Custom Action Builder. Use this helper class to build a Kelvin custom action.

Parameters:

• 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: datetime | timedelta

description: str | None = None

payload: Dict = FieldInfo(annotation=NoneType, required=False, default_factory=dict)

trace_id: str | None = None

custom_action_id: UUID | None = None

to_message()

Return type:

CustomActionMsg

classmethod from_message(msg)

Return type:

CustomAction

Parameters:

msg (CustomActionMsg)

result(success, message=None, metadata=None)

Create a result for the custom action.

Parameters:

• success (bool) – Whether the action was successful

• message (Optional[str]) – Optional message

• metadata (Optional[Dict]) – Optional metadata for the action ack

Returns:

The custom action ack

Return type:

CustomActionResult

class kelvin.message.CustomActionResult(*args, **kwargs)

Bases: MessageBuilder[CustomActionResultMsg]

Custom Action Result Builder. Use this helper class to build a Kelvin custom action ack. :param resource: The kelvin resource targeted by the action, represented by a KRN (usually

KRNAssetDataStream)

Parameters:

• action_id (UUID) – The ID of the action

• success (bool) – Whether the action was successful

• message (Optional[str]) – Optional message

• metadata (Dict | None) – Dict[str, Any]: Optional metadata for the action ack

• trace_id (Optional[str]) – Optional trace ID for the action

• resource (KRN)

resource: KRN

action_id: UUID

success: bool

message: str | None = None

metadata: Dict | None = None

trace_id: str | None = None

to_message()

Return type:

CustomActionResultMsg

classmethod from_message(msg)

Return type:

CustomActionResult

Parameters:

msg (CustomActionResultMsg)

class kelvin.message.RuntimeManifest(**kwargs)

Bases: Message

Parameters:

• type (KMessageTypeRuntimeManifest)

• resource (KRN | None)

• id (UUID)

• trace_id (str | None)

• source (KRN | None)

• timestamp (datetime)

• payload (RuntimeManifestPayload)

TYPE_: ClassVar[KMessageType | None] = KMessageTypeRuntimeManifest('runtime_manifest', '')

type: KMessageTypeRuntimeManifest

payload: RuntimeManifestPayload

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kelvin.message.WayEnum(*values)

Bases: str, Enum

output = 'output'

input_cc = 'input-cc'

input_cc_output = 'input-cc+output'

input = 'input'

output_cc = 'output-cc'

input_output_cc = 'input+output-cc'

class kelvin.message.AssetDataQualityMessage(**kwargs)

Bases: Message

A message containing data quality information for an asset.

Parameters:

• type (KMessageTypeData)

• resource (KRNAssetDataQuality)

• id (UUID)

• trace_id (str | None)

• source (KRN | None)

• timestamp (datetime)

• payload (Any)

type

The data message type.

resource

The KRN identifying the asset and data quality metric.

type: KMessageTypeData

resource: KRNAssetDataQuality

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kelvin.message.AssetDataStreamDataQualityMessage(**kwargs)

Bases: Message

A message containing data quality information for a specific asset datastream.

Parameters:

• type (KMessageTypeData)

• resource (KRNAssetDataStreamDataQuality)

• id (UUID)

• trace_id (str | None)

• source (KRN | None)

• timestamp (datetime)

• payload (Any)

type

The data message type.

resource

The KRN identifying the asset, datastream, and data quality metric.

type: KMessageTypeData

resource: KRNAssetDataStreamDataQuality

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Utils

Utility functions.

kelvin.message.utils.flatten(x, separator='.')

Flatten nested mappings.

Return type:

Iterator[Tuple[str, Any]]

Parameters:

• x (Mapping[str, Any])

• separator (str)

kelvin.message.utils.to_rfc3339_timestamp(x)

Convert datetime to RFC-3339 timestamp (UTC).

Return type:

str

Parameters:

x (datetime)

kelvin.message.utils.from_rfc3339_timestamp(x)

Convert RFC-3339 timestamp to datetime (UTC).

Return type:

datetime

Parameters:

x (str)