v0.2.0

Author: kingature

README.md

@@ -67,11 +67,11 @@ payload:      # List of input elements used for data annotations (optional).
 Sensors are bundled in groups. Within a group **sensor names must be unique**.
 An example config file is provided as [`group.yaml`](examples/config/group.yaml). There, you also find example configs of all supported sensors.
 
-The payload can be used for data annotations. You can find an overview of supported widgets at [`payload.md`](examples/payload.md).
+The payload can be used for data annotations. You can find an overview of supported widgets at [`payload.md`](docs/payload.md).
 
 ### 2. Using Code
 
-For programmatic control, please refer to the [`demo.py`](examples/demo.py) example.
+For programmatic control, please refer to the [`demo.py`](examples/record_data.py) example.
 - Run with the default config ([`digit.yaml`](examples/config/digit.yaml))
     ```bash
     python examples/simple/demo.py

examples/payload.md docs/payload.mdexamples/payload.md renamed to docs/payload.md

@@ -1,41 +1,21 @@
-"""
-This script demonstrates how to access and read saved sensor data from recordings.
-
-## Data Structure:
-- During recording, data is stored in smaller chunks as `.pkl` files.
-- Each `.pkl` file contains a dictionary mapping `sensor_name` to its corresponding sensor data.
-- Sensor data is a dictionary where each `data_stream` (e.g., 'camera' for Digit sensor) maps to a list of data points.
-- Each data point is a dictionary with:
-  - `delta`: Time elapsed since recording started.
-  - `data`: The actual recorded data point.
-"""
-
-import os
-import pickle
-from typing import Generator, Dict, Any
-
-
-def read_dataset(directory: str) -> Generator[Dict[str, Any], None, None]:
-    """
-    Reads and yields sensor data from all `.pkl` files in the specified directory.
-
-    Args:
-        directory (str): Path to the dataset folder.
-
-    Yields:
-        dict: Parsed sensor data from each `.pkl` file.
-    """
-    if not os.path.exists(directory):
-        raise FileNotFoundError(f"Directory not found: {directory}")
-
-    for filename in sorted(os.listdir(directory)):
-        if filename.endswith('.pkl'):
-            file_path = os.path.join(directory, filename)
-            with open(file_path, 'rb') as file:
-                yield pickle.load(file)
-
-
-if __name__ == "__main__":
-    dataset_path = "../datasets/my_dataset"
-    for data_chunk in read_dataset(dataset_path):
-        print(data_chunk)
+from opentouch_interface.decoder import Decoder
+
+# 1. Load the data set
+dataset = Decoder('<full-path-to-touch-file>')
+
+# 2. Inspect the data set
+print(f'The following sensors have been captured: {dataset.sensor_names}')
+print('The sensors have the following streams:')
+for sensor in dataset.sensor_names:
+    print(f'\t- {sensor}: {dataset.stream_names_of(sensor)}')
+
+# 3. Isolate the camera data
+sensor_name = dataset.sensor_names[0]
+data_stream = 'camera'
+camera_data = dataset.stream_data_of(sensor_name, data_stream)
+
+camera_data = dataset.stream_data_of('Finger', 'audio')
+for event in camera_data:
+    delta = event['delta']
+    data = event['data']
+    print(delta, type(data))

examples/access_data.py

@@ -20,6 +20,13 @@ def run(cfg: DictConfig):
     config = OmegaConf.to_container(cfg, resolve=True)
     sensor_group = SensorGroup(config=config)
 
+    # Add metadata (e.g., data of recording) to payload
+    sensor_group.payload.add({
+        'type': 'text_input',
+        'label': 'time',
+        'default': str(time.time())
+    })
+
     # Retrieve and configure a specific sensor by name
     digit = sensor_group.get_sensor('First Gripper')
     digit.set('rgb', [0, 0, 15])

examples/datasets.py

@@ -0,0 +1,84 @@
+from abc import abstractmethod
+from typing import Dict, Any
+
+from torch import Tensor
+
+from opentouch.core.base_nn import BaseNeuralNetwork
+
+
+class BaseCNN(BaseNeuralNetwork):
+    """
+    Base class for building Convolutional Neural Networks (CNNs) that supports both
+    PyTorch and ONNX inference. Handles model construction, forward pass, prediction,
+    and saving models in ONNX format with metadata.
+
+    Attributes:
+        input_channels (int): Number of input channels (e.g., 3 for RGB images).
+        num_classes (int): Number of output classes for classification.
+        label_mapping (dict, optional): Mapping from class indices to labels.
+    """
+
+    def __init__(self, label_mapping: dict, input_channels: int = 3) -> None:
+        """Initialize the BaseCNN model."""
+        super().__init__()
+        self.input_channels: int = input_channels
+        self.label_mapping: dict = label_mapping
+        self.num_classes: int = len(label_mapping)
+        self.build()  # Call the subclass's build method to define the architecture
+        self.to(self.device)  # Move model to device (CPU or GPU)
+
+    @property
+    def metadata(self) -> Dict[str, Any]:
+        """
+        Constructs metadata with the basic attributes required for all models, including input channels,
+        number of classes, label mapping and output type.
+        """
+        base_metadata = super().metadata
+        base_metadata.update({
+            'input_channels': self.input_channels,
+            'num_classes': self.num_classes,
+            'label_mapping': self.label_mapping,
+            'output_type': 'classification'
+        })
+        return base_metadata
+
+    @abstractmethod
+    def build(self) -> None:
+        """To be implemented in derived classes to define the model architecture."""
+        pass
+
+    @abstractmethod
+    def preprocess(self, x: Tensor) -> Tensor:
+        """
+        Preprocesses the input tensor before it's fed into the model. This method can be overridden
+        in subclasses for custom preprocessing.
+
+        Args:
+            x (Tensor): Input tensor.
+
+        Returns:
+            Tensor: Preprocessed input tensor.
+        """
+        return x
+
+    def forward(self, x: Tensor) -> Tensor:
+        """
+        Forward pass of the model. Ensures the input tensor is processed and passed through
+        the defined model. Raises an error if the model isn't built.
+
+        Args:
+            x (Tensor): Input tensor.
+
+        Returns:
+            Tensor: The output of the model after the forward pass.
+        """
+        if not self.model:
+            raise NotImplementedError("The model is not defined. Implement build() in your subclass.")
+
+        # Move input to the correct device
+        x = x.to(self.device)
+
+        # Preprocess input
+        x = self.preprocess(x)
+
+        return self.model(x)  # Forward pass through the model

examples/demo.py examples/record_data.pyexamples/demo.py renamed to examples/record_data.py

@@ -0,0 +1,57 @@
+from typing import Dict, Any
+
+import torch
+from abc import abstractmethod
+
+from opentouch.core.base_model import BaseModel
+
+
+class BaseFilter(BaseModel):
+    """
+    A base abstract class for all filters, providing utility methods for metadata handling,
+    filter saving in ONNX format, and enforcing standard PyTorch module behavior for filtering.
+    """
+
+    def __init__(self):
+        super().__init__()
+
+    @property
+    def metadata(self) -> Dict[str, Any]:
+        """
+        Constructs metadata with the basic attributes required for all models, including input channels,
+        number of classes, and output type.
+        """
+        base_metadata = super().metadata
+        base_metadata.update({
+            'output_type': 'image'
+        })
+        return base_metadata
+
+    @property
+    @abstractmethod
+    def description(self) -> str:
+        """
+        Abstract property for a description of the filter's purpose.
+        """
+        pass
+
+    @abstractmethod
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        """
+        Abstract method that must be implemented to define the forward pass for the filter.
+        Args:
+            x (torch.Tensor): Input tensor to be filtered.
+        Returns:
+            torch.Tensor: Output tensor after passing through the filter.
+        """
+        pass
+
+    @abstractmethod
+    def onnx_export(self) -> Dict[str, Any]:
+        """
+        Constructs the parameters needed for torch.onnx.export().
+
+        Returns:
+            dict: A dictionary containing the parameters needed for ONNX export.
+        """
+        pass

…h_interface/dashboard/models/__init__.py opentouch/__init__.pyopentouch_interface/dashboard/models/__init__.py renamed to opentouch/__init__.py opentouch_interface/dashboard/models/__init__.py renamed to opentouch/__init__.py

@@ -0,0 +1,98 @@
+import json
+import zipfile
+from abc import ABC, abstractmethod
+from pathlib import Path
+from typing import Dict, Any
+
+import torch
+import torch.nn as nn
+
+
+class BaseModel(nn.Module, ABC):
+    """
+    A base abstract class for all models, providing utility methods for metadata handling,
+    model saving in ONNX format, and enforcing standard PyTorch model behavior.
+    """
+
+    def __init__(self):
+        super().__init__()
+
+    @property
+    @abstractmethod
+    def description(self) -> str:
+        """Abstract property for a description of the model's purpose."""
+        pass
+
+    @property
+    def metadata(self) -> Dict[str, Any]:
+        """
+        Constructs metadata with the basic attributes required for all models.
+
+        Returns:
+            dict: Metadata containing model_type and description.
+        """
+        return {
+            'model_type': self.__class__.__name__,
+            'description': self.description
+        }
+
+    @abstractmethod
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        """
+        Abstract method that must be implemented to define the forward pass for the model.
+        Args:
+            x (torch.Tensor): Input tensor.
+        Returns:
+            torch.Tensor: Output tensor after passing through the model.
+        """
+        pass
+
+    @abstractmethod
+    def onnx_export(self) -> Dict[str, Any]:
+        """
+        Constructs the parameters needed for torch.onnx.export().
+
+        Returns:
+            dict: A dictionary containing the parameters needed for ONNX export.
+        """
+        pass
+
+    def save(self, path: str) -> None:
+        """
+        Saves the model in ONNX format and stores it in a zip file along with metadata.
+
+        Args:
+            path (str): The path to save the model.
+        """
+        # Gather ONNX export parameters
+        onnx_params = self.onnx_export()
+        onnx_path = Path(path).with_suffix(".onnx")
+        metadata_path = Path(path + "_metadata.json")
+
+        # Perform the ONNX export using the current model (`self`)
+        torch.onnx.export(
+            self,
+            onnx_params['example_input'],
+            str(onnx_path),  # This is the required third argument - file path
+            export_params=onnx_params.get('export_params', True),
+            opset_version=onnx_params.get('opset_version', 17),
+            input_names=onnx_params.get('input_names', ['input']),
+            output_names=onnx_params.get('output_names', ['output']),
+            # dynamic_axes=onnx_params.get('dynamic_axes', {'input': {0: 'batch_size'}}),
+        )
+
+        # onnx_program.save(str(onnx_path))  # Saves a file called onnx_path.onnx
+
+        # Save metadata to a JSON file
+        metadata = self.metadata
+        metadata_path.write_text(json.dumps(metadata))
+
+        # Zip ONNX model and metadata file
+        zip_path = Path(path).with_suffix(".zip")
+        with zipfile.ZipFile(zip_path, 'w') as model_zip:
+            model_zip.write(onnx_path, onnx_path.name)
+            model_zip.write(metadata_path, metadata_path.name)
+
+        # Clean up temporary files
+        onnx_path.unlink()
+        metadata_path.unlink()