perception

The Perception system breaks down the environment into features that can be re-assembled as concepts.

FeatureNodeType = TypeVar('FeatureNodeType', bound=FeatureNode) module-attribute

FeatureType = TypeVar('FeatureType') module-attribute

PerceptionData = VisionData | Settled | Feature[Any] module-attribute

PerceptionEvent = Event[PerceptionData] module-attribute

cache_registry = defaultdict(WeakValueDictionary) module-attribute

fe_list = [] module-attribute

AreaFeature dataclass

Bases: Feature[FeatureNodeType]

Source code in roc/perception.py

@dataclass(kw_only=True)
class AreaFeature(Feature[FeatureNodeType]):
    type: int
    points: set[tuple[XLoc, YLoc]]
    size: int

    def get_points(self) -> set[tuple[XLoc, YLoc]]:
        return self.points

    def node_hash(self) -> int:
        return hash((self.type, self.size))

points instance-attribute

size instance-attribute

type instance-attribute

__init__(*, feature_name, origin_id, type, points, size)

get_points()

Source code in roc/perception.py

def get_points(self) -> set[tuple[XLoc, YLoc]]:
    return self.points

node_hash()

Source code in roc/perception.py

def node_hash(self) -> int:
    return hash((self.type, self.size))

Detail

Bases: Edge

Source code in roc/perception.py

class Detail(Edge):
    allowed_connections: EdgeConnectionsList = [("FeatureGroup", "FeatureNode")]

allowed_connections = [('FeatureGroup', 'FeatureNode')] class-attribute instance-attribute

Direction

Bases: str, Enum

Source code in roc/perception.py

class Direction(str, Enum):
    up = "UP"
    down = "DOWN"
    left = "LEFT"
    right = "RIGHT"
    up_right = "UP_RIGHT"
    up_left = "UP_LEFT"
    down_right = "DOWN_RIGHT"
    down_left = "DOWN_LEFT"

    def __str__(self) -> str:
        return self.value

down = 'DOWN' class-attribute instance-attribute

down_left = 'DOWN_LEFT' class-attribute instance-attribute

down_right = 'DOWN_RIGHT' class-attribute instance-attribute

left = 'LEFT' class-attribute instance-attribute

right = 'RIGHT' class-attribute instance-attribute

up = 'UP' class-attribute instance-attribute

up_left = 'UP_LEFT' class-attribute instance-attribute

up_right = 'UP_RIGHT' class-attribute instance-attribute

__str__()

Source code in roc/perception.py

def __str__(self) -> str:
    return self.value

Feature dataclass

Bases: ABC, Generic[FeatureNodeType]

Source code in roc/perception.py

@dataclass(kw_only=True)
class Feature(ABC, Generic[FeatureNodeType]):
    feature_name: str
    origin_id: tuple[str, str]

    def to_nodes(self) -> FeatureNodeType:
        # check local cache
        cache = cache_registry[self.feature_name]

        h = self.node_hash()
        if h in cache:
            return cache[h]  # type: ignore

        # if cache miss, find node in database
        n = self._dbfetch_nodes()
        if n is None:
            # if node doesn't exist, create it
            n = self._create_nodes()
            # n.labels.add("Feature")
            # n.labels.add(self.feature_name)

        cache[h] = n
        return n

    @abstractmethod
    def get_points(self) -> set[tuple[XLoc, YLoc]]: ...

    @abstractmethod
    def _create_nodes(self) -> FeatureNodeType: ...

    @abstractmethod
    def _dbfetch_nodes(self) -> FeatureNodeType | None: ...

    @abstractmethod
    def node_hash(self) -> int: ...

feature_name instance-attribute

origin_id instance-attribute

__init__(*, feature_name, origin_id)

get_points() abstractmethod

Source code in roc/perception.py

@abstractmethod
def get_points(self) -> set[tuple[XLoc, YLoc]]: ...

node_hash() abstractmethod

Source code in roc/perception.py

@abstractmethod
def node_hash(self) -> int: ...

to_nodes()

Source code in roc/perception.py

def to_nodes(self) -> FeatureNodeType:
    # check local cache
    cache = cache_registry[self.feature_name]

    h = self.node_hash()
    if h in cache:
        return cache[h]  # type: ignore

    # if cache miss, find node in database
    n = self._dbfetch_nodes()
    if n is None:
        # if node doesn't exist, create it
        n = self._create_nodes()
        # n.labels.add("Feature")
        # n.labels.add(self.feature_name)

    cache[h] = n
    return n

