Feature: Add PublishValue and Subscribe from kuksa.val.v2

.github/workflows/kuksa-client.yml

@@ -132,13 +132,18 @@ jobs:
       - name: Checkout kuksa-python-sdk
         uses: actions/checkout@v4
         with:
+          fetch-depth: 0
           submodules: 'true'
       - name: Install pip
         run: |
           python -m pip --quiet --no-input install --upgrade pip
+      - name: Initiate submodules
+        run: |
+          git submodule update --remote
       - name: Install dependencies with pip
         run: |
           cd kuksa-client
+          python3 -m proto
           pip install -r requirements.txt -e .
           pip install -r test-requirements.txt
       - name: Run tests

.gitmodules

@@ -1,3 +1,6 @@
-[submodule "submodules/kuksa-databroker"]
+
+
+[submodule "kuksa-databroker"]
 	path = submodules/kuksa-databroker
-	url = https://github.com/eclipse-kuksa/kuksa-databroker
+	url = https://github.com/eclipse-kuksa/kuksa-databroker.git
+	branch = feature/databroker-api-v2

docs/building.md

@@ -7,6 +7,12 @@ git submodule update --init
 cd kuksa-client
 ```
 
+Hint: If you want to use another branch than master exchange the first command with
+
+```console
+git submodule update --recursive --remote --init
+```
+
 First we suggest you create a dedicated [python virtual environment](https://docs.python.org/3/library/venv.html) for kuksa-client:
 
 ```console
@@ -15,6 +21,12 @@ python3 -m venv ~/.venv/kuksa-client
 source ~/.venv/kuksa-client/bin/activate  # Run this every time you want to activate kuksa-client's virtual environment
 ```
 
+To use the right api interfaces of databroker run the following:
+```console
+python3 -m proto
+```
+This should copy the corresponding proto files to the kuksa-client directory.
+
 Your prompt should change to somehting indicating you are in the virutal environment now, e.g.
 
 ```console

kuksa-client/Dockerfile

@@ -18,6 +18,9 @@ RUN pip install --upgrade pip build pyinstaller
 # We must copy the whole repository otherwise version lookup by tag would not work
 COPY . /kuksa-python-sdk/
 WORKDIR /kuksa-python-sdk/kuksa-client
+RUN git submodule update --recursive --remote --init
+# install files from submodules to kuksa-client repo to generate protos out of it
+RUN python3 -m proto
 
 RUN python3 -m build
 # We install globally on build container, so pyinstaller can easily gather all files

kuksa-client/kuksa_client/__main__.py

@@ -136,13 +136,14 @@ def path_completer(self, text, line, begidx, endidx):
 
     def subscribeCallback(self, logPath, resp):
         if logPath is None:
-            self.async_alert(
-                highlight(
-                    json.dumps(json.loads(resp), indent=2),
-                    lexers.JsonLexer(),
-                    formatters.TerminalFormatter(),
+            with self.terminal_lock:
+                self.async_alert(
+                    highlight(
+                        json.dumps(json.loads(resp), indent=2),
+                        lexers.JsonLexer(),
+                        formatters.TerminalFormatter(),
+                    )
                 )
-            )
         else:
             with logPath.open("a", encoding="utf-8") as logFile:
                 logFile.write(resp + "\n")

kuksa-client/kuksa_client/cli_backend/grpc.py

@@ -81,9 +81,18 @@ def __init__(self, config):
         self.run = False
 
         self.AttrDict = {
-            "value": (kuksa_client.grpc.Field.VALUE, kuksa_client.grpc.View.CURRENT_VALUE),
-            "targetValue": (kuksa_client.grpc.Field.ACTUATOR_TARGET, kuksa_client.grpc.View.TARGET_VALUE),
-            "metadata": (kuksa_client.grpc.Field.METADATA, kuksa_client.grpc.View.METADATA),
+            "value": (
+                kuksa_client.grpc.Field.VALUE,
+                kuksa_client.grpc.View.CURRENT_VALUE,
+            ),
+            "targetValue": (
+                kuksa_client.grpc.Field.ACTUATOR_TARGET,
+                kuksa_client.grpc.View.TARGET_VALUE,
+            ),
+            "metadata": (
+                kuksa_client.grpc.Field.METADATA,
+                kuksa_client.grpc.View.METADATA,
+            ),
         }
 
     def connection_established(self) -> bool:
@@ -112,8 +121,10 @@ def getValue(self, path: str, attribute="value", timeout=5):
     def getValues(self, paths: Iterable[str], attribute="value", timeout=5):
         if attribute in self.AttrDict:
             field, view = self.AttrDict[attribute]
-            entries = [kuksa_client.grpc.EntryRequest(
-                path=path, view=view, fields=(field,)) for path in paths]
+            entries = [
+                kuksa_client.grpc.EntryRequest(path=path, view=view, fields=(field,))
+                for path in paths
+            ]
             requestArgs = {'entries': entries}
             return self._sendReceiveMsg(("get", requestArgs), timeout)
 
@@ -127,28 +138,33 @@ def setValues(self, updates: Dict[str, Any], attribute="value", timeout=5):
         if attribute in self.AttrDict:
             field, _ = self.AttrDict[attribute]
             entry_updates = []
+            try_v2 = False
             for path, value in updates.items():
 
                 if field is kuksa_client.grpc.Field.VALUE:
                     entry = kuksa_client.grpc.DataEntry(
-                        path=path, value=kuksa_client.grpc.Datapoint(value=value))
+                        path=path,
+                        value=kuksa_client.grpc.Datapoint(value=value),
+                    )
+                    try_v2 = True
                 elif field is kuksa_client.grpc.Field.ACTUATOR_TARGET:
                     entry = kuksa_client.grpc.DataEntry(
-                        path=path, actuator_target=kuksa_client.grpc.Datapoint(
-                            value=value),
+                        path=path,
+                        actuator_target=kuksa_client.grpc.Datapoint(value=value),
                     )
                 elif field is kuksa_client.grpc.Field.METADATA:
                     try:
                         metadata_dict = json.loads(value)
                     except json.JSONDecodeError:
                         return json.dumps({"error": "Metadata value needs to be a valid JSON object"})
                     entry = kuksa_client.grpc.DataEntry(
-                        path=path, metadata=kuksa_client.grpc.Metadata.from_dict(
-                            metadata_dict),
+                        path=path,
+                        metadata=kuksa_client.grpc.Metadata.from_dict(metadata_dict),
                     )
