v0.1.0

Author: Ogban Ugot

README.md

@@ -1,11 +1,94 @@
 # avenieca-python (WIP!)
-Python SDK for AveniECA
+Python SDK for publishing state signals to the AveniECA suite.
 
-## run
-```bash
-python consumer.py localhost:9092 testtopic
+## Usage
+
+### Stream continuously to a topic
+```python
+import os
+import numpy as np
+from avenieca.utils import Config
+from avenieca.utils import Signal
+from avenieca.producers import Stream
+
+# Define a handler that returns a signal dict like  
+# the sample in avenieca.utils.signal
+
+def handler():
+    Signal["valence"] = 10
+    Signal["state"] = np.array([0.2, 0.3, 0.8])
+    return Signal
+
+# Initialize Kafka configuration for the Stream
+Config["bootstrap_servers"] = os.environ["KAFKA_URL"]
+Config["topic"] = "left_wheel" #digital twin subscriber-topic
+
+#Initialize the Stream object with a sync_rate 
+# (the rate at which to publish signals).
+stream = Stream(config=Config, sync_rate=1)
+stream.publish(handler)
 ```
+
+### Publish one signal as an event
+```python
+import os
+import numpy as np
+from avenieca.utils import Config
+from avenieca.utils import Signal
+from avenieca.producers import Event
+
+# Initialize Kafka configuration for the Event
+Config["bootstrap_servers"] = os.environ["KAFKA_URL"]
+Config["topic"] = "left_wheel" #digital twin subscriber-topic
+
+# Define the signal
+Signal["valence"] = 9
+Signal["state"] = np.array([0.2, 0.3, 0.8])
+
+event = Event(config=Config)
+event.publish(Signal)
+```
+
+### Consume from kafka topic
+```python
+import os
+import numpy as np
+from avenieca.utils import Config
+from avenieca.utils.signal import get_state_as_list, get_state_as_array
+from avenieca.consumer import Consumer
+
+# Initialize Kafka configuration for the Event
+Config["bootstrap_servers"] = os.environ["KAFKA_URL"]
+Config["topic"] = "left_wheel" #digital twin subscriber-topic
+
+# Define a handler to process incoming messages
+def handler(data):
+    valence = data["valence"]
+    state = data["state"]
+    assert valence == 10
+    assert state == "[0.2, 0.3, 0.8]"
+
+client = Consumer(config=Config)
+client.consume(handler, True) # pass in handler
+
+# You can use util functions in your handler to 
+# convert the state signal from byte string to 
+# np.ndarray or python list
+def handler(data):
+    assert data["valence"] == 10
+    assert data["state"] == "[0.2, 0.3, 0.8]"
+    get_state_as_list(data)
+    assert data["state"] == [0.2, 0.3, 0.8]
+
+def handler(data):
+    assert data["valence"] == 10
+    assert data["state"] == "[0.2, 0.3, 0.8]"
+    get_state_as_array(data)
+    assert True, np.array_equal(data["state"], np.array([0.2, 0.3, 0.8]))
+```
+
+## Tests
 ```bash