FeatureExtractor

Bases: Perception, Generic[FeatureType], ABC

Source code in roc/perception.py

class FeatureExtractor(Perception, Generic[FeatureType], ABC):
    def __init__(self) -> None:
        super().__init__()
        fe_list.append(ref(self))

    def do_perception(self, e: PerceptionEvent) -> None:
        f = self.get_feature(e)
        if f is None:
            return

        self.pb_conn.send(f)

    def settled(self) -> None:
        self.pb_conn.send(Settled())

    @abstractmethod
    def get_feature(self, e: PerceptionEvent) -> Feature[Any] | None: ...

    @classmethod
    def list(cls) -> list[str]:
        ret: list[str] = []
        for fe_ref in fe_list:
            fe = fe_ref()
            if fe is None:
                continue
            ret.append(str(fe.id))

        return ret

__init__()

Source code in roc/perception.py

def __init__(self) -> None:
    super().__init__()
    fe_list.append(ref(self))

do_perception(e)

Source code in roc/perception.py

def do_perception(self, e: PerceptionEvent) -> None:
    f = self.get_feature(e)
    if f is None:
        return

    self.pb_conn.send(f)

get_feature(e) abstractmethod

Source code in roc/perception.py

@abstractmethod
def get_feature(self, e: PerceptionEvent) -> Feature[Any] | None: ...

list() classmethod

Source code in roc/perception.py

@classmethod
def list(cls) -> list[str]:
    ret: list[str] = []
    for fe_ref in fe_list:
        fe = fe_ref()
        if fe is None:
            continue
        ret.append(str(fe.id))

    return ret

settled()

Source code in roc/perception.py

def settled(self) -> None:
    self.pb_conn.send(Settled())

FeatureNode

Bases: Node

Source code in roc/perception.py

class FeatureNode(Node):
    def __hash__(self) -> int:
        # XXX: this is dangerous because ID changes when a node is saved
        # should be okay for this use case though
        return self.id

    def __str__(self) -> str:
        return f"""{self.__class__.__name__}({",".join(self.attr_strs)})"""

    @property
    @abstractmethod
    def attr_strs(self) -> list[str]: ...

attr_strs abstractmethod property

__hash__()

Source code in roc/perception.py

def __hash__(self) -> int:
    # XXX: this is dangerous because ID changes when a node is saved
    # should be okay for this use case though
    return self.id

__str__()

Source code in roc/perception.py

def __str__(self) -> str:
    return f"""{self.__class__.__name__}({",".join(self.attr_strs)})"""

HashingNoneFeature

Bases: Exception

Source code in roc/perception.py

class HashingNoneFeature(Exception):
    pass

Perception

Bases: Component, ABC

The abstract class for Perception components. Handles perception bus connections and corresponding clean-up.

Source code in roc/perception.py

class Perception(Component, ABC):
    """The abstract class for Perception components. Handles perception bus
    connections and corresponding clean-up.
    """

    bus = EventBus[PerceptionData]("perception")

    def __init__(self) -> None:
        super().__init__()
        self.pb_conn = self.connect_bus(Perception.bus)
        self.pb_conn.listen(self.do_perception)

    @abstractmethod
    def do_perception(self, e: PerceptionEvent) -> None: ...

    @classmethod
    def init(cls) -> None:
        global perception_bus
        cls.bus = EventBus[PerceptionData]("perception")

bus = EventBus[PerceptionData]('perception') class-attribute instance-attribute

pb_conn = self.connect_bus(Perception.bus) instance-attribute

__init__()

Source code in roc/perception.py

def __init__(self) -> None:
    super().__init__()
    self.pb_conn = self.connect_bus(Perception.bus)
    self.pb_conn.listen(self.do_perception)

do_perception(e) abstractmethod

Source code in roc/perception.py

@abstractmethod
def do_perception(self, e: PerceptionEvent) -> None: ...

init() classmethod

Source code in roc/perception.py

@classmethod
def init(cls) -> None:
    global perception_bus
    cls.bus = EventBus[PerceptionData]("perception")

PointFeature dataclass

Bases: Feature[FeatureNodeType]

Source code in roc/perception.py

@dataclass(kw_only=True)
class PointFeature(Feature[FeatureNodeType]):
    type: int
    point: tuple[XLoc, YLoc]

    def get_points(self) -> set[tuple[XLoc, YLoc]]:
        return {self.point}

    def node_hash(self) -> int:
        return self.type

