S2 pairing frbc example

examples/example_frbc_rm.py

@@ -1,5 +1,4 @@
 import argparse
-import re
 from functools import partial
 import logging
 import sys
@@ -157,7 +156,7 @@ def stop(s2_connection, signal_num, _current_stack_frame):
     print(f"Received signal {signal_num}. Will stop S2 connection.")
     s2_connection.stop()
 
-def start_s2_session(url, client_node_id=str(uuid.uuid4())):
+def start_s2_session(url, client_node_id=str(uuid.uuid4()), bearer_token=None):
     s2_conn = S2Connection(
         url=url,
         role=EnergyManagementRole.RM,
@@ -172,7 +171,8 @@ def start_s2_session(url, client_node_id=str(uuid.uuid4())):
             provides_power_measurements=[CommodityQuantity.ELECTRIC_POWER_L1]
         ),
         reconnect=True,
-        verify_certificate=False
+        verify_certificate=False,
+        bearer_token=bearer_token
     )
     signal.signal(signal.SIGINT, partial(stop, s2_conn))
     signal.signal(signal.SIGTERM, partial(stop, s2_conn))
@@ -181,7 +181,11 @@ def start_s2_session(url, client_node_id=str(uuid.uuid4())):
 
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(description="A simple S2 reseource manager example.")
-    parser.add_argument('endpoint', type=str, help="WebSocket endpoint uri for the server (CEM) e.g. ws://localhost:8080/backend/rm/s2python-frbc/cem/dummy_model/ws")
+    parser.add_argument(
+        'endpoint',
+        type=str,
+        help="WebSocket endpoint uri for the server (CEM) e.h. ws://localhost:8080/websocket/s2/my-first-websocket-rm"
+    )
     args = parser.parse_args()
 
     start_s2_session(args.endpoint)

examples/example_with_pairing_frbc_rm.py

@@ -0,0 +1,41 @@
+import argparse
+import uuid
+import logging
+
+from example_frbc_rm import start_s2_session
+from s2python.s2_pairing import S2Pairing
+from s2python.generated.gen_s2_pairing import S2NodeDescription, Deployment
+from s2python.generated.gen_s2 import EnergyManagementRole
+
+logger = logging.getLogger("s2python")
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="A simple S2 reseource manager example.")
+    parser.add_argument('endpoint',
+                        type=str,
+                        help="Rest endpoint to start S2 pairing. E.g. https://localhost/requestPairing")
+    parser.add_argument('pairing_token',
+                        type=str,
+                        help="The pairing toekn for teh endpoint. You should get this from the S2 server e.g. ca14fda4")
+    args = parser.parse_args()
+
+    nodeDescription: S2NodeDescription = \
+        S2NodeDescription(brand="TNO",
+                          logoUri = "https://www.tno.nl/publish/pages/5604/tno-logo-1484x835_003_.jpg",
+                          type = "demo frbc example",
+                          modelName = "S2 pairing example stub",
+                          userDefinedName = "TNO S2 pairing example for frbc",
+                          role = EnergyManagementRole.RM,
+                          deployment = Deployment.LAN)
+    client_node_id: str = str(uuid.uuid4())
+
+    pairing: S2Pairing = S2Pairing(request_pairing_endpoint = args.endpoint,
+                                   token = args.pairing_token,
+                                   s2_client_node_description = nodeDescription,
+                                   client_node_id = client_node_id)
+
+    logger.info("Pairing details: \n%s", pairing.pairing_details)
+
+    start_s2_session(pairing.pairing_details.connection_details.connectionUri,
+                     bearer_token=pairing.pairing_details.decrypted_challenge)

src/s2python/generated/gen_s2_pairing.py

@@ -8,16 +8,18 @@
 from typing import List
 
 from pydantic import BaseModel, ConfigDict, Field
-from s2python.common import EnergyManagementRole as S2Role
 
 
-
-class Deployment(Enum):
+class S2Role(str, Enum):
+    CEM = 'CEM'
+    RM = 'RM'
+
+class Deployment(str, Enum):
     WAN = 'WAN'
     LAN = 'LAN'
 
 
-class Protocols(Enum):
+class Protocols(str, Enum):
     WebSocketSecure = 'WebSocketSecure'
 
 
@@ -50,7 +52,7 @@ class PairingRequest(BaseModel):
     token: str
     publicKey: str
     s2ClientNodeId: str
-    s2ClientNodeDescription: str
+    s2ClientNodeDescription: S2NodeDescription
     supportedProtocols: List[Protocols]
 
 
