Merge pull request #1751 from fetchai/feature/soef-generic-command

AEA-904 soef generic command. integration tests added

packages/fetchai/connections/soef/connection.py

@@ -21,6 +21,7 @@
 import asyncio
 import copy
 import logging
+import urllib
 from asyncio import CancelledError
 from concurrent.futures._base import CancelledError as ConcurrentCancelledError
 from concurrent.futures.thread import ThreadPoolExecutor
@@ -88,6 +89,7 @@ class ModelNames:
     personality_agent = "personality_agent"
     search_model = "search_model"
     ping = "ping"
+    generic_command = "generic_command"
 
 
 class SOEFException(Exception):
@@ -484,6 +486,7 @@ async def register_service(  # pylint: disable=unused-argument
             "personality_agent": self._set_personality_piece_handler,
             "set_service_key": self._set_service_key_handler,
             "ping": self._ping_handler,
+            "generic_command": self._generic_command_handler,
         }  # type: Dict[str, Callable]
         data_model_name = service_description.data_model.name
 
@@ -493,9 +496,14 @@ async def register_service(  # pylint: disable=unused-argument
             )
 
         handler = data_model_handlers[data_model_name]
-        await handler(service_description)
+        await handler(service_description, oef_message, oef_search_dialogue)
 
-    async def _ping_handler(self, service_description: Description) -> None:
+    async def _ping_handler(
+        self,
+        service_description: Description,
+        oef_message: OefSearchMessage,  # pylint: disable=unused-argument
+        oef_search_dialogue: OefSearchDialogue,  # pylint: disable=unused-argument
+    ) -> None:
         """
         Perform ping command.
 
@@ -506,6 +514,51 @@ async def _ping_handler(self, service_description: Description) -> None:
         self._check_data_model(service_description, ModelNames.ping)
         await self._ping_command()
 
+    async def _generic_command_handler(
+        self,
+        service_description: Description,
+        oef_message: OefSearchMessage,
+        oef_search_dialogue: OefSearchDialogue,
+    ) -> None:
+        """
+        Perform ping command.
+
+        :param service_description: Service description
+
+        :return None
+        """
+        if not self.in_queue:  # pragma: no cover
+            """not connected."""
+            return
+
+        self._check_data_model(service_description, ModelNames.generic_command)
+        command = service_description.values.get("command", None)
+        params = service_description.values.get("parameters", {})
+
+        if params:
+            params = urllib.parse.parse_qs(params)
+
+        content = await self._generic_oef_command(command, params)
+
+        message = oef_search_dialogue.reply(
+            performative=OefSearchMessage.Performative.SUCCESS,
+            target_message=oef_message,
+            agents_info=AgentsInfo(
+                {
+                    "response": {"content": content},
+                    "command": service_description.values,
+                }
+            ),
+        )
+        envelope = Envelope(
+            to=message.to,
+            sender=message.sender,
+            protocol_id=message.protocol_id,
+            message=message,
+            context=oef_search_dialogue.envelope_context,
+        )
+        await self.in_queue.put(envelope)
+
     async def _ping_command(self) -> None:
         """Perform ping on registered agent."""
         await self._generic_oef_command("ping", {})
@@ -528,7 +581,12 @@ async def _ping_periodic(self, period: float = 30 * 60) -> None:
                     self.logger.exception("Error on periodic ping command!")
                 await asyncio.sleep(period)
 
-    async def _set_service_key_handler(self, service_description: Description) -> None:
+    async def _set_service_key_handler(
+        self,
+        service_description: Description,
+        oef_message: OefSearchMessage,  # pylint: disable=unused-argument
+        oef_search_dialogue: OefSearchDialogue,  # pylint: disable=unused-argument
+    ) -> None:
         """
         Set service key from service description.
 
@@ -585,7 +643,10 @@ async def _set_service_key(self, key: str, value: Union[str, int, float]) -> Non
         await self._generic_oef_command("set_service_key", {"key": key, "value": value})
 
     async def _remove_service_key_handler(
-        self, service_description: Description
+        self,
+        service_description: Description,
+        oef_message: OefSearchMessage,  # pylint: disable=unused-argument
+        oef_search_dialogue: OefSearchDialogue,  # pylint: disable=unused-argument
     ) -> None:
         """
         Remove service key from service description.
@@ -611,7 +672,10 @@ async def _remove_service_key(self, key: str) -> None:
         await self._generic_oef_command("remove_service_key", {"key": key})
 
     async def _register_location_handler(
-        self, service_description: Description
+        self,
+        service_description: Description,
+        oef_message: OefSearchMessage,  # pylint: disable=unused-argument
+        oef_search_dialogue: OefSearchDialogue,  # pylint: disable=unused-argument
     ) -> None:
         """
         Register service with location.