point instance-attribute

type instance-attribute

__init__(*, feature_name, origin_id, type, point)

get_points()

Source code in roc/perception.py

def get_points(self) -> set[tuple[XLoc, YLoc]]:
    return {self.point}

node_hash()

Source code in roc/perception.py

def node_hash(self) -> int:
    return self.type

Settled

Source code in roc/perception.py

class Settled:
    pass

VisionData

Vision data received from the environment.

Source code in roc/perception.py

class VisionData:
    """Vision data received from the environment."""

    def __init__(
        self,
        glyphs: npt.NDArray[Any],
        chars: npt.NDArray[Any],
        colors: npt.NDArray[Any],
    ) -> None:
        self.glyphs = glyphs
        self.chars = chars
        self.colors = colors

    @staticmethod
    def from_dict(d: dict[str, Any]) -> VisionData:
        """Creates VisionData from an arbitrary dictionary

        Args:
            d (dict[str, Any]): The dictionary to create VisionData from. Must
            have 'chars', 'glyphs', and 'colors' members.

        Returns:
            VisionData: The newly created vision data.
        """

        def to_numpy(d: dict[str, Any], k: str) -> np.ndarray[Any, Any]:
            if not k in d:
                raise Exception(f"Expected '{k}' to exist in dict for VisionData.from_dict()")

            v = d[k]
            if not isinstance(v, np.ndarray):
                return np.array(v)
            return v

        glyphs = to_numpy(d, "glyphs")
        chars = to_numpy(d, "chars")
        colors = to_numpy(d, "colors")
        return VisionData(glyphs, chars, colors)

    @staticmethod
    def for_test(test_data: list[list[int]]) -> VisionData:
        """Creates VisionData for a test case, using a static 2D list of values
        to create all aspects of the VisionData

        Args:
            test_data (list[list[int]]): The test data to convert into VisionData

        Returns:
            VisionData: The created VisionData
        """
        a = np.array(test_data)
        return VisionData(a.copy(), a.copy(), a.copy())

chars = chars instance-attribute

colors = colors instance-attribute

glyphs = glyphs instance-attribute

__init__(glyphs, chars, colors)

Source code in roc/perception.py

def __init__(
    self,
    glyphs: npt.NDArray[Any],
    chars: npt.NDArray[Any],
    colors: npt.NDArray[Any],
) -> None:
    self.glyphs = glyphs
    self.chars = chars
    self.colors = colors

for_test(test_data) staticmethod

Creates VisionData for a test case, using a static 2D list of values to create all aspects of the VisionData

Parameters:

Name	Type	Description	Default
test_data	list[list[int]]	The test data to convert into VisionData	required

Returns:

Name	Type	Description
VisionData	VisionData	The created VisionData

Source code in roc/perception.py

@staticmethod
def for_test(test_data: list[list[int]]) -> VisionData:
    """Creates VisionData for a test case, using a static 2D list of values
    to create all aspects of the VisionData

    Args:
        test_data (list[list[int]]): The test data to convert into VisionData

    Returns:
        VisionData: The created VisionData
    """
    a = np.array(test_data)
    return VisionData(a.copy(), a.copy(), a.copy())

from_dict(d) staticmethod

Creates VisionData from an arbitrary dictionary

Parameters:

Name	Type	Description	Default
d	dict[str, Any]	The dictionary to create VisionData from. Must	required

Returns:

Name	Type	Description
VisionData	VisionData	The newly created vision data.

Source code in roc/perception.py

@staticmethod
def from_dict(d: dict[str, Any]) -> VisionData:
    """Creates VisionData from an arbitrary dictionary

    Args:
        d (dict[str, Any]): The dictionary to create VisionData from. Must
        have 'chars', 'glyphs', and 'colors' members.

    Returns:
        VisionData: The newly created vision data.
    """

    def to_numpy(d: dict[str, Any], k: str) -> np.ndarray[Any, Any]:
        if not k in d:
            raise Exception(f"Expected '{k}' to exist in dict for VisionData.from_dict()")

        v = d[k]
        if not isinstance(v, np.ndarray):
            return np.array(v)
        return v

    glyphs = to_numpy(d, "glyphs")
    chars = to_numpy(d, "chars")
    colors = to_numpy(d, "colors")
    return VisionData(glyphs, chars, colors)