@@ -59,7 +61,7 @@ class PairingResponse(BaseModel):
         extra='forbid',
     )
     s2ServerNodeId: str
-    serverNodeDescription: str
+    serverNodeDescription: S2NodeDescription
     requestConnectionUri: str
 
 

src/s2python/s2_pairing.py

@@ -2,11 +2,11 @@
 import uuid
 import datetime
 from dataclasses import dataclass
-from typing import Tuple, Union
+from typing import Tuple, Union, Mapping, Any
+import json
 import requests
 
-from jwskate import JweCompact, Jwk
-from binapy.binapy import BinaPy
+from jwskate import JweCompact, Jwk, Jwt, SignedJwt
 
 from s2python.generated.gen_s2_pairing import (Protocols,
                                                PairingRequest,
@@ -28,10 +28,10 @@ class PairingDetails:
     """The result of an S2 pairing
        :param pairing_response: Details about the server.
        :param connection_details: Details about how to connect.
-       :param supported_protocols: The decrypted challenge needed as bearer token."""
+       :param decrypted_challenge: The decrypted challenge needed as bearer token."""
     pairing_response: PairingResponse
     connection_details: ConnectionDetails
-    decrypted_challenge: BinaPy
+    decrypted_challenge: str
 
 class S2Pairing:  # pylint: disable=too-many-instance-attributes
     _pairing_details: PairingDetails
@@ -77,19 +77,18 @@ def _pair(self) -> None:
         self._paring_timestamp =  datetime.datetime.now()
 
         rsa_key_pair = Jwk.generate_for_alg(KEY_ALGORITHM).with_kid_thumbprint()
-
         pairing_request: PairingRequest = PairingRequest(token=self._token,
                                                         publicKey=rsa_key_pair.public_jwk().to_pem(),
                                                         s2ClientNodeId=self._client_node_id,
                                                         s2ClientNodeDescription=self._s2_client_node_description,
                                                         supportedProtocols=self._supported_protocols)
 
         response = requests.post(self._request_pairing_endpoint,
-                                 json=pairing_request.model_dump_json(),
-                                 timeout=REQTEST_TIMEOUT,
+                                 json = pairing_request.dict(),
+                                 timeout = REQTEST_TIMEOUT,
                                  verify = self._verify_certificate)
         response.raise_for_status()
-        pairing_response: PairingResponse = PairingResponse.parse_raw(response.json())
+        pairing_response: PairingResponse = PairingResponse.parse_raw(response.text)
 
         connection_request: ConnectionRequest = ConnectionRequest(s2ClientNodeId=self._client_node_id,
                                                                    supportedProtocols=self._supported_protocols)
@@ -102,13 +101,14 @@ def _pair(self) -> None:
                                                                                               'requestConnection')
 
         response = requests.post(restest_pairing_uri,
-                                 json=connection_request.model_dump_json(),
-                                 timeout=REQTEST_TIMEOUT,
+                                 json = connection_request.dict(),
+                                 timeout = REQTEST_TIMEOUT,
                                  verify = self._verify_certificate)
         response.raise_for_status()
-        connection_details: ConnectionDetails = ConnectionDetails.parse_raw(response.json())
-        challenge = JweCompact(connection_details.challenge).decrypt(rsa_key_pair)
-        self._pairing_details = PairingDetails(pairing_response, connection_details, challenge)
+        connection_details: ConnectionDetails = ConnectionDetails.parse_raw(response.text)
+        challenge: Mapping[str, Any] = json.loads(JweCompact(connection_details.challenge).decrypt(rsa_key_pair))
+        decrypted_challenge_token: SignedJwt = Jwt.unprotected(challenge)
+        self._pairing_details = PairingDetails(pairing_response, connection_details, str(decrypted_challenge_token))
 
 
     @property

src/s2python/validate_values_mixin.py

@@ -59,7 +59,10 @@ def inner(*args: List[Any], **kwargs: Dict[str, Any]) -> Any:
     return inner
 
 
-def catch_and_convert_exceptions(input_class: Type[S2MessageComponent[B_co]]) -> Type[S2MessageComponent[B_co]]:
+S = TypeVar("S", bound=S2MessageComponent)
+
+
+def catch_and_convert_exceptions(input_class: Type[S]) -> Type[S]:
     input_class.__init__ = convert_to_s2exception(input_class.__init__)  # type: ignore[method-assign]
     input_class.__setattr__ = convert_to_s2exception(input_class.__setattr__)  # type: ignore[method-assign]
     input_class.model_validate_json = convert_to_s2exception(  # type: ignore[method-assign]