Cortex (Predictions)

Use the Cortex API to predict next states

The Cortex API provides prediction capabilities, allowing you to predict the next possible states for a given twin or aggregate.

NextStateRequest Model

Used to request state predictions:

from avenieca.api.model import NextStateRequest

nsr = NextStateRequest(
    module_id="aggregate001",      # Required: Module identifier
    range=2,                       # Optional: Range parameter (default: 2)
    recall=2,                      # Optional: Recall parameter (default: 2)
    n=1,                          # Optional: Number of predictions (default: 1)
    status=None,                   # Optional: Filter by status
    current_state=None,            # Optional: Current state ID
    previous_state=None,           # Optional: Previous state IDs
    store_response=False,          # Optional: Store response flag
    store_sequence=False           # Optional: Store sequence flag
)

Parameters

• module_id (str) - The module identifier for predictions
• range (int) - Range parameter for prediction algorithm
• recall (int) - Recall parameter for prediction algorithm
• n (int) - Number of predictions to return
• status (str, optional) - Filter sequences by status
• current_state (int, optional) - Current state ESS ID
• previous_state (list, optional) - List of previous state IDs
• store_response (bool) - Whether to store the response
• store_sequence (bool) - Whether to store as a sequence

Prediction Methods

predictions()

Get predictions with states mapped back to embedding inputs:

from avenieca.api.model import NextStateRequest

nsr = NextStateRequest(
    module_id="aggregate001",
    recall=20,
    range=20,
    n=1,
    status="e"
)

res, status = eca.cortex.predictions(data=nsr)
if status == 200:
    print("Current State:")
    for twin in res.current_state:
        print(f"  ESS ID: {twin.ess_id}")
        print(f"  Module: {twin.module_id}")
        print(f"  State: {twin.state}")
    
    print("\nNext States:")
    for twins_list in res.next_state:
        for twin in twins_list.list:
            print(f"  ESS ID: {twin.ess_id}")
            print(f"  State: {twin.state}")

predictions_raw()

Get predictions with raw aggregate values (numeric state vectors):

from avenieca.api.model import NextStateRequest

nsr = NextStateRequest(
    module_id="aggregate001",
    recall=20,
    range=20,
    n=1,
    status="e"
)

res, status = eca.cortex.predictions_raw(data=nsr)
if status == 200:
    print("Current State:")
    for twin in res.current_state:
        print(f"  ESS ID: {twin.ess_id}")
        print(f"  State vector: {twin.state}")
    
    print("\nNext States:")
    for twins_list in res.next_state:
        for twin in twins_list.list:
            print(f"  State vector: {twin.state}")

Response Models

NextStateResponse

Returned by predictions() with mapped states:

@dataclass
class NextStateResponse(Base):
    current_state: List[Twin]   # Current state twins
    next_state: List[Twins]     # Predicted next states

@dataclass
class Twin(Base):
    aggregate_id: int  # Aggregate ID
    ess_id: int       # ESS database ID
    module_id: str    # Module identifier
    state: str        # State mapped to embedding input

NextStateResponseRaw

Returned by predictions_raw() with raw state vectors:

@dataclass
class NextStateResponseRaw(Base):
    current_state: List[TwinRaw]   # Current state twins
    next_state: List[TwinsRaw]     # Predicted next states

@dataclass
class TwinRaw(Base):
    state: List[float]  # Raw state vector
    aggregate_id: int   # Aggregate ID
    ess_id: int        # ESS database ID
    module_id: str     # Module identifier

Complete Example

import os
from avenieca.api.eca import ECA
from avenieca.api.model import Config, NextStateRequest

# Initialize client
config = Config(
    uri="http://localhost:2580/v1",
    username=os.getenv("USERNAME"),
    password=os.getenv("PASSWORD")
)
eca = ECA(config)

# Create prediction request
nsr = NextStateRequest(
    module_id="aggregate001",
    recall=20,
    range=20,
    n=1,
    status="e"
)

# Get predictions with mapped states
res, status = eca.cortex.predictions(data=nsr)
if status == 200:
    print(f"Found {len(res.next_state)} possible next states")
    for i, twins in enumerate(res.next_state):
        print(f"\nPrediction {i + 1}:")
        for twin in twins.list:
            print(f"  Module: {twin.module_id}")
            print(f"  State: {twin.state}")

# Get raw predictions
res_raw, status = eca.cortex.predictions_raw(data=nsr)
if status == 200:
    print("\nRaw predictions:")
    for twin in res_raw.current_state:
        print(f"Current state vector: {twin.state}")

Use Cases

• Predictive maintenance - Predict equipment failure states
• Behavior modeling - Model user or system behavior patterns
• State transitions - Understand likely state transitions
• Anomaly detection - Identify unexpected state sequences