feat(robot-server): Robot server protocol session model (#6371)

* ProtocolSessionEvent model created
* rename ProtocolCommandExecutor.commands to events.
* unit tests for protocol command executer
* use timestamp instead of startedAt and completedAt. add '.start' to before and '.end' to after event.
* generate unique ids for protocol sourced events
* use utc_now
* who knew that the 'text' field in the payload of a command could be a format string using the other keys in payload

closes #6225

robot-server/robot_server/service/session/session_types/protocol/execution/command_executor.py

@@ -1,7 +1,8 @@
 import asyncio
 import logging
 import typing
-from dataclasses import asdict
+
+from opentrons.util.helpers import deep_get, utc_now
 
 if typing.TYPE_CHECKING:
     from opentrons.api.dev_types import State
@@ -20,6 +21,7 @@
 from robot_server.service.session.session_types.protocol.execution.worker \
     import _Worker, WorkerListener, WorkerDirective
 from robot_server.util import duration
+from robot_server.service.session.session_types.protocol import models
 
 
 log = logging.getLogger(__name__)
@@ -73,8 +75,8 @@ def __init__(self,
             ProtocolCommand.resume: self._worker.handle_resume,
             ProtocolCommand.pause: self._worker.handle_pause,
         }
-        # TODO: Amit 8/3/2020 - proper schema for command list
-        self._commands: typing.List[typing.Any] = []
+        self._events: typing.List[models.ProtocolSessionEvent] = []
+        self._id_maker = IdMaker()
 
     @staticmethod
     def create_worker(configuration: SessionConfiguration,
@@ -105,9 +107,6 @@ async def execute(self, command: Command) -> CompletedCommand:
                 f"Can't execute '{command_def}' during "
                 f"state '{self.current_state}'")
 
-        # TODO: Amit 8/3/2020 - proper schema for command list
-        self._commands.append(asdict(command))
-
         handler = self._handlers.get(command_def)
         if not handler:
             raise UnsupportedCommandException(
@@ -117,16 +116,24 @@ async def execute(self, command: Command) -> CompletedCommand:
         with duration() as timed:
             await handler()
 
+        self._events.append(
+            models.ProtocolSessionEvent(
+                source=models.EventSource.session_command,
+                event=command.content.name,
+                commandId=command.meta.identifier,
+                timestamp=timed.end,
+            )
+        )
+
         return CompletedCommand(
             content=command.content,
             meta=command.meta,
             result=CommandResult(started_at=timed.start,
                                  completed_at=timed.end))
 
     @property
-    def commands(self):
-        # TODO: Amit 8/3/2020 - proper schema for command list
-        return self._commands
+    def events(self) -> typing.List[models.ProtocolSessionEvent]:
+        return self._events
 
     @property
     def current_state(self) -> 'State':
@@ -162,14 +169,55 @@ async def on_protocol_event(self, cmd: typing.Any):
         topic = cmd.get('topic')
         if topic == Session.TOPIC:
             payload = cmd.get('payload')
-            if isinstance(payload, Session):
-                self.current_state = payload.state
-            else:
+            if isinstance(payload, dict):
                 self.current_state = payload.get('state')
+            elif hasattr(payload, 'state'):
+                self.current_state = payload.state
         else:
-            # TODO: Amit 8/3/2020 - proper schema for command list
-            self._commands.append({
-                    'name': cmd.get('name'),
-                    'desc': cmd['payload']['text'],
-                    'when': cmd.get('$')
-            })
+            dollar_val = cmd.get('$')
+            event_name = cmd.get('name')
+            event = None
+            if dollar_val == 'before':
+                # text may be a format string using the payload vals as kwargs
+                text = deep_get(cmd, ('payload', 'text',), "")
+                if text:
+                    text = text.format(**cmd.get('payload', {}))
+                event = models.ProtocolSessionEvent(
+                    source=models.EventSource.protocol_event,
+                    event=f'{event_name}.start',
+                    commandId=self._id_maker.create_id(),
+                    params={'text': text},
+                    timestamp=utc_now(),
+                )
+            elif dollar_val == 'after':
+                result = deep_get(cmd, ('payload', 'return',))
+                event = models.ProtocolSessionEvent(
+                    source=models.EventSource.protocol_event,
+                    event=f'{event_name}.end',
+                    commandId=self._id_maker.use_last_id(),
+                    timestamp=utc_now(),
+                    result=result,
+                )
+
+            if event:
+                self._events.append(event)
+
+
+class IdMaker:
+    """Helper to create ids for pairs of before/after command pairs"""
+
+    def __init__(self):
+        """Constructor"""
+        self._id_stack: typing.List[str] = []
+        self._next_id = 1
+
+    def create_id(self) -> str:
+        """Create a new id on the stack"""
+        s = str(self._next_id)
+        self._id_stack.append(s)
+        self._next_id += 1
+        return s
+
+    def use_last_id(self) -> str:
+        """Use the the most recently created id"""
+        return self._id_stack.pop()

robot-server/robot_server/service/session/session_types/protocol/models.py

@@ -1,9 +1,32 @@
 import typing
-from pydantic import BaseModel
+from datetime import datetime
+from enum import Enum
+
+from pydantic import BaseModel, Field
+
+
+class EventSource(str, Enum):
+    session_command = "sessionCommand"
+    protocol_event = "protocol"
+
+
+class ProtocolSessionEvent(BaseModel):
+    """An event occurring during a protocol session"""
+    source: EventSource = \
+        Field(..., description="Initiator of this event")
+    event: str = \
+        Field(..., description="The event that occurred")
+    timestamp: datetime
+    commandId: typing.Optional[str] = None
+    params: typing.Optional[typing.Dict[str, typing.Any]] = None
+    result: typing.Optional[str] = None
 
 
 class ProtocolSessionDetails(BaseModel):
-    protocolId: str
+    protocolId: str = \
+        Field(...,
+              description="The protocol used by this session")
     currentState: typing.Optional[str]
-    # TODO: Amit 8/3/2020 - proper schema for command types
-    commands: typing.List[typing.Any]
+    events: typing.List[ProtocolSessionEvent] =\
+        Field(...,
+              description="The events that have occurred thus far")

robot-server/robot_server/service/session/session_types/protocol/session.py

@@ -52,7 +52,7 @@ def _get_response_details(self) -> models.SessionDetails:
         return ProtocolSessionDetails(
             protocolId=self._uploaded_protocol.meta.identifier,
             currentState=self._command_executor.current_state,
-            commands=self._command_executor.commands
+            events=self._command_executor.events
         )
 
     @property

robot-server/tests/integration/sessions/test_protocol.tavern.yaml

@@ -109,7 +109,7 @@ stages:
             details:
               protocolId: "{protocol_id}"
               currentState: !anystr
-              commands: !anylist
+              events: !anylist
         links: !anydict
   - name: Get the session
     request:

robot-server/tests/service/session/session_types/protocol/execution/test_command_executor.py

@@ -1,13 +1,16 @@
-from dataclasses import asdict
-from unittest.mock import MagicMock, patch
+from unittest.mock import MagicMock, patch, PropertyMock
+from datetime import datetime
 import pytest
 
-from robot_server.service.session.command_execution.command import Command,\
-    CommandContent
+from robot_server.service.session.command_execution.command import Command, CommandContent  # noqa: E501
 from robot_server.service.session.errors import UnsupportedCommandException
 from robot_server.service.session.models import ProtocolCommand, EmptyModel
+from robot_server.service.session.session_types.protocol.execution import \
+    command_executor
 from robot_server.service.session.session_types.protocol.execution.command_executor import ProtocolCommandExecutor  # noqa: E501
-from robot_server.service.session.session_types.protocol.execution.worker import _Worker  # noqa:
+from robot_server.service.session.session_types.protocol.execution.worker import _Worker  # noqa: E501
+from robot_server.service.session.session_types.protocol.models import \
+    ProtocolSessionEvent, EventSource
 
 
 @pytest.fixture()
@@ -31,6 +34,18 @@ def protocol_command_executor(mock_worker):
     return ProtocolCommandExecutor(None, None)
 
 
+@pytest.fixture
+def dt() -> datetime:
+    return datetime(2000, 4, 1)
+
+
+@pytest.fixture
+def patch_utc_now(dt: datetime):
+    with patch.object(command_executor, 'utc_now') as p:
+        p.return_value = dt
+        yield p
+
+
 @pytest.mark.parametrize(argnames="current_state,accepted_commands",
                          argvalues=[
                              ['loaded',
@@ -72,7 +87,7 @@ async def test_command_state_reject(loop,
 @pytest.mark.parametrize(argnames="command,worker_method_name",
                          argvalues=[
                              [ProtocolCommand.start_run, "handle_run"],
-                             [ProtocolCommand.start_simulate, "handle_simulate"],  # noqa:
+                             [ProtocolCommand.start_simulate, "handle_simulate"],  # noqa: E501
                              [ProtocolCommand.cancel, "handle_cancel"],
                              [ProtocolCommand.pause, "handle_pause"],
                              [ProtocolCommand.resume, "handle_resume"]
@@ -82,7 +97,7 @@ async def test_execute(loop, command, worker_method_name,
     # Patch the state command filter to allow all commands
     with patch.object(ProtocolCommandExecutor,
                       "STATE_COMMAND_MAP",
-                      new={protocol_command_executor.current_state: ProtocolCommand}):  # noqa:
+                      new={protocol_command_executor.current_state: ProtocolCommand}):  # noqa: E501
         protocol_command = Command(content=CommandContent(
             name=command,
             data=EmptyModel())
@@ -91,4 +106,150 @@ async def test_execute(loop, command, worker_method_name,
         # Worker handler was called
         getattr(mock_worker, worker_method_name).assert_called_once()
         # Command is added to command list
-        assert protocol_command_executor.commands == [asdict(protocol_command)]
+        assert len(protocol_command_executor.events) == 1
+        assert protocol_command_executor.events[0].event == command
+
+
+class TestOnProtocolEvent:
+
+    async def test_topic_session_payload(self, protocol_command_executor):
+        mock_session = MagicMock()
+        mock_session.state = "some_state"
+        await protocol_command_executor.on_protocol_event({
+            'topic': 'session',
+            'payload': mock_session
+        })
+        assert protocol_command_executor.current_state == "some_state"
+
+    async def test_topic_dict_payload(self, protocol_command_executor):
+        await protocol_command_executor.on_protocol_event({
+            'topic': 'session',
+            'payload': {
+                'state': 'some_state'
+            }
+        })
+        assert protocol_command_executor.current_state == "some_state"
+
+    @pytest.mark.parametrize(argnames="payload",
+                             argvalues=[{},
+                                        {'topic': 'soosion'},
+                                        {'$': 'soosion'}
+                                        ])
+    async def test_body_invalid(self, protocol_command_executor, payload):
+        ProtocolCommandExecutor.current_state = PropertyMock()
+        await protocol_command_executor.on_protocol_event(payload)
+        assert len(protocol_command_executor.events) == 0
+        ProtocolCommandExecutor.current_state.assert_not_called()
+
+    async def test_before(self, protocol_command_executor, patch_utc_now, dt):
+        payload = {
+            '$': 'before',
+            'name': 'some event',
+            'payload': {
+                'text': 'this is what happened'
+            }
+        }
+        await protocol_command_executor.on_protocol_event(payload)
+        assert protocol_command_executor.events == [
+            ProtocolSessionEvent(source=EventSource.protocol_event,
+                                 event="some event.start",
+                                 commandId="1",
+                                 timestamp=dt,
+                                 params={'text': 'this is what happened'})
+        ]
+
+    async def test_before_text_is_none(self, protocol_command_executor,
+                                       patch_utc_now, dt):
+        payload = {
+            '$': 'before',
+            'name': 'some event',
+            'payload': {
+                'text': None
+            }
+        }
+        await protocol_command_executor.on_protocol_event(payload)
+        assert protocol_command_executor.events == [
+            ProtocolSessionEvent(source=EventSource.protocol_event,
+                                 event="some event.start",
+                                 commandId="1",
+                                 timestamp=dt,
+                                 params={'text': None})
+        ]
+
+    async def test_before_text_is_format_string(self,
+                                                protocol_command_executor,
+                                                patch_utc_now,
+                                                dt):
+        payload = {
+            '$': 'before',
+            'name': 'some event',
+            'payload': {
+                'text': '{oh} {no}',
+                'oh': 2,
+                'no': 5
+            }
+        }
+        await protocol_command_executor.on_protocol_event(payload)
+        assert protocol_command_executor.events == [
+            ProtocolSessionEvent(source=EventSource.protocol_event,
+                                 event="some event.start",
+                                 commandId="1",
+                                 timestamp=dt,
+                                 params={'text': '2 5'})
+        ]
+
+    async def test_after(self, protocol_command_executor, patch_utc_now, dt):
+        payload = {
+            '$': 'after',
+            'name': 'some event',
+            'payload': {
+                'text': 'this is it',
+                'return': "done"
+            }
+        }
+        protocol_command_executor._id_maker.create_id()
+        await protocol_command_executor.on_protocol_event(payload)
+        assert protocol_command_executor.events == [
+            ProtocolSessionEvent(source=EventSource.protocol_event,
+                                 event="some event.end",
+                                 commandId="1",
+                                 timestamp=dt,
+                                 result="done"
+                                 )
+        ]
+
+    async def test_ids(self, protocol_command_executor, patch_utc_now, dt):
+        before = {
+            '$': 'before',
+            'name': 'event'
+        }
+        after = {
+            '$': 'after',
+            'name': 'event'
+        }
+
+        await protocol_command_executor.on_protocol_event(before)
+        await protocol_command_executor.on_protocol_event(after)
+
+        await protocol_command_executor.on_protocol_event(before)
+        await protocol_command_executor.on_protocol_event(before)
+        await protocol_command_executor.on_protocol_event(after)
+        await protocol_command_executor.on_protocol_event(after)
+
+        await protocol_command_executor.on_protocol_event(before)
+        await protocol_command_executor.on_protocol_event(before)
+        await protocol_command_executor.on_protocol_event(before)
+        await protocol_command_executor.on_protocol_event(after)
+        await protocol_command_executor.on_protocol_event(after)
+        await protocol_command_executor.on_protocol_event(after)
+
+        await protocol_command_executor.on_protocol_event(before)
+        await protocol_command_executor.on_protocol_event(after)
+
+        ids = [x.commandId for x in protocol_command_executor.events]
+        assert ids == [
+            "1", "1",
+            "2", "3", "3", "2",
+            "4", "5", "6", "6", "5", "4",
+            "7", "7"
+        ]