sen::db Python bindings¶

import sen_db_python as sen

inp = sen.Input("/path/to/recording_directory")
print(inp.path)        # str -- the recording directory
summary = inp.summary  # Summary object

cursor = inp.begin()
cursor.advance()              # move to the first entry
while not cursor.atEnd:
    entry = cursor.entry
    # entry.time     -> datetime.timedelta since epoch
    # entry.payload  -> one of PropertyChange / Event / Keyframe /
    #                   Creation / Deletion / End
    ...
    cursor.advance()

if isinstance(entry.payload, sen.PropertyChange):
    ...
elif isinstance(entry.payload, sen.Event):
    ...

snap = keyframe.snapshots[0]
print(snap.className)              # 'school.Student'
print(snap.propertyNames)          # ['firstName', 'surName', 'focusLevel', ...]
print(snap.focusLevel)             # 0.42

def find_event(reg, class_name, event_name):
    spec = reg.getTypeSpec(class_name)
    if spec is None:
        return None
    data = spec["data"]["value"]
    for event in data.get("events", []):
        if event["name"] == event_name:
            return event
    for parent in data.get("parents", []):
        found = find_event(reg, parent, event_name)
        if found is not None:
            return found
    return None

inp = sen.Input("/path/to/recording")
cursor = inp.begin()
cursor.advance()
objects = {}
while not cursor.atEnd:
    p = cursor.entry.payload
    if isinstance(p, sen.Creation):
        objects[p.objectId] = {"name": p.name, "className": p.className, "bus": p.busName}
    cursor.advance()

from datetime import datetime

EPOCH = datetime(1970, 1, 1)
wall_clock = EPOCH + entry.time

from collections import defaultdict
counts = defaultdict(int)
# ... walk ...
counts[key] += 1
print(dict(counts))

samples = defaultdict(list)
# ... walk: samples[key].append(value) ...
for key, values in samples.items():
    print(f"{key}: n={len(values)} min={min(values):.3f} "
          f"mean={sum(values)/len(values):.3f} max={max(values):.3f}")

top = sorted(counts.items(), key=lambda kv: -kv[1])[:10]
for key, count in top:
    print(f"{key}: {count}")

from datetime import timedelta
window = timedelta(seconds=5)
bins = defaultdict(int)
# ... walk: bins[int(entry.time / window)] += 1 ...
for idx in sorted(bins):
    print(f"t={idx*5:>4}s n={bins[idx]}")

EXAMPLES_PER_KEY = 3
examples = defaultdict(list)
# ... walk: keep only the first few per key ...
if len(examples[key]) < EXAMPLES_PER_KEY:
    examples[key].append(value)

import sen_db_python as sen
from collections import defaultdict
from datetime import datetime

EPOCH = datetime(1970, 1, 1)
inp = sen.Input("/data/recordings/school_run_42")

# 1. Sanity-check the schema via the type registry.
reg = inp.getTypes()
student_spec = reg.getTypeSpec("school.Student")
assert student_spec is not None
property_names = [p["name"] for p in student_spec["data"]["value"]["properties"]]
assert "focusLevel" in property_names

# 2. Walk the recording once, collecting samples per student and counting events.
samples = defaultdict(list)
names_by_id = {}
event_counts = defaultdict(int)

cursor = inp.begin()
cursor.advance()
while not cursor.atEnd:
    p = cursor.entry.payload
    if isinstance(p, sen.Creation):
        names_by_id[p.objectId] = p.name
    elif isinstance(p, sen.PropertyChange) and p.name == "focusLevel":
        samples[p.objectId].append(p.value)
    elif isinstance(p, sen.Event):
        event_counts[p.name] += 1
    cursor.advance()

# 3. Aggregate and report.
print(f"Window: {EPOCH + inp.summary.firstTime} .. {EPOCH + inp.summary.lastTime}")

print("\nFocus statistics:")
for object_id, values in samples.items():
    name = names_by_id.get(object_id, f"<id {object_id}>")
    mean = sum(values) / len(values)
    print(f"  {name:<20} n={len(values):>4} mean={mean:.3f}")

print("\nEvent counts:")
for name, count in sorted(event_counts.items(), key=lambda kv: -kv[1]):
    print(f"  {name:<24} {count}")