feat(backend-python): add basic telestion library for python

backend-features/.gitignore

@@ -1,3 +1,4 @@
 # python venv
 .venv
 
+__pycache__

backend-features/dev-mode.feature

@@ -6,7 +6,7 @@ Feature: Development mode
     Given I have no service configuration
     When I start the service with "--dev" without NATS
     Then the service should start
-    And the service should be configured with "NATS_URL" set to "localhost:4222"
+    And the service should be configured with "NATS_URL" set to either "localhost:4222" or "nats://localhost:4222"
     And the service should be configured with "NATS_USER" set to "undefined"
     And the service should be configured with "NATS_PASSWORD" set to "undefined"
 
@@ -18,4 +18,4 @@ Feature: Development mode
     Then the service should start
     And the service should connect to NATS
     And the service should be configured with "DATA_DIR" set to "/tmp"
-    And the service should be configured with "NATS_URL" set to "nats:4255"
+    And the service should be configured with "NATS_URL" set to either "nats:4255" or "nats://nats:4255"

backend-features/steps/service.py

@@ -57,6 +57,14 @@ def assert_service_configured_with(context, key, expected_value):
             expected_value), f"Expected {key} to be {expected_value}, got {context.result['config'][key]}"
 
 
+@then(u'the service should be configured with "{key}" set to either "{expected_value}" or "{alternative}"')
+def assert_service_configured_with_alternative(context, key, expected_value, alternative):
+    try:
+        assert_service_configured_with(context, key, alternative)
+    except AssertionError:
+        assert_service_configured_with(context, key, expected_value)
+
+
 @then(u'the NATS connection API should be available to the service')
 def assert_nats_api_available(context):
     assert 'result' in context, 'Service must be started before checking connection'

backend-python/.gitignore

@@ -0,0 +1,3 @@
+.venv/
+
+__pycache__

backend-python/Dockerfile

@@ -0,0 +1,12 @@
+# Use the official python image as the base image
+FROM python:3.12-alpine
+
+# Set the working directory inside the container
+WORKDIR /app
+
+# Copy the requirements and the rest
+COPY . .
+RUN pip install -r requirements.txt
+
+# Set the entrypoint to the built binary
+ENTRYPOINT ["python3", "./testbed.py"]

backend-python/LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2024 WüSpace e. V.
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

backend-python/README.md

@@ -0,0 +1,8 @@
+# Telestion Python Backend
+
+This repository contains everything to build Telestion backend services in Python.
+Additional examples help in setting everything up.
+
+The Python backend is mainly added to allow for easier interfacing with many scientific libraries, such as numpy,
+scipy, tensorflow or pytorch.
+Creating static graphs with Matplotlib that can be rendered in the frontend could also be created.

backend-python/requirements.txt

@@ -0,0 +1,2 @@
+nats-py>=2.9.0,<3
+pydantic>=2.10.4,<3

backend-python/src/telestion/__init__.py

@@ -0,0 +1,5 @@
+from . import backend
+
+__all__ = [
+    'backend',
+]

backend-python/src/telestion/backend/__init__.py

@@ -0,0 +1,7 @@
+from . import lib
+from . import config
+
+__all__ = [
+    'config',
+    'lib'
+]

backend-python/src/telestion/backend/config.py

