KVS functionality in the embedded web server (#161)

* Updated to savant-protobuf 0.2.2
* Implemented KVS handlers
* Implemented tests for the embedded webserver
* Implemented Python API/ABI samples for KVS
* updated docs

Cargo.toml

@@ -27,7 +27,7 @@ serde_json = "1.0"
 thiserror = "2"
 
 [workspace.package]
-version = "0.4.7"
+version = "0.4.8"
 edition = "2021"
 authors = ["Ivan Kudriavtsev <ivan.a.kudryavtsev@gmail.com>"]
 description = "Savant Rust core functions library"

docs/source/index.rst

@@ -21,3 +21,4 @@ Core Library
    modules/savant_rs/match_query
    modules/savant_rs/zmq
    modules/savant_rs/webserver
+   modules/savant_rs/webserver_kvs

docs/source/modules/savant_rs/webserver_kvs.rst

@@ -0,0 +1,6 @@
+savant_rs.webserver.kvs
+-----------------------------
+
+.. automodule:: savant_rs.webserver.kvs
+   :members:
+   :undoc-members:

python/webserver_kvs.py

@@ -0,0 +1,97 @@
+import savant_rs.webserver as ws
+from savant_rs.primitives import Attribute, AttributeValue
+import savant_rs.webserver.kvs as kvs
+
+import requests
+from time import sleep
+
+attr = Attribute(namespace="some", name="attr", hint="x", values=[
+    AttributeValue.bytes(dims=[8, 3, 8, 8], blob=bytes(3 * 8 * 8), confidence=None),
+    AttributeValue.bytes_from_list(dims=[4, 1], blob=[0, 1, 2, 3], confidence=None),
+    AttributeValue.integer(1, confidence=0.5),
+    AttributeValue.float(1.0, confidence=0.5),
+    AttributeValue.floats([1.0, 2.0, 3.0])
+])
+
+
+def abi():
+    global attr
+
+    kvs.set_attributes([attr], 1000)
+
+    attributes = kvs.search_attributes("*", "*")
+    assert len(attributes) == 1
+
+    attributes = kvs.search_attributes(None, "*")
+    assert len(attributes) == 1
+
+    attribute = kvs.get_attribute("some", "attr")
+    assert attribute.name == attr.name and attribute.namespace == attr.namespace
+
+    nonexistent_attribute = kvs.get_attribute("some", "other")
+    assert nonexistent_attribute is None
+
+    removed_attribute = kvs.del_attribute("some", "attr")
+
+    kvs.set_attributes([removed_attribute], 500)
+
+    sleep(0.55)
+
+    auto_removed_attribute = kvs.get_attribute("some", "attr")
+    assert auto_removed_attribute is None
+
+    kvs.del_attributes("*", "*")
+
+
+def api(base_url: str):
+    global attr
+    binary_attributes = kvs.serialize_attributes([attr])
+
+    response = requests.post(f'{base_url}/kvs/set', data=binary_attributes)
+    assert response.status_code == 200
+
+    response = requests.post(f'{base_url}/kvs/set-with-ttl/1000', data=binary_attributes)
+    assert response.status_code == 200
+
+    response = requests.post(f'{base_url}/kvs/delete/*/*')
+    assert response.status_code == 200
+
+    response = requests.post(f'{base_url}/kvs/set', data=binary_attributes)
+    assert response.status_code == 200
+
+    response = requests.post(f'{base_url}/kvs/delete-single/some/attr')
+    assert response.status_code == 200
+    removed_attributes = kvs.deserialize_attributes(response.content)
+    assert len(removed_attributes) == 1
+
+    response = requests.post(f'{base_url}/kvs/delete-single/some/attr')
+    assert response.status_code == 200
+    removed_attributes = kvs.deserialize_attributes(response.content)
+    assert len(removed_attributes) == 0
+
+    response = requests.post(f'{base_url}/kvs/set', data=binary_attributes)
+    assert response.status_code == 200
+
+    response = requests.get(f'{base_url}/kvs/search/*/*')
+    assert response.status_code == 200
+    attributes = kvs.deserialize_attributes(response.content)
+    assert len(attributes) == 1
+
+    response = requests.get(f'{base_url}/kvs/search-keys/*/*')
+    assert response.status_code == 200
+    attributes = response.json()
+    assert attributes == [["some", "attr"]]
+
+    response = requests.get(f'{base_url}/kvs/get/some/attr')
+    assert response.status_code == 200
+    attributes = kvs.deserialize_attributes(response.content)
+    assert len(attributes) == 1
+
+
+if __name__ == "__main__":
+    abi()
+    port = 8080
+    ws.init_webserver(port)
+    sleep(0.1)
+    api(f'http://localhost:{port}')
+    ws.stop_webserver()

savant_core/Cargo.toml

@@ -31,14 +31,14 @@ thiserror = { workspace = true }
 
 # unique to savant_core
 actix-web = "4"
-bytes = "1.9"
 crc32fast = "1"
 crossbeam = "0.8"
 derive_builder = "0.20"
 etcd_dynamic_state = { git = "https://github.com/insight-platform/etcd_dynamic_state", tag = "0.2.12" }
 etcd-client = { version = "0.13", features = ["tls"] }
 jmespath = { version = "0.3", features = ["sync"] }
 libloading = "0.8"
+moka = { version = "0.12", features = ["future"] }
 lru = { version = "0.12", features = ["hashbrown"] }
 nix = { version = "0.29", features = ["process", "signal"] }
 opentelemetry_sdk = { version = "0.24.1", features = ["rt-tokio"] }
@@ -47,11 +47,11 @@ reqwest = { version = "0.12.7", default-features = false, features = ["rustls-tl
 opentelemetry-stdout = { version = "0.5.0", features = ["trace"] }
 opentelemetry-semantic-conventions = "0.16.0"
 opentelemetry-jaeger-propagator = "0.3.0"
-prost = "0.12"
-prost-types = "0.12"
+prost = "0.13"
 rayon = "1.10"
 regex = "1"
-savant-protobuf = { git = "https://github.com/insight-platform/savant-protobuf", tag = "0.2.0" }
+savant-protobuf = { git = "https://github.com/insight-platform/savant-protobuf", tag = "0.2.2" }
+globset = "0.4"
 
 serde_yaml = "0.9"
 uuid = { version = "1.11", features = ["fast-rng", "v7"] }

savant_core/src/primitives.rs

@@ -8,6 +8,7 @@ pub mod bbox;
 pub use bbox::*;
 
 pub mod any_object;
+pub mod attribute_set;
 pub mod attribute_value;
 pub mod eos;
 pub mod frame;
@@ -22,6 +23,7 @@ pub use segment::*;
 
 pub mod rust {
     pub use super::attribute::Attribute;
+    pub use super::attribute_set::AttributeSet;
     pub use super::attribute_value::AttributeValue;
     pub use super::bbox::BBoxMetricType;
     pub use super::bbox::RBBox;

savant_core/src/primitives/attribute_set.rs

@@ -0,0 +1,57 @@
+use crate::json_api::ToSerdeJsonValue;
+use crate::primitives::{Attribute, WithAttributes};
+use crate::protobuf::from_pb;
+use savant_protobuf::generated;
+use serde_json::Value;
+
+#[derive(Debug, PartialEq, Clone, serde::Serialize, Default)]
+pub struct AttributeSet {
+    pub attributes: Vec<Attribute>,
+}
+
+impl ToSerdeJsonValue for AttributeSet {
+    fn to_serde_json_value(&self) -> Value {
+        serde_json::json!(self)
+    }
+}
+
+impl From<Vec<Attribute>> for AttributeSet {
+    fn from(attributes: Vec<Attribute>) -> Self {
+        Self { attributes }
+    }
+}
+
+impl AttributeSet {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    pub fn deserialize(bytes: &[u8]) -> anyhow::Result<Vec<Attribute>> {
+        let deser = from_pb::<generated::AttributeSet, AttributeSet>(bytes)?;
+        Ok(deser.attributes)
+    }
+
+    pub fn json(&self) -> String {
+        serde_json::to_string(&self.to_serde_json_value()).unwrap()
+    }
+
+    pub fn json_pretty(&self) -> String {
+        serde_json::to_string_pretty(&self.to_serde_json_value()).unwrap()
+    }
+}
+
+impl WithAttributes for AttributeSet {
+    fn with_attributes_ref<F, R>(&self, f: F) -> R
+    where
+        F: FnOnce(&Vec<Attribute>) -> R,
+    {
+        f(&self.attributes)
+    }
+
+    fn with_attributes_mut<F, R>(&mut self, f: F) -> R
+    where
+        F: FnOnce(&mut Vec<Attribute>) -> R,
+    {
+        f(&mut self.attributes)
+    }
+}

savant_core/src/protobuf/serialize.rs

@@ -1,12 +1,15 @@
+use crate::primitives::attribute_set::AttributeSet;
 use crate::primitives::frame::VideoFrameProxy;
 use crate::primitives::frame_batch::VideoFrameBatch;
 use crate::primitives::frame_update::VideoFrameUpdate;
 use crate::primitives::object::VideoObject;
 use crate::primitives::rust::UserData;
+use crate::primitives::Attribute;
 use savant_protobuf::generated;
 use std::convert::Infallible;
 
 mod attribute;
+mod attribute_set;
 mod bounding_box;
 mod intersection_kind;
 mod message_envelope;
@@ -30,6 +33,8 @@ pub enum Error {
     UuidParse(uuid::Error),
     #[error("An object has parent {0} which does not belong to the same frame")]
     InvalidVideoFrameParentObject(i64),
+    #[error("Failed to convert protobuf enum balue to Rust enum value: {0}")]
+    EnumConversionError(i32),
 }
 
 impl From<uuid::Error> for Error {
@@ -88,6 +93,8 @@ impl ToProtobuf<'_, generated::VideoFrameUpdate> for VideoFrameUpdate {}
 impl ToProtobuf<'_, generated::VideoFrameBatch> for VideoFrameBatch {}
 impl ToProtobuf<'_, generated::VideoObject> for VideoObject {}
 impl ToProtobuf<'_, generated::UserData> for UserData {}
+impl ToProtobuf<'_, generated::AttributeSet> for AttributeSet {}
+impl ToProtobuf<'_, generated::Attribute> for Attribute {}
 
 #[cfg(test)]
 mod tests {

savant_core/src/protobuf/serialize/attribute.rs

@@ -2,6 +2,7 @@ use crate::primitives::any_object::AnyObject;
 use crate::primitives::attribute_value::{AttributeValue, AttributeValueVariant};
 use crate::primitives::{Attribute, IntersectionKind, RBBox};
 use crate::protobuf::serialize;
+use prost::UnknownEnumValue;
 use savant_protobuf::generated;
 use std::sync::Arc;
 
@@ -175,7 +176,14 @@ impl TryFrom<&generated::attribute_value::Value> for AttributeValueVariant {
             }
             generated::attribute_value::Value::Intersection(i) => {
                 AttributeValueVariant::Intersection(crate::primitives::Intersection {
-                    kind: IntersectionKind::from(&i.data.as_ref().unwrap().kind.try_into()?),
+                    kind: IntersectionKind::from(
+                        &i.data
+                            .as_ref()
+                            .unwrap()
+                            .kind
+                            .try_into()
+                            .map_err(|e: UnknownEnumValue| Self::Error::EnumConversionError(e.0))?,
+                    ),
                     edges: i
                         .data
                         .as_ref()

savant_core/src/protobuf/serialize/attribute_set.rs

@@ -0,0 +1,110 @@
+use crate::primitives::attribute_set::AttributeSet;
+use crate::primitives::Attribute;
+use crate::protobuf::serialize;
+use savant_protobuf::generated;
+
+impl From<&AttributeSet> for generated::AttributeSet {
+    fn from(ud: &AttributeSet) -> Self {
+        let attributes = ud
+            .attributes
+            .iter()
+            .map(generated::Attribute::from)
+            .collect();
+
+        generated::AttributeSet { attributes }
+    }
+}
+
+impl TryFrom<&generated::AttributeSet> for AttributeSet {
+    type Error = serialize::Error;
+
+    fn try_from(value: &generated::AttributeSet) -> Result<Self, Self::Error> {
+        let attributes = value
+            .attributes
+            .iter()
+            .filter(|a| a.is_persistent)
+            .map(Attribute::try_from)
+            .collect::<Result<_, _>>()?;
+
+        Ok(AttributeSet { attributes })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::primitives::attribute_set::AttributeSet;
+    use crate::primitives::attribute_value::AttributeValue;
+    use crate::primitives::Attribute;
+    use savant_protobuf::generated;
+
+    #[test]
+    fn test_attribute_set() {
+        assert_eq!(
+            AttributeSet {
+                attributes: vec![
+                    (Attribute::new(
+                        "namespace",
+                        "name",
+                        vec![AttributeValue::string("value", Some(1.0))],
+                        &Some("hint"),
+                        true,
+                        true
+                    ))
+                ]
+            },
+            AttributeSet::try_from(&generated::AttributeSet {
+                attributes: vec![generated::Attribute {
+                    namespace: "namespace".to_string(),
+                    name: "name".to_string(),
+                    hint: Some("hint".to_string()),
+                    is_persistent: true,
+                    values: vec![generated::AttributeValue {
+                        confidence: Some(1.0),
+                        value: Some(generated::attribute_value::Value::String(
+                            generated::StringAttributeValueVariant {
+                                data: "value".to_string()
+                            }
+                        ))
+                    }],
+                    is_hidden: true,
+                }]
+                .into_iter()
+                .collect(),
+            })
+            .unwrap()
+        );
+        assert_eq!(
+            generated::AttributeSet {
+                attributes: vec![generated::Attribute {
+                    namespace: "namespace".to_string(),
+                    name: "name".to_string(),
+                    hint: Some("hint".to_string()),
+                    is_persistent: true,
+                    values: vec![generated::AttributeValue {
+                        confidence: Some(1.0),
+                        value: Some(generated::attribute_value::Value::String(
+                            generated::StringAttributeValueVariant {
+                                data: "value".to_string()
+                            }
+                        ))
+                    }],
+                    is_hidden: true,
+                }]
+                .into_iter()
+                .collect(),
+            },
+            generated::AttributeSet::from(&AttributeSet {
+                attributes: vec![
+                    (Attribute::new(
+                        "namespace",
+                        "name",
+                        vec![AttributeValue::string("value", Some(1.0))],
+                        &Some("hint"),
+                        true,
+                        true
+                    ))
+                ]
+            })
+        );
+    }
+}