Hands on Object Relational Mapping in CoraPlex

This tutorial will walk you through the serialization of a minimal plan in coraplex. First we will import sqlalchemy, create an in-memory database and connect a session to it.

import sqlalchemy.orm
from krrood.ormatic.utils import create_engine

engine = create_engine("sqlite+pysqlite:///:memory:", echo=False)
session = sqlalchemy.orm.Session(engine)

Next, we need a mapper_registry to map our classes to the database tables. We will use the default mapper_registry from sqlalchemy.

import coraplex.orm.ormatic_interface
from coraplex.orm.ormatic_interface import *

coraplex.orm.ormatic_interface.Base.metadata.create_all(engine)

Next, we will write a simple plan where the robot raises its torso and transports a milk carton: it parks its arms, navigates to the milk, faces and looks at it, picks it up, navigates to the target, faces and looks at it, and places the milk there.

Only grounded actions can be stored, so every step of the transport is given here: where the robot stands, which grasp it takes the milk by and with which arm. A step that is still an EQL query, as the ones TransportAction.from_graspable_by_closest_grasps builds, is refused with a QueryCannotBePersisted.

from coraplex.robot_plans import *
from coraplex.robot_plans.actions.composite.transporting import (
    MoveAndPickUpAction,
    MoveAndPlaceAction,
    TransportAction,
)
from coraplex.robot_plans.actions.core.robot_body import MoveTorsoAction
from coraplex.testing import setup_world
from semantic_digital_twin.robots.pr2 import PR2, TorsoState
from semantic_digital_twin.semantic_annotations.semantic_annotations import Milk
from krrood_spatial_math.spatial_types import Pose
from cramph.composites import Sequence

world = setup_world()
pr2_view = PR2.from_world(world)
from coraplex.plans.context_extensions import RobotAccess
from cramph.context import StatechartContext
from coraplex.plans.executors import KinematicallySimulatedPlanExecutor
from cramph.statechart import Statechart

extensions = [RobotAccess(pr2_view)]


def run(plan):
    """
    Run `plan` simulated in `world`, with the robot and settings in `extensions`.
    """
    executor = KinematicallySimulatedPlanExecutor(
        StatechartContext.with_extensions(world, extensions)
    )
    statechart = Statechart(context=executor.context)
    statechart.add_node(plan)
    executor.compile(statechart)
    executor.execute()
    return executor


milk = world.get_semantic_annotations_by_type(Milk)[0]
description = TransportAction(
    pick_up=MoveAndPickUpAction.from_standing_position(
        standing_position=Pose.from_xyz_rpy(
            1.63, 1.98, 0.0, reference_frame=world.root
        ),
        grasp=milk.grasp_candidates()[0],
        arm=pr2_view.left_arm,
    ),
    place=MoveAndPlaceAction.from_standing_position(
        standing_position=Pose.from_xyz_rpy(
            1.64, 3.24, 0.0, reference_frame=world.root
        ),
        target_location=Pose.from_xyz_quaternion(
            2.4, 3, 1.05, 0.0, 0.0, 0.0, 1.0, reference_frame=world.root
        ),
        object_designator=milk,
    ),
)
plan = Sequence(nodes=[MoveTorsoAction(torso_state=TorsoState.HIGH), description])
run(plan)

The data obtained throughout the plan execution, including robot states, poses, action descriptions and more will be logged into the database once we insert the plan.

from krrood.ormatic.data_access_objects.helper import to_dao, get_data_access_object_class

session.add(to_dao(plan))
session.commit()

Now we can query the database to see what we have logged. Let’s say we want to see all the NavigateActions that occurred.

from sqlalchemy import select
from coraplex.robot_plans.actions.core.navigation import NavigateAction

navigations = session.scalars(select(get_data_access_object_class(NavigateAction))).all()
print(*navigations, sep="\n")

This should print both NavigateActions of the plan: the one to where the robot picks up the milk and the one to where it places it.

Due to the inheritance mapped in the ORM package, we can also get all executed actions with just one query.

from coraplex.robot_plans.actions.base import Action

actions = session.scalars(select(get_data_access_object_class(Action))).all()
print(*actions, sep="\n")

This should print every action that occurred during the plan execution, including the steps the composite actions are made of: besides the MoveTorsoAction and the TransportAction, it lists the TransportAction’s ParkArmsActions, its MoveAndPickUpAction and MoveAndPlaceAction, and the steps inside those, such as the NavigateActions, the PickUpAction and the PlaceAction.

If you want to know more about the memory component, read the documentation of the KRR component.