from pydantic import BaseModel
from rid_lib import RID
from rid_lib.ext import Manifest, Bundle
from rid_lib.types import KoiNetNode
from .event import Event, EventType
[docs]
class KnowledgeObject(BaseModel):
"""A normalized knowledge representation for internal processing.
Represents an RID, manifest, bundle, or event. Contains three fields
(`normalized_event_type`, `source`, `network_targets`) used for
decision making in the knowledge processing pipeline.
The source indicates which node this object originated from, or
`None` if it was generated by this node.
The normalized event type indicates how the knowledge object is
viewed from the perspective of this node, and what cache actions
will take place. (`NEW`, `UPDATE`) -> cache write, `FORGET` ->
cache delete, `None` -> no cache action, and end of pipeline.
The network targets indicate other nodes in the network this
knowledge object will be sent to. The event sent to them will be
constructed from this knowledge object's RID, manifest, contents,
and normalized event type.
"""
rid: RID
manifest: Manifest | None = None
contents: dict | None = None
event_type: EventType | None = None
normalized_event_type: EventType | None = None
source: KoiNetNode | None = None
network_targets: set[KoiNetNode] = set()
def __repr__(self):
return f"<KObj '{self.rid}' event type: '{self.event_type}' -> '{self.normalized_event_type}', source: '{self.source}'>"
[docs]
@classmethod
def from_rid(
cls,
rid: RID,
event_type: EventType | None = None,
source: KoiNetNode | None = None
) -> "KnowledgeObject":
"""Creates a `KnowledgeObject` from an `RID`."""
return cls(
rid=rid,
event_type=event_type,
source=source
)
[docs]
@classmethod
def from_manifest(
cls,
manifest: Manifest,
event_type: EventType | None = None,
source: KoiNetNode | None = None
) -> "KnowledgeObject":
"""Creates a `KnowledgeObject` from a `Manifest`."""
return cls(
rid=manifest.rid,
manifest=manifest,
event_type=event_type,
source=source
)
[docs]
@classmethod
def from_bundle(
cls,
bundle: Bundle,
event_type: EventType | None = None,
source: KoiNetNode | None = None
) -> "KnowledgeObject":
"""Creates a `KnowledgeObject` from a `Bundle`."""
return cls(
rid=bundle.rid,
manifest=bundle.manifest,
contents=bundle.contents,
event_type=event_type,
source=source
)
[docs]
@classmethod
def from_event(
cls,
event: Event,
source: KoiNetNode | None = None
) -> "KnowledgeObject":
"""Creates a `KnowledgeObject` from an `Event`."""
return cls(
rid=event.rid,
manifest=event.manifest,
contents=event.contents,
event_type=event.event_type,
source=source
)
@property
def bundle(self) -> Bundle:
"""Bundle representation of knowledge object."""
if self.manifest is None or self.contents is None:
raise ValueError("Knowledge object missing manifest or contents, cannot convert to `Bundle`.")
return Bundle(
manifest=self.manifest,
contents=self.contents
)
@property
def normalized_event(self) -> Event:
"""Event representation of knowledge object."""
if self.normalized_event_type is None:
raise ValueError("Internal event's normalized event type is None, cannot convert to Event")
elif self.normalized_event_type == EventType.FORGET:
return Event(
rid=self.rid,
event_type=EventType.FORGET
)
else:
return Event(
rid=self.rid,
event_type=self.normalized_event_type,
manifest=self.manifest,
contents=self.contents
)