lpcmanager

Add user-friendly WAMP-SMT scheme for SmartehCloud connection.

14 months ago, Edouard Tisserant
252df96f475c
Parents c996edcec59a
Children 144130828994
Add user-friendly WAMP-SMT scheme for SmartehCloud connection.
2 files changed, 39 insertions(+), 31 deletions(-)
  • +39 -12
    LPCExtension.py
  • +0 -19
    WampAuthentication.py
    • --- a/LPCExtension.py Mon Mar 31 14:27:15 2025 +0200
    +++ b/LPCExtension.py Tue Apr 01 12:08:54 2025 +0200
    @@ -56,16 +56,47 @@
    # --------- Connectors Extension ------------
    #
    -import connectors
    from functools import partial
    +from connectors import WAMP
    +
    +SMARTECLOUD_DEFAULT_HOST = "cloud.smarteh.com"
    +SMARTECLOUD_DEFAULT_PORT = "5678"
    +SMARTECLOUD_DEFAULT_PATH = "/ws"
    +SMARTECLOUD_DEFAULT_REALM = "Automation"
    +
    -# On demand monkey patching
    -def CustomWAMPFactory(*args, **kwargs):
    - from connectors import WAMP
    - from WampAuthentication import WampSession
    - return WAMP._WAMP_connector_factory(WampSession, *args, **kwargs)
    -# TODO
    -# connectors.connectors["WAMP"] = lambda:CustomWAMPFactory
    +def CustomWAMPFactory(uri, *args, **kwargs):
    + scheme, location = uri.split("://")
    + if scheme.upper() == "WAMPS-SMT":
    + if "#" in location:
    + url, anchor = location.split('#',1)
    + if "#" in anchor:
    + realm, plcID = anchor.split("#")
    + else:
    + realm = SMARTECLOUD_DEFAULT_REALM
    + plcID = anchor
    + idx = url.find("/")
    + if idx != -1:
    + urlhost = url[:idx]
    + urlpath = url[idx:]
    + else:
    + urlhost = url
    + urlpath = SMARTECLOUD_DEFAULT_PATH
    + if ":" in urlhost:
    + hostname, port = urlhost.split(":")
    + else:
    + hostname = urlhost
    + port = SMARTECLOUD_DEFAULT_PORT
    + else:
    + hostname = SMARTECLOUD_DEFAULT_HOST
    + port = SMARTECLOUD_DEFAULT_PORT
    + urlpath = SMARTECLOUD_DEFAULT_PATH
    + realm = SMARTECLOUD_DEFAULT_REALM
    + plcID = location
    + uri = "WAMPS-CRT://%s:%s%s#%s#%s"%(hostname, port, urlpath, realm, plcID)
    + return WAMP._WAMP_connector_factory(WAMP.WampSession, uri, *args, **kwargs)
    +
    +WAMP.WAMP_connector_factory = CustomWAMPFactory
    from erpc_interface.erpc_PLCObject.client import BeremizPLCObjectServiceClient
    @@ -113,10 +144,6 @@
    # RunFeedPipe uses no JSON encoding for performance
    setattr(BeremizPLCObjectServiceClient, "RunFeedPipe", JsonOutputMethodProxyFactory("RunFeedPipe"))
    -# TODO
    -# from LPCconnector import LPC_connector_factory
    -# connectors.connectors["LPC"] = lambda: LPC_connector_factory
    -
    #
    # --------- Targets/Toolchains Extension ------------
    #
    --- a/WampAuthentication.py Mon Mar 31 14:27:15 2025 +0200
    +++ /dev/null Thu Jan 01 00:00:00 1970 +0000
    @@ -1,19 +0,0 @@
    -#!/usr/bin/env python
    -# -*- coding: utf-8 -*-
    -
    -# XXX remove that crap, seriously
    -
    -from connectors import WAMP
    -from autobahn.wamp import auth
    -
    -class WampSession(WAMP.WampSession):
    - def onConnect(self):
    - self.join("Automation", ["wampcra"], "smarteh")
    - print("CRA checking ...")
    -
    - def onChallenge(self, challenge):
    - if challenge.method == "wampcra":
    - key = auth.derive_key("smarteh", "salt123", 100, 16)
    - return auth.compute_wcs(key, challenge.extra['challenge'])
    - else:
    - raise Exception("don't know how to handle authmethod {}".format(challenge.method))