Sequences track temporal relationships between ESS entries for a twin or aggregate.
Sequence Models
SequenceInsert
Used to create or update sequences:
from avenieca.api.model import SequenceInsert
sequence = SequenceInsert(
module_id="air_conditioner", # Required: Module identifier
instance_id=10, # Required: ESS instance ID
status="n", # Optional: Status (default: "n")
context=None # Optional: Context string
)
SequenceResponse
Returned when retrieving sequences:
@dataclass
class SequenceResponse(Base):
id: int # Database ID
module_id: str # Module identifier
instance_id: int # ESS instance ID
status: str # Status
created_at: str # Creation timestamp
updated_at: str # Last update timestamp
context: Optional[str] # Context string
Sequence Methods
create()
Create a new sequence:
from avenieca.api.model import SequenceInsert
sequence = SequenceInsert(
module_id="air_conditioner",
instance_id=10,
status="e",
context=None
)
res, status = eca.sequence.create(data=sequence)
if status == 201:
print(f"Created sequence with ID: {res.id}")
print(f"Module: {res.module_id}")
print(f"Instance: {res.instance_id}")
get_all()
Retrieve all sequences for a module:
res, status = eca.sequence.get_all(module_id="air_conditioner")
if status == 200:
for seq in res:
print(f"Sequence {seq.id}: instance={seq.instance_id}, status={seq.status}")
get_one()
Get a specific sequence by database ID:
res, status = eca.sequence.get_one(module_id="air_conditioner", db_id=4)
if status == 200:
print(f"Instance ID: {res.instance_id}")
print(f"Status: {res.status}")
print(f"Created: {res.created_at}")
update()
Update an existing sequence:
from avenieca.api.model import SequenceInsert
updated_sequence = SequenceInsert(
module_id="air_conditioner",
instance_id=10,
status="c", # Changed status
context="Updated context"
)
res, status = eca.sequence.update(
module_id="air_conditioner",
db_id=4,
data=updated_sequence
)
if status == 200:
print(f"Updated sequence {res.id}")
Sequences cannot be deleted through the API. Once created, they remain as part of the episodic history. To mark a sequence as complete, update its status field.
Complete Example
Here’s a complete workflow for working with sequences:
import os
from avenieca.api.eca import ECA
from avenieca.api.model import Config, SequenceInsert
# Initialize client
config = Config(
uri="http://localhost:2580/v1",
username=os.getenv("USERNAME"),
password=os.getenv("PASSWORD")
)
eca = ECA(config)
# Create a sequence
sequence = SequenceInsert(
module_id="aggregate001",
instance_id=34,
status="sk"
)
res, status = eca.sequence.create(data=sequence)
if status == 201:
sequence_id = res.id
print(f"Created sequence {sequence_id}")
# Get the sequence
res, status = eca.sequence.get_one(
module_id="aggregate001",
db_id=sequence_id
)
print(f"Retrieved: {res.module_id} - {res.status}")
# Update the sequence
updated = SequenceInsert(
module_id="aggregate001",
instance_id=34,
status="e"
)
res, status = eca.sequence.update(
module_id="aggregate001",
db_id=sequence_id,
data=updated
)
print(f"Updated status to: {res.status}")
# Get all sequences
res, status = eca.sequence.get_all(module_id="aggregate001")
print(f"Total sequences: {len(res)}")
Status Codes
Common status values used in sequences:
n - New/Not startede - Ended/Completedsk - Skippedc - Current/In progress
These status codes help track the lifecycle of sequences in your ECA system.