@@ -0,0 +1,154 @@
+import argparse
+import json
+import os
+from pathlib import Path
+from typing import Any, TypeVar
+
+from pydantic import BaseModel, Field, ConfigDict
+
+
+class TelestionConfig(BaseModel):
+    dev: bool = False
+    nats_url: str = Field(alias="NATS_URL")
+    nats_user: str | None = Field(alias="NATS_USER", default=None)
+    nats_password: str | None = Field(alias="NATS_PASSWORD", default=None)
+    config_file: Path | None = Field(alias="CONFIG_FILE", default=None)
+    config_key: str | None = Field(alias="CONFIG_KEY", default=None)
+    service_name: str = Field(alias="SERVICE_NAME")
+    data_dir: Path = Field(alias="DATA_DIR")
+
+    # To include all envs and config parts -> it is recommended to add a custom subtype
+    model_config = ConfigDict(
+        extra='allow'
+    )
+
+
+# With this we allow users to extend TelestionConfig for finer control over custom config fields
+_TelestionConfigT = TypeVar("_TelestionConfigT", bound=TelestionConfig)
+
+
+def build_config(clazz: type[_TelestionConfigT] = None, **kwargs) -> _TelestionConfigT:
+    if clazz is None:
+        clazz = TelestionConfig
+
+    cli_args = _parse_cli()
+
+    def _from_env_or_cli(key: str):
+        return cli_args.get(key, os.environ.get(key, None))
+
+    config_path = _from_env_or_cli('CONFIG_FILE')
+    config_key = _from_env_or_cli('CONFIG_KEY')
+
+    config_assembly: dict[str, Any] = dict()
+    if 'dev' in cli_args and cli_args['dev']:
+        # 1. Add default config
+        config_assembly.update(defaults())
+
+    if config_path is not None:
+        config_path = Path(config_path)
+        # 2. Insert config file
+        config_assembly.update(_parse_config_file(config_path, config_key))
+
+    # 3. Add Environment Variables
+    config_assembly.update(os.environ)
+
+    # 4. Add CLI args
+    config_assembly.update(cli_args)
+
+    # 5. Add manual overwrites
+    config_assembly.update(kwargs)
+
+    return clazz(**config_assembly)
+
+
+def defaults() -> dict[str, Any]:
+    return {
+        'NATS_URL': "nats://localhost:4222",
+        'SERVICE_NAME': f"dev-{os.getpid()}",
+        'DATA_DIR': Path("./data")
+    }
+
+
+def _parse_cli() -> dict[str, Any]:
+    description = "CLI Interface for the Telestion Services. This is one way to setup your Telestion application."
+    epilog = "For more information please visit https://github.com/wuespace/telestion or \
+    https://telestion.wuespace.de/"
+    parser = argparse.ArgumentParser(
+        description=description,
+        epilog=epilog,
+        prog="Telestion-CLI (Python)",
+        argument_default=argparse.SUPPRESS,
+        add_help=True,
+        exit_on_error=False
+    )
+
+    parser.add_argument("--dev", action='store_true', help="If set, program will start in development mode")
+    parser.add_argument("--version", action='version', version="%(prog)s v1.0-alpha")
+
+    parser.add_argument("--NATS_URL", help="NATS url of the server the service can connect to")
+    parser.add_argument("--NATS_USER", help="NATS user name for the authentication with the server")
+    parser.add_argument("--NATS_PASSWORD", help="NATS password for the authentication with the server \
+    (Note: It is recommended to set this via the environment variables or the config!)")
+
+    parser.add_argument("--CONFIG_FILE", help="file path to the config of the service", type=Path)
+    parser.add_argument("--CONFIG_KEY", help="object key of a config file")
+
+    parser.add_argument("--SERVICE_NAME", help="name of the service also used in the nats service registration")
+    parser.add_argument("--DATA_DIR", help="path where the service can store persistent data", type=Path)
+
+    namespace, unknown_args = parser.parse_known_args()
+    parsed_args = vars(namespace)
+
+    # parse also unknown args
+    return _parse_unknown_args(unknown_args, parsed_args)
+
+
+def _parse_config_file(config_p: Path, key: str = None) -> dict[str, Any]:
+    with open(config_p, 'r') as config_f:
+        content = json.load(config_f)
+
+    if key is None:
+        return content
+
+    return content[key]
+
+
+def _parse_unknown_args(unknown_args: list[str], parsed_args: dict[str, Any]) -> dict[str, Any]:
+    # cases to handle:
+    # 1. key == ""
+    # 2. key == "abc", vals == None
+    # 3. key == "abc", vals == 'abc'
+    # 4. key == "abc", vals == ['abc', 'foo', 'bar']
+
+    key: str = ""
+    for unknown_arg in unknown_args:
+        if unknown_arg.startswith("-"):
+            if '=' in unknown_arg:
+                # handle case of equal signs in args
+                split = unknown_arg.split('=')
+                _parse_unknown_args([split[0], *split[1:]], parsed_args)
+                key = ""
+                continue
+
+            # handles 2.
+            if key not in parsed_args and key != "":
+                parsed_args[key] = True
+
+            key = unknown_arg[(2 if unknown_arg.startswith("--") else 1):]
+            continue
+
+        if key == "":   # handles 1.
+            raise ValueError("Parsing unknown arguments failed! Did you forget to specify a flag for the argument?")
+
+        if key in parsed_args:
+            if isinstance(parsed_args[key], list):
+                # handles 4.
+                parsed_args[key].append(unknown_arg)
+            else:
+                # handles 3.
+                parsed_args[key] = [parsed_args[key], unknown_arg]
+        else:
+            # handles 3. -> 4.
+            parsed_args[key] = unknown_arg
+
+    return parsed_args

backend-python/src/telestion/backend/lib.py

