In AveniECA, a digital twin is a virtual representation of a real-world entity, process, or system that maintains its own state history and can predict future states based on past experiences. Digital twins are the core abstraction that enables the ECA model’s episodic learning and predictive capabilities.
Every digital twin receives state updates through a Kafka topic called a sub_topic (subscriber topic). This is the input channel through which real-world state flows into the twin.
from avenieca.producers import Streamfrom avenieca.data import Signalfrom avenieca.config.broker import Brokerimport osdef capture_state(): # Read from real-world source value = read_sensor() return Signal( state=[value], valence=evaluate(value) )# Configure connection to twinconfig = Broker( url=os.environ["KAFKA_URL"], sub_topic="left_wheel", # Twin's input channel group="vehicle_sensors", pub_topic="" # Not publishing output in this example)# Stream to the twinstream = Stream(config=config, sync_rate=1.0)stream.publish(capture_state)
The Episodic State Space is where digital twins store their state history. Every signal sent to a twin is stored as an ESS entry, creating a complete record of the twin’s evolution over time.
from avenieca.api.model import ESSInsert# Creating an ESS entry for a twiness_entry = ESSInsert( module_id="air_conditioner", # Which twin state=[25.0], # State vector valence=10.0, # Assessment score=4, # Importance embedding_input=1, # Semantic link context=None # Optional metadata)res, status = eca.ess.create(data=ess_entry)
from avenieca.api.eca import ECAfrom avenieca.api.model import Config# Initialize API clientconfig = Config( uri="http://localhost:2580/v1", username=username, password=password)eca = ECA(config)# Get all states for a twinstates, status = eca.ess.get_all(module_id="air_conditioner")# Get specific state by IDstate, status = eca.ess.get_one(module_id="air_conditioner", db_id=8)# Search for similar statesfrom avenieca.api.model import Searchsearch_results, status = eca.ess.search(data=Search( module_id="air_conditioner", state=[18.0], # Find states similar to this limit=5))
Purpose: Represent a single, specific aspect or component
# Simple individual moduletemp_signal = Signal( state=[22.5], # Just temperature valence=10.0)# Store to ESSess = ESSInsert( module_id="temperature", state=[22.5], valence=10.0, score=1)
Digital twins use their episodic memory (sequences of past states) to predict likely future states through the Cortex API:
from avenieca.api.model import NextStateRequest# Request predictionrequest = NextStateRequest( module_id="climate_aggregate", recall=20, # Look back at last 20 states range=20, # Consider states within range of 20 n=3, # Predict 3 most likely next states status="e" # Use sequences with status "e")response, status = eca.cortex.predictions(data=request)# response.current_state: The twin's current state# response.next_state: List of n predicted next states
Compare actual next state with predictions. Large deviations indicate anomalies:
predictions = eca.cortex.predictions(data=request)actual_next = get_current_state()# Check if actual matches any predictionis_anomaly = not any( is_similar(actual_next, pred) for pred in predictions.next_state)
Preventive Maintenance
Predict when equipment will enter failure states:
# Predict next 10 statesfuture = predict_sequence(module_id="machine", steps=10)# Check if any predicted state has negative valencefailure_imminent = any( state.valence < -50 for state in future)
Optimization
Test different actions to find optimal outcomes:
# Simulate different AC power levelsoutcomes = []for power in [25, 50, 75, 100]: simulated_state = create_state(ac_power=power) prediction = predict_outcome(simulated_state) outcomes.append((power, prediction.valence))# Choose action with best predicted valencebest_action = max(outcomes, key=lambda x: x[1])