Add signature to adapters so that help works as expected

csp/impl/wiring/adapters.py

@@ -1,4 +1,6 @@
+import inspect
 from datetime import timedelta
+from typing_extensions import override
 
 from csp.impl.__cspimpl import _cspimpl
 from csp.impl.mem_cache import csp_memoized_graph_object
@@ -43,6 +45,20 @@ def __call__(cls, *args, **kwargs):
     def using(cls, name=None, **__forced_tvars):
         return lambda *args, **kwargs: cls._instantiate(__forced_tvars, name, *args, **kwargs)
 
+    @property
+    def __signature__(cls):
+        # Implement so that `help` works properly on adapter definitions.
+        parameters = [
+            inspect.Parameter(
+                input_def.name,
+                inspect.Parameter.POSITIONAL_OR_KEYWORD,
+                annotation=input_def.typ,
+                default=cls._signature.defaults.get(input_def.name, inspect.Parameter.empty),
+            )
+            for input_def in cls._signature.inputs
+        ]
+        return inspect.Signature(parameters)
+
 
 # Every AdapterDef instance represents an instance of a wiring-time input or output adapter
 class AdapterDef:
@@ -348,7 +364,18 @@ def impl(mgr, engine, scalars):
     )
 
 
-add_graph_output = output_adapter_def(
+@override
+def add_graph_output(
+    key: object,
+    input: tstype.ts["T"],  # noqa: F821
+    tick_count: int = -1,
+    tick_history: timedelta = timedelta(),
+):
+    # Stub for IDE auto-complete/static type checking
+    ...
+
+
+add_graph_output = output_adapter_def(  # noqa: F811
     "add_graph_output",
     _cspimpl._graph_output_adapter,
     key=object,

csp/impl/wiring/signature.py

@@ -273,3 +273,7 @@ def ts_inputs(self):
     @property
     def scalars(self):
         return self._scalars
+
+    @property
+    def defaults(self):
+        return self._defaults

csp/tests/impl/test_outputadapter.py

@@ -1,5 +1,6 @@
 """this test is derived from e_14_user_adapters_05 and e_14_user_adapters_06"""
 
+import inspect
 import random
 import threading
 import unittest
@@ -198,3 +199,9 @@ def test_with_manager(self):
                     self.assertIn("publication_data_1", entry)
                 elif "symbol=data_3" in entry:
                     self.assertIn("publication_data_3", entry)
+
+    def test_help(self):
+        # for `help` to work on output adapters, signature must be defined
+        sig = inspect.signature(MyBufferWriterAdapter)
+        self.assertEqual(sig.parameters["input"].annotation, ts["T"])
+        self.assertEqual(sig.parameters["output_buffer"].annotation, list)

csp/tests/impl/test_pushadapter.py

@@ -1,3 +1,4 @@
+import inspect
 import threading
 import time
 import unittest
@@ -240,6 +241,15 @@ def graph():
         result = list(x[1] for x in result)
         self.assertEqual(result, expected)
 
+    def test_help(self):
+        # for `help` to work on adapters, signature must be defined
+        sig = inspect.signature(test_adapter)
+        self.assertEqual(sig.parameters["typ"].annotation, "T")
+        self.assertEqual(sig.parameters["interval"].annotation, int)
+        self.assertEqual(sig.parameters["ticks_per_interval"].annotation, int)
+        self.assertEqual(sig.parameters["push_mode"].annotation, PushMode)
+        self.assertEqual(sig.parameters["push_group"].annotation, object)
+
 
 if __name__ == "__main__":
     unittest.main()