@@ -0,0 +1,119 @@
+import json
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+import nats
+from nats.aio.client import Client as NatsClient, Msg as NatsMsg, DEFAULT_FLUSH_TIMEOUT  # mostly for type hinting
+from nats.aio.subscription import Subscription
+
+from telestion.backend.config import build_config, _TelestionConfigT
+
+
+@dataclass
+class Service:
+    """Helper Class for starting NATS clients also exposing the parsed config."""
+    nc: NatsClient | None   # None if Options.nats = False
+    """Configured and started NATS client for this service."""
+    data_dir: Path
+    """Directory where all data (temporary and persistent) should be stored."""
+    service_name: str
+    """Name of this service. Note that it is not necessarily unique!"""
+    config: _TelestionConfigT
+    """TelestionConfig instance for this service """
+
+    # wrapper methods for NatsClient instance for convenience
+    async def publish(self, *args, **kwargs) -> None:
+        """Wrapper for https://nats-io.github.io/nats.py/modules.html#nats.aio.client.Client.publish"""
+        await self.nc.publish(*args, **kwargs)
+
+    async def subscribe(self, *args, **kwargs) -> Subscription:
+        """Wrapper for https://nats-io.github.io/nats.py/modules.html#nats.aio.client.Client.subscribe"""
+        return await self.nc.subscribe(*args, **kwargs)
+
+    async def request(self, *args, **kwargs) -> NatsMsg:
+        """Wrapper for Client.request(subject, payload, timeout, old_style, headers)"""
+        return await self.nc.request(*args, **kwargs)
+
+    async def flush(self, timeout: int = DEFAULT_FLUSH_TIMEOUT) -> None:
+        """Wrapper for https://nats-io.github.io/nats.py/modules.html#nats.aio.client.Client.flush"""
+        await self.nc.flush(timeout)
+
+    async def drain(self) -> None:
+        """Wrapper for https://nats-io.github.io/nats.py/modules.html#nats.aio.client.Client.drain"""
+        await self.nc.drain()
+
+    async def close(self) -> None:
+        """Wrapper for https://nats-io.github.io/nats.py/modules.html#nats.aio.client.Client.close"""
+        await self.nc.close()
+
+
+async def start_service(
+        nats_disabled: bool = False,
+        config: _TelestionConfigT = None,
+        nats_connecting_options: dict[str, Any] | None = None
+) -> Service:
+    """Creates a Service with the parsed config and spins up a new NATS client if configured to do so."""
+    if config is None:
+        config = build_config(_nats_connecting_options=nats_connecting_options)
+
+    if nats_connecting_options is None:
+        nats_connecting_options = {}
+
+    nc = None if nats_disabled else await nats.connect(servers=_prepare_nats_url(config), **nats_connecting_options)
+    return Service(nc, config.data_dir, config.service_name, config)
+
+
+# Macros
+def json_encode(msg: Any, encoding='utf-8', errors='strict', **dumps_kwargs) -> bytes:
+    """
+    Helper function to encode messages to json.
+    This convenience macro helps to reduce encoding/decoding boilerplate.
+    For finer control implement this function by your own and customize to your needs.
+
+    :param msg:             message to encode
+    :param encoding:        encoding to use (default: utf-8)
+    :param errors:          way to handle encoding errors, see #bytes.encode()
+    :param dumps_kwargs:    additional arguments to pass to json.dumps
+    :return: encoded json message as utf-8 bytes
+    """
+    return json.dumps(msg, **dumps_kwargs).encode(encoding=encoding, errors=errors)
+
+
+def json_decode(msg: str | bytes | bytearray, encoding='utf-8', errors='strict', **loads_kwargs) -> Any:
+    """
+    Helper function to decode messages from json into an object.
+    This convenience macro helps to reduce encoding/decoding boilerplate.
+    For finer control implement this function by your own and customize to your needs.
+
+    :param msg:             message to decode
+    :param encoding:        encoding used to encode the bytes
+    :param errors:          way to handle encoding errors, see #bytes.decode()
+    :param loads_kwargs:    additional arguments for json.loads()
+    :return: if successful, returns the parsed json message
+    """
+    if not isinstance(msg, str):
+        # ensure to support any encoding supported by python
+        msg = msg.decode(encoding=encoding, errors=errors)
+
+    return json.loads(msg, **loads_kwargs)
+
+
+def _prepare_nats_url(config: _TelestionConfigT) -> str:
+    """
+    Helper function that creates the valid url for the NATS client.
+    Because the Python interface does not support user authentication out of the box with a separate design this is done
+    via the connecting url.
+
+    :param config: parsed config from all sources
+    :return: created url from parsed config url, user and password
+    """
+    url = config.nats_url
+
+    if config.nats_user is None or config.nats_password is None:
+        return url
+
+    if '://' in url:
+        _, url = url.split('://', 1)
+
+    return f"nats://{config.nats_user}:{config.nats_password}@{url}"