-                entry_updates.append(kuksa_client.grpc.EntryUpdate(
-                    entry=entry, fields=(field,)))
-            requestArgs = {'updates': entry_updates}
+                entry_updates.append(
+                    kuksa_client.grpc.EntryUpdate(entry=entry, fields=(field,))
+                )
+            requestArgs = {"updates": entry_updates, "try_v2": try_v2}
             return self._sendReceiveMsg(("set", requestArgs), timeout)
         return json.dumps({"error": "Invalid Attribute"})
 
@@ -175,11 +191,14 @@ def subscribe(self, path: str, callback, attribute="value", timeout=5):
     def subscribeMultiple(self, paths: Iterable[str], callback, attribute="value", timeout=5):
         if attribute in self.AttrDict:
             field, view = self.AttrDict[attribute]
-            entries = [kuksa_client.grpc.SubscribeEntry(
-                path=path, view=view, fields=(field,)) for path in paths]
+            entries = [
+                kuksa_client.grpc.SubscribeEntry(path=path, view=view, fields=(field,))
+                for path in paths
+            ]
             requestArgs = {
-                'entries': entries,
-                'callback': callback_wrapper(callback),
+                "entries": entries,
+                "try_v2": True,
+                "callback": callback_wrapper(callback),
             }
             return self._sendReceiveMsg(("subscribe", requestArgs), timeout)
 
@@ -222,8 +241,7 @@ def _sendReceiveMsg(self, req, timeout):
     # Async function to handle the gRPC calls
     async def _grpcHandler(self, vss_client: kuksa_client.grpc.aio.VSSClient):
         self.run = True
-        subscriber_manager = kuksa_client.grpc.aio.SubscriberManager(
-            vss_client)
+        subscriber_manager = kuksa_client.grpc.aio.SubscriberManager(vss_client)
         self.grpc_connection_established = True
         while self.run:
             try:
@@ -273,7 +291,9 @@ def updateVSSTree(self, jsonStr, timeout=5):
     async def mainLoop(self):
         if self.insecure:
 
-            async with kuksa_client.grpc.aio.VSSClient(self.serverIP, self.serverPort, token=self.token) as vss_client:
+            async with kuksa_client.grpc.aio.VSSClient(
+                self.serverIP, self.serverPort, token=self.token
+            ) as vss_client:
                 logger.info("gRPC channel connected.")
                 await self._grpcHandler(vss_client)
         else:
@@ -282,7 +302,7 @@ async def mainLoop(self):
                 self.serverPort,
                 root_certificates=self.cacertificate,
                 tls_server_name=self.tls_server_name,
-                token=self.token
+                token=self.token,
             ) as vss_client:
                 logger.info("Secure gRPC channel connected.")
                 await self._grpcHandler(vss_client)