Pass config from topic

simulink-wheel/aux-files/simulink_wrapper.m

@@ -1,16 +1,39 @@
-function outputMatrix = simulink_wrapper(inputMatrix)
+function outputMatrix = simulink_wrapper(inputMatrix, configJson)
     mdl = "simulink_model_example";
 
     %---------------------------------------------------------------
-    % 1. Extract time and input signals from input matrix
+    % 1. Parse JSON configuration (if provided)
+    %    Config values are applied as model workspace variables
+    %---------------------------------------------------------------
+    if nargin < 2 || isempty(configJson)
+        config = struct();
+    else
+        try
+            config = jsondecode(configJson);
+        catch ME
+            fprintf("Warning: Failed to parse config JSON: %s. Using empty config.\n", ME.message);
+            config = struct();
+        end
+    end
+
+    % Print first config received (for debugging)
+    persistent config_printed
+    if isempty(config_printed)
+        config_printed = true;
+        fprintf("First config received:\n");
+        disp(config);
+    end
+
+    %---------------------------------------------------------------
+    % 2. Extract time and input signals from input matrix
     %    Assumes inputMatrix = [time, signal1, signal2, ..., signalN]
     %---------------------------------------------------------------
     t = inputMatrix(:,1);         % Time vector
     u = inputMatrix(:,2:end);     % Signal values
     n_signals = size(u,2);        % Number of input signals
 
     %---------------------------------------------------------------
-    % 2. Build Dataset object
+    % 3. Build Dataset object
     %    Timeseries order must match Inport order in the Simulink model
     %---------------------------------------------------------------
     inports = Simulink.SimulationData.Dataset;
@@ -20,30 +43,47 @@
     end
 
     %------------------------------------------------------------------
-    % 3. Configure SimulationInput object (efficient for repeated calls)
+    % 4. Configure SimulationInput object (efficient for repeated calls)
     %------------------------------------------------------------------
     % Use a persistent SimulationInput object to avoid recompilation
     persistent s0_loaded
     if isempty(s0_loaded)
         fprintf("Compiling model...\n")
         s0 = Simulink.SimulationInput(mdl);
-        
+
         % Configure for deployment (Rapid Accelerator + safe options)
         % This prepares the model for repeated high-performance execution
         s0 = simulink.compiler.configureForDeployment(s0);
-        
+
         s0_loaded = s0;
         fprintf("Compiled\n")
     end
 
     % Clone and update external inputs for this specific run
     s = s0_loaded.setExternalInput(inports);
-    
+
     % Set simulation stop time based on last time sample
     s = s.setModelParameter("StopTime", num2str(t(end)));
 
     %------------------------------------------------------------------
-    % 4. Run the simulation
+    % 5. Apply JSON config as model variables
+    %    Each field in the config JSON becomes a workspace variable
+    %------------------------------------------------------------------
+    if isstruct(config)
+        configFields = fieldnames(config);
+        for i = 1:numel(configFields)
+            paramName = configFields{i};
+            paramValue = config.(paramName);
+            try
+                s = s.setVariable(paramName, paramValue);
+            catch ME
+                fprintf("Warning: Failed to set variable '%s': %s\n", paramName, ME.message);
+            end
+        end
+    end
+
+    %------------------------------------------------------------------
+    % 6. Run the simulation
     %------------------------------------------------------------------
     out = sim(s);
     % Only print sim output once
@@ -58,7 +98,7 @@
     yout = out.yout; 
 
     %---------------------------------------------------------------
-    % 5. Extract full output signal values into a matrix [n_times x total_output_width]
+    % 7. Extract full output signal values into a matrix [n_times x total_output_width]
     %---------------------------------------------------------------
     if isa(yout, 'Simulink.SimulationData.Dataset')
         n_outputs = yout.numElements;

simulink-wheel/main.py

@@ -1,24 +1,68 @@
 from quixstreams import Application
 import numpy as np
 import os
+import json
 import quixmatlab
 
 # Initiate quixmatlab
 quixmatlab_client = quixmatlab.initialize()
 print("Exported MATLAB functions:", dir(quixmatlab_client))
 
+# Default configuration (used if config is missing or invalid)
+DEFAULT_CONFIG = {}
+
+
+def prepare_config_json(config_data) -> str:
+    """
+    Prepare config data as a JSON string to pass to MATLAB.
+
+    Handles:
+        - Missing field (None) -> empty JSON object
+        - Already a dict -> serialize to JSON string
+        - Already a string -> validate and pass through
+        - Invalid data -> empty JSON object
+
+    Returns:
+        str: Valid JSON string to pass to MATLAB wrapper
+    """
+    if config_data is None:
+        return json.dumps(DEFAULT_CONFIG)
+
+    # If it's already a dict, serialize to JSON string
+    if isinstance(config_data, dict):
+        return json.dumps(config_data)
+
+    # If it's a string, validate it's valid JSON
+    if isinstance(config_data, str):
+        if not config_data.strip():
+            return json.dumps(DEFAULT_CONFIG)
+        try:
+            # Validate JSON by parsing, then return original string
+            json.loads(config_data)
+            return config_data
+        except json.JSONDecodeError as e:
+            print(f"Warning: Invalid JSON in config field: {e}. Using empty config.")
+            return json.dumps(DEFAULT_CONFIG)
+
+    print(f"Warning: Unexpected config type: {type(config_data)}. Using empty config.")
+    return json.dumps(DEFAULT_CONFIG)
+
+
 # Define matlab function call
 def matlab_processing(row: dict):
-    # Prepare function inputs
+    # Prepare function inputs from Kafka data
     x = row["x"]
     y = row["y"]
-    theta = np.pi/4 # 45 degrees in radians
 
-    # Call function here
-    input_matrix = np.array([[0, x, y, theta]])
-    # print("Input", input_matrix)
-    output_matrix = quixmatlab_client.simulink_wrapper(input_matrix)
-    # print("Output", output_matrix)
+    # Prepare config JSON to pass directly to MATLAB
+    config_json = prepare_config_json(row.get("config"))
+
+    # Build input matrix: [time, x, y]
+    # Config parameters are passed separately via JSON to the MATLAB wrapper
+    input_matrix = np.array([[0, x, y]])
+
+    # Pass both input matrix and config JSON to MATLAB wrapper
+    output_matrix = quixmatlab_client.simulink_wrapper(input_matrix, config_json)
 
     # Incorporating result to row
     row["x_new"] = output_matrix[0][0]