-python producer.py localhost:9092 testtwin1
+python -m pytest test/
 ```
 

avenieca/__init__.py

@@ -0,0 +1,4 @@
+from avenieca.producers import *
+from avenieca.consumer import *
+from avenieca.producer import *
+from avenieca.utils import *

avenieca/consumer.py

@@ -0,0 +1,33 @@
+import json
+from kafka import KafkaConsumer
+
+
+class Consumer:
+    """
+    Base consumer for consuming messages (signals) from a digital twin.
+
+    :param config: configuration dictionary
+    """
+    def __init__(self,
+                 config: dict,
+                 ):
+        self.config = config
+        self.topic = config["topic"]
+        self.client = KafkaConsumer(
+            self.topic,
+            bootstrap_servers=config["bootstrap_servers"],
+            auto_offset_reset=config["auto_offset_reset"]
+        )
+
+    def consume(self, func, sync_once=False):
+        """
+        :param func: handler to process received messages
+        :param sync_once: run consume loop once
+        :return: none
+        """
+        for msg in self.client:
+            byte_val = msg.value
+            data = json.loads(byte_val)
+            func(data)
+            if sync_once:
+                break

avenieca/producer.py

@@ -0,0 +1,26 @@
+import json
+from kafka import KafkaProducer
+
+
+class Producer:
+    """
+    Base producer for publishing messages.
+
+    :param config: configuration dictionary
+    """
+    def __init__(self,
+                 config: dict,
+                 ):
+        self.config = config
+        self.topic = config["topic"]
+        self.client = KafkaProducer(bootstrap_servers=config["bootstrap_servers"])
+
+    def send(self, data: dict):
+        """
+        :param data: serialized signal dictionary
+        :return: FutureRecordMetadata
+        """
+        json_object = json.dumps(data).encode("utf-8")
+        result = self.client.send(self.topic, json_object)
+        return result
+

avenieca/producers/__init__.py

@@ -0,0 +1,2 @@
+from avenieca.producers.stream import *
+from avenieca.producers.event import *

avenieca/producers/event.py

@@ -0,0 +1,35 @@
+from avenieca.producer import Producer
+from avenieca.utils.signal import verify_signal
+
+
+class Event(Producer):
+    """
+    Event producer class for syncing from an event driven source.
+    Use this class if you want to handle the outer syncing logic, then pass the
+    signal to the publish method to publish.
+
+    :param config: configuration dictionary
+    """
+    def __init__(self, config: dict):
+        super().__init__(config)
+        self.config = config
+        self.sync = True
+
+    def publish(self, signal: dict):
+        """
+        call this method with the signal dictionary to publish once to a digital twin
+
+        :param signal: signal data
+        :return: None
+        """
+        if self.sync:
+            verify_signal(signal)
+            return self.send(signal)
+
+    @property
+    def config(self):
+        return self.config
+
+    @config.setter
+    def config(self, value):
+        self._config = value

avenieca/producers/stream.py

@@ -0,0 +1,44 @@
+import time
+from avenieca.producer import Producer
+from avenieca.utils.signal import verify_signal
+
+
+class Stream(Producer):
+    """
+    Stream producer class for continuous syncing at a sync_rate. Use this class if you want the
+    library to handle the syncing logic for you. Provide a handler that returns the signal
+    data for publishing.
+
+    :param config: config dictionary
+    :param sync_rate: int (seconds) or float (sub-seconds)
+    """
+
+    def __init__(self, config: dict, sync_rate: [int, float]):
+        super().__init__(config)
+        self.config = config
+        self.sync_rate = sync_rate
+        self.sync = True
+
+    def publish(self, func, sync_once=False):
+        """
+        Basic publish stream timed by the sync_rate
+
+        :param func: handler to return the signal (dict data) for publishing
+        :param sync_once: run the sync loop once
+        :return: none
+        """
+        while self.sync:
+            signal = func()
+            verify_signal(signal)
+            self.send(signal)
+            if sync_once:
+                break
+            time.sleep(self.sync_rate)
+
+    @property
+    def config(self):
+        return self.config
+
+    @config.setter
+    def config(self, value):
+        self._config = value

avenieca/utils/__init__.py

@@ -0,0 +1,2 @@
+from avenieca.utils.config import *
+from avenieca.utils.signal import *

avenieca/utils/config.py

@@ -0,0 +1,5 @@
+Config = {
+    "bootstrap_servers": str,
+    "topic": str,
+    "auto_offset_reset": "earliest"
+}

avenieca/utils/signal.py

@@ -0,0 +1,73 @@
+import json
+
+import numpy as np
+
+Signal = {
+    "valence": None,
+    "score": None,
+    "state": None,
+}
+
+DEFAULT_STATE_DIM = 1
+
+
+def get_state_as_array(signal, dtype=np.float64):
+    state = signal["state"]
+    if type(state) == str:
+        state = json.loads(state)
+    arr = np.array(state, dtype=dtype)
+    signal["state"] = arr
+
+
+def get_state_as_list(signal, dtype=np.float64):
+    state = signal["state"]
+    if type(state) == str:
+        state = json.loads(state)
+    arr = np.array(state, dtype=dtype)
+    signal["state"] = arr.tolist()
+
+
+def verify_signal(signal):
+    assert type(signal) == dict
+    assert len(signal) == 3
+
+    if signal["state"] is None:
+        raise Exception("signal state cannot be None")
+
+    if type(signal["state"]) == str:
+        verify_str_shape(signal["state"])
+
+    if type(signal["state"]) == list:
+        arr_list = signal["state"]
+        verify_list_shape(arr_list)
+        if all(isinstance(item, (int, float)) for item in arr_list):
+            signal["state"] = json.dumps(arr_list)
+            return
+        else:
+            raise Exception("signal state values must be int or float")
+
+    if type(signal["state"]) == np.ndarray:
+        try:
+            arr_list = signal["state"].tolist()
+            verify_list_shape(arr_list)
+            signal["state"] = json.dumps(arr_list)
+            return
+        except Exception as e:
+            raise Exception("error converting state signal from numpy array to byte string: {}".format(e))
+
+
+def verify_str_shape(state):
+    if state == "":
+        raise Exception("signal state cannot be empty")
+    state = json.loads(state)
+    verify_list_shape(state)
+
+
+def verify_list_shape(state, dtype=np.float64):
+    arr = np.array(state, dtype=dtype)
+    verify_np_shape(arr)
+
+
+def verify_np_shape(state: np.ndarray):
+    if state.ndim > DEFAULT_STATE_DIM:
+        raise Exception("state signal should be of dimension {} got {}".format(DEFAULT_STATE_DIM, state.ndim))