@@ -686,7 +750,10 @@ async def _set_location(
         self.agent_location = agent_location
 
     async def _set_personality_piece_handler(
-        self, service_description: Description
+        self,
+        service_description: Description,
+        oef_message: OefSearchMessage,  # pylint: disable=unused-argument
+        oef_search_dialogue: OefSearchDialogue,  # pylint: disable=unused-argument
     ) -> None:
         """
         Set the personality piece.
@@ -826,9 +893,9 @@ async def unregister_service(  # pylint: disable=unused-argument
         if data_model_name == "location_agent":
             await handler()
         else:
-            await handler(service_description)
+            await handler(service_description, oef_message, oef_search_dialogue)
 
-    async def _unregister_agent(self) -> None:
+    async def _unregister_agent(self) -> None:  # pylint: disable=unused-argument
         """
         Unnregister a service_name from the SOEF.
 

packages/fetchai/connections/soef/connection.yaml

@@ -8,7 +8,7 @@ aea_version: '>=0.6.0, <0.7.0'
 fingerprint:
   README.md: QmWwjETX39APP9RD5oVPeeCiDnUoDvjAcoVe2FK2Jc6anM
   __init__.py: Qmd5VBGFJHXFe1H45XoUh5mMSYBwvLSViJuGFeMgbPdQts
-  connection.py: QmW6JjJsAybe2w4CrjahvLuV3n3nGUXMTzYPQ7LmTq35E6
+  connection.py: QmfK5khjQuqewKVuqyedgmToQvRbh69oT8f5WcSvzvvjJL
 fingerprint_ignore_patterns: []
 protocols:
 - fetchai/oef_search:0.6.0

packages/fetchai/protocols/oef_search/README.md

@@ -25,7 +25,8 @@ speech_acts:
   search_result:
     agents: pt:list[pt:str]
     agents_info: ct:AgentsInfo
-  success: {}
+  success:
+    agents_info: ct:AgentsInfo
   oef_error:
     oef_error_operation: ct:OefErrorOperation
 ...

packages/fetchai/protocols/oef_search/message.py

@@ -255,7 +255,13 @@ def _is_consistent(self) -> bool:
                     ),
                 )
             elif self.performative == OefSearchMessage.Performative.SUCCESS:
-                expected_nb_of_contents = 0
+                expected_nb_of_contents = 1
+                enforce(
+                    type(self.agents_info) == CustomAgentsInfo,
+                    "Invalid type for content 'agents_info'. Expected 'AgentsInfo'. Found '{}'.".format(
+                        type(self.agents_info)
+                    ),
+                )
             elif self.performative == OefSearchMessage.Performative.OEF_ERROR:
                 expected_nb_of_contents = 1
                 enforce(

packages/fetchai/protocols/oef_search/oef_search.proto

@@ -52,7 +52,9 @@ message OefSearchMessage{
         AgentsInfo agents_info = 2;
     }
 
-    message Success_Performative{}
+    message Success_Performative{
+        AgentsInfo agents_info = 1;
+    }
 
     message Oef_Error_Performative{
         OefErrorOperation oef_error_operation = 1;

packages/fetchai/protocols/oef_search/protocol.yaml

@@ -6,14 +6,14 @@ description: A protocol for interacting with an OEF search service.
 license: Apache-2.0
 aea_version: '>=0.6.0, <0.7.0'
 fingerprint:
-  README.md: QmfGEYrF54F6i5RTGgLgzfhMCMgYBY8Q1iHwb9UhE96MDi
+  README.md: QmaGSTqxvQFKccBnLovhBbfSH3C3Sorrj7kFyZqW9qptLa
   __init__.py: QmRvTtynKcd7shmzgf8aZdcA5witjNL5cL2a7WPgscp7wq
   custom_types.py: QmYAkKYj9gGHaij7uTejoJe9KRhNcsU4sJC1utMfhUYhg3
   dialogues.py: QmQPLnW3jAs6tLLmhkX4C7texGRHM9bfdjs83dUH5TkJ4v
-  message.py: QmXj8MgvPtJKXZyhTnTfpXEZoAz4Z1u6wbMwMkKCRgvujL
-  oef_search.proto: QmfZrvXMHWTNp7QV4Vpm3TwU2h34PRHM8KkEWUVJWc7ihS
-  oef_search_pb2.py: QmWUYB2StNy6BqPzJjyc2WrA7y1dWFxF81m8BwRBBThDMC
-  serialization.py: QmdCFKKquz26fFRa3qguLgFa1GrGmCWgfLbmNmgtTSePAH
+  message.py: QmU8jH94qxrcr9eUtXWn5PzqmHT7NBc62gs53HPXKVk1Ts
+  oef_search.proto: QmNU8WsxT6XNFFneKKeDaZkNn3CEFDfZQkmKv9TyhyxzDB
+  oef_search_pb2.py: QmSAFT1xxYRzJU6h1aFVDuYcx672sZ2vzV6c2ico7f4BLK
+  serialization.py: QmVmZnv4kHr2E9UaYACiwnTuT79N4hSe7NZZ2ogB2bD2K2
 fingerprint_ignore_patterns: []
 dependencies:
   protobuf: {}

packages/fetchai/protocols/oef_search/serialization.py

@@ -76,6 +76,8 @@ def encode(msg: Message) -> bytes:
             oef_search_msg.search_result.CopyFrom(performative)
         elif performative_id == OefSearchMessage.Performative.SUCCESS:
             performative = oef_search_pb2.OefSearchMessage.Success_Performative()  # type: ignore
+            agents_info = msg.agents_info
+            AgentsInfo.encode(performative.agents_info, agents_info)
             oef_search_msg.success.CopyFrom(performative)
         elif performative_id == OefSearchMessage.Performative.OEF_ERROR:
             performative = oef_search_pb2.OefSearchMessage.Oef_Error_Performative()  # type: ignore
@@ -132,7 +134,9 @@ def decode(obj: bytes) -> Message:
             agents_info = AgentsInfo.decode(pb2_agents_info)
             performative_content["agents_info"] = agents_info
         elif performative_id == OefSearchMessage.Performative.SUCCESS:
-            pass
+            pb2_agents_info = oef_search_pb.success.agents_info
+            agents_info = AgentsInfo.decode(pb2_agents_info)
+            performative_content["agents_info"] = agents_info
         elif performative_id == OefSearchMessage.Performative.OEF_ERROR:
             pb2_oef_error_operation = oef_search_pb.oef_error.oef_error_operation
             oef_error_operation = OefErrorOperation.decode(pb2_oef_error_operation)

packages/hashes.csv

@@ -28,7 +28,7 @@ fetchai/connections/p2p_libp2p,QmdhTkGsBwQ7Zrogp5eW1FcC6p1dii7TDpsmy5h9eRMxb9
 fetchai/connections/p2p_libp2p_client,QmU1bNVaYjipA1uwad4CGxMgsMSNaP9n6E11SwvF2in9Jh
 fetchai/connections/p2p_stub,QmX8EWvFok3UAdqXGgir4TrhfDj2kLhfjuXRg82b6G6XtW
 fetchai/connections/scaffold,QmNUta43nLexAHaXLgmLQYEjntLXSV6MLNvc6Q2CTx7eBS
-fetchai/connections/soef,QmYPr7sPQXUz2qKJ1ifomwmtUTv6aHGRdzEbuVJ1ZbzmCn
+fetchai/connections/soef,QmQWyjmHg6HWsLEuighe2vrWQSKBZAXvoXSXF33XhaVyE8
 fetchai/connections/stub,QmSuUAA2k1E1M9dpeAgkWpfKdssjqghSDfFfrQzC9ZLKbD
 fetchai/connections/tcp,QmU93B7yajCHwXsgDXXhv1uSuXi7VmFBhdahcFosbrr6nA
 fetchai/connections/webhook,QmaJunWQT1ypkqGnzrj5gYjjQBywjTHJmyktCZXZsa8hv8
@@ -41,7 +41,7 @@ fetchai/protocols/gym,QmVLULhH4gn8suFe8Z31X5mVYp3fFSGays5MHDNzKnGA4k
 fetchai/protocols/http,QmZJVqY2cxjky6p3o76TUemSsFDws8Dx33voG9WMdrKcry
 fetchai/protocols/ledger_api,Qmak3uKpeSfT5f4oqes3oUcNSg7vy7WKE2tUmSu8CZUNJ5
 fetchai/protocols/ml_trade,QmNVZvbcgERSypYUopKMHvd6F8B81rQqtLVqUvzMeEEwJN
-fetchai/protocols/oef_search,QmfKNSFNLtLimS1YSiWB9e8Ha5xUzFkFjRUXwFxELCgV41
+fetchai/protocols/oef_search,QmYjitTtVuVynTWv4TE34Wm2b37LAfyKgZEaiMfZv1jH3Z
 fetchai/protocols/scaffold,QmZhHsoA7JzNSoSUABFWqyRELaei4BtKYce1QKVcHhnQJN
 fetchai/protocols/signing,QmPhHfj2N2iCRTMEDYiifC3AkytZRFpVMoH7g63NwGcKwH
 fetchai/protocols/state_update,QmQmQt1VHwxmnNmwxKcrF4Qno6iHzVhhnUUjsqPtpfpQUV

tests/test_packages/test_connections/test_soef/models.py

@@ -67,3 +67,13 @@
     [Attribute("location", Location, True, "The location where the agent is.")],
     "A data model to perform search.",
 )
+
+
+AGENT_GENERIC_COMMAND_MODEL = DataModel(
+    ModelNames.generic_command,
+    [
+        Attribute("command", str, True, "Command name to execute."),
+        Attribute("parameters", str, False, "Url encoded parameters string."),
+    ],
+    "A data model to describe the generic soef command.",
+)