Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -12,3 +12,4 @@ venv/
MANIFEST
htmlcov/
.coverage
site/
3 changes: 3 additions & 0 deletions docs/api/parametric.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
# Parametric Sequences

::: knowledgecomplex.parametric
37 changes: 33 additions & 4 deletions docs/tutorial.md
Original file line number Diff line number Diff line change
Expand Up @@ -254,7 +254,36 @@ filt[1] # set of element IDs at step 1
filt2 = Filtration.from_function(kc, lambda eid: some_score(eid))
```

## 7. Clique inference
## 7. Parametric sequences

A `ParametricSequence` views a single complex through a parameterized filter. The complex holds all elements across all parameter values; the filter selects which are "active" at each value. Unlike a Filtration, subcomplexes can shrink — elements can appear and disappear.

```python
from knowledgecomplex import ParametricSequence

# Complex has people with active_from/active_until attributes
seq = ParametricSequence(
kc,
values=["Q1", "Q2", "Q3", "Q4"],
filter=lambda elem, t: elem.attrs.get("active_from", "0") <= t < elem.attrs.get("active_until", "9999"),
)

seq["Q2"] # set of element IDs active at Q2
seq.birth("carol") # "Q2" — first value where carol appears
seq.death("bob") # "Q3" — first value where bob disappears
seq.active_at("bob") # ["Q1", "Q2"]
seq.new_at(1) # elements appearing at Q2
seq.removed_at(2) # elements disappearing at Q3 (bob left)
seq.is_monotone # False — people can leave
seq.subcomplex_at(0) # is the Q1 slice boundary-closed?

for value, ids in seq:
print(f"{value}: {len(ids)} elements")
```

The complex is the territory; the parameterized filter is the map.

## 8. Clique inference

Discover higher-order structure hiding in the edge graph:

Expand All @@ -271,7 +300,7 @@ added = infer_faces(kc, "coverage")
preview = infer_faces(kc, "coverage", dry_run=True)
```

## 8. Export and load
## 9. Export and load

```python
# Export schema + instance to a directory
Expand All @@ -292,7 +321,7 @@ save_graph(kc, "data.jsonld", format="json-ld")
load_graph(kc, "data.ttl") # additive loading
```

## 9. Verification and audit
## 10. Verification and audit

```python
# Throwing verification
Expand All @@ -317,7 +346,7 @@ report = audit_file("data/instance.ttl", shapes="data/shapes.ttl",
ontology="data/ontology.ttl")
```

## 10. Pre-built ontologies
## 11. Pre-built ontologies

Three ontologies ship with the package:

Expand Down
100 changes: 38 additions & 62 deletions examples/08_temporal_sweep/temporal_sweep.py
Original file line number Diff line number Diff line change
Expand Up @@ -69,74 +69,50 @@
print(f"Built timeline: {len(kc.element_ids())} total elements across all time")
print()

# ── Manual parameterized sweep ─────────────────────────────────────────────
# ── Parameterized sweep using ParametricSequence ──────────────────────────

# Since we store active_from/active_until as string attributes, we can
# query for elements active at a specific time by comparing attribute values.
from knowledgecomplex import ParametricSequence

def active_filter(elem, t):
"""Element is active at time t if active_from <= t < active_until."""
af = elem.attrs.get("active_from", "0")
au = elem.attrs.get("active_until", "9999")
return af <= t < au

seq = ParametricSequence(kc, values=["1", "2", "3", "4", "5"], filter=active_filter)

print("=== Active subcomplex at each quarter ===")
for t in ["1", "2", "3", "4", "5"]:
# Get active people at time t
active_people = set()
for pid in kc.element_ids(type="Person"):
elem = kc.element(pid)
af = elem.attrs.get("active_from", "0")
au = elem.attrs.get("active_until", "9999")
if af <= t < au:
active_people.add(pid)

# Get active edges at time t
active_edges = set()
for eid in kc.element_ids(type="WorksWith"):
elem = kc.element(eid)
af = elem.attrs.get("active_from", "0")
au = elem.attrs.get("active_until", "9999")
if af <= t < au:
# Only include if both endpoints are active
boundary = kc.boundary(eid)
if boundary <= active_people:
active_edges.add(eid)

# Get active faces
active_faces = set()
for fid in kc.element_ids(type="Squad"):
boundary = kc.boundary(fid)
if boundary <= active_edges:
active_faces.add(fid)

active = active_people | active_edges | active_faces
is_sub = kc.is_subcomplex(active)

print(f" Q{t}: {len(active_people)} people, "
f"{len(active_edges)} collabs, "
f"{len(active_faces)} squads "
f"(valid subcomplex: {is_sub})")
print(f" people: {sorted(active_people)}")

# Show who's new and who left
if t != "1":
prev_t = str(int(t) - 1)
prev_people = set()
for pid in kc.element_ids(type="Person"):
elem = kc.element(pid)
af = elem.attrs.get("active_from", "0")
au = elem.attrs.get("active_until", "9999")
if af <= prev_t < au:
prev_people.add(pid)
joined = active_people - prev_people
left = prev_people - active_people
if joined:
print(f" joined: {sorted(joined)}")
if left:
print(f" left: {sorted(left)}")
for t, active in seq:
print(f" Q{t}: {len(active)} elements "
f"(valid subcomplex: {seq.subcomplex_at(seq.values.index(t))})")
print(f" {sorted(active)}")

i = seq.values.index(t)
new = seq.new_at(i)
removed = seq.removed_at(i)
if new:
print(f" joined: {sorted(new)}")
if removed:
print(f" left: {sorted(removed)}")
print()

# ── Lifecycle queries ──────────────────────────────────────────────────────

print("=== Lifecycle ===")
for person in ["alice", "bob", "carol", "dave", "eve"]:
birth = seq.birth(person)
death = seq.death(person)
active = seq.active_at(person)
print(f" {person:6s} birth=Q{birth} death={'Q'+death if death else 'still active':14s} active={active}")
print()

# ── Key insight ────────────────────────────────────────────────────────────

print("=== Key insight ===")
print(" This is NOT a filtration — the subcomplex at Q4 is not a superset")
print(" of Q3 (bob left). But each time-slice is a valid subcomplex,")
print(" and the full complex contains the complete history.")
print(f"=== Key insight ===")
print(f" is_monotone: {seq.is_monotone}")
print(f" This is NOT a filtration — bob leaves at Q3, so Q3 is not a")
print(f" superset of Q2. But the complex holds the complete history,")
print(f" and the parameterized filter slices it at any time.")
print()
print(" The complex is the territory; the time-slice queries are the maps.")
print(f" The complex is the territory; the parameterized filter is the map.")
print("Done.")
2 changes: 2 additions & 0 deletions knowledgecomplex/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
from knowledgecomplex.schema import SchemaBuilder, vocab, text, TextDescriptor, Codec
from knowledgecomplex.graph import KnowledgeComplex, Element
from knowledgecomplex.filtration import Filtration
from knowledgecomplex.parametric import ParametricSequence
from knowledgecomplex.exceptions import ValidationError, SchemaError, UnknownQueryError
from knowledgecomplex.audit import AuditReport, AuditViolation, audit_file
from knowledgecomplex.io import save_graph, load_graph, dump_graph
Expand Down Expand Up @@ -60,6 +61,7 @@
"KnowledgeComplex", "Element",
# Filtrations
"Filtration",
"ParametricSequence",
# Exceptions
"ValidationError", "SchemaError", "UnknownQueryError",
# File I/O
Expand Down
166 changes: 166 additions & 0 deletions knowledgecomplex/parametric.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,166 @@
"""
knowledgecomplex.parametric — Parametric sequences over knowledge complexes.

A ParametricSequence represents a single complex viewed through a
parameterized filter. Each parameter value selects a subcomplex —
the sequence of subcomplexes can grow, shrink, or change arbitrarily
as the parameter varies.

Unlike :class:`~knowledgecomplex.filtration.Filtration`, which enforces
monotone nesting, a ParametricSequence is observational — it computes
slices lazily from a filter function.
"""

from __future__ import annotations
from typing import Any, Callable, Iterator, TYPE_CHECKING

if TYPE_CHECKING:
from knowledgecomplex.graph import KnowledgeComplex, Element


class ParametricSequence:
"""
A complex viewed through a parameterized filter.

One complex holds all elements. A filter function decides which
elements are active at each parameter value. The result is a
sequence of subcomplexes indexed by parameter values.

Parameters
----------
kc : KnowledgeComplex
The complex containing all elements.
values : list
Ordered parameter values (e.g. ``["Q1", "Q2", "Q3", "Q4"]``).
filter : Callable[[Element, value], bool]
Returns True if the element is active at the given parameter value.

Example
-------
>>> seq = ParametricSequence(kc, values=["1","2","3","4"],
... filter=lambda elem, t: elem.attrs.get("active_from","0") <= t)
>>> seq[0] # element IDs active at "1"
>>> seq.birth("carol") # first value where carol appears
"""

def __init__(
self,
kc: "KnowledgeComplex",
values: list,
filter: Callable[["Element", Any], bool],
) -> None:
self._kc = kc
self._values = list(values)
self._filter = filter
self._cache: dict[int, frozenset[str]] = {}

def _compute(self, index: int) -> frozenset[str]:
"""Compute and cache the element set at a given index."""
if index not in self._cache:
value = self._values[index]
ids = frozenset(
eid for eid in self._kc.element_ids()
if self._filter(self._kc.element(eid), value)
)
self._cache[index] = ids
return self._cache[index]

def __repr__(self) -> str:
return f"ParametricSequence(steps={len(self._values)}, monotone={self.is_monotone})"

# --- Indexing ---

def __getitem__(self, key: int | Any) -> set[str]:
if isinstance(key, int):
return set(self._compute(key))
# Try to look up by parameter value
try:
index = self._values.index(key)
except ValueError:
raise KeyError(f"Parameter value {key!r} not in values list")
return set(self._compute(index))

def __len__(self) -> int:
return len(self._values)

def __iter__(self) -> Iterator[tuple[Any, set[str]]]:
for i, value in enumerate(self._values):
yield value, set(self._compute(i))

# --- Properties ---

@property
def complex(self) -> "KnowledgeComplex":
"""The parent KnowledgeComplex."""
return self._kc

@property
def values(self) -> list:
"""The ordered parameter values."""
return list(self._values)

@property
def is_monotone(self) -> bool:
"""True if every step is a superset of the previous (filtration-like)."""
for i in range(1, len(self._values)):
if not (self._compute(i - 1) <= self._compute(i)):
return False
return True

# --- Queries ---

def birth(self, element_id: str) -> Any:
"""Return the first parameter value where the element appears.

Raises
------
ValueError
If the element does not appear at any parameter value.
"""
for i, value in enumerate(self._values):
if element_id in self._compute(i):
return value
raise ValueError(f"Element '{element_id}' not found at any parameter value")

def death(self, element_id: str) -> Any | None:
"""Return the first parameter value where the element disappears.

Returns None if the element is present at all values after its birth,
or if it never appears.
"""
appeared = False
for i in range(len(self._values)):
present = element_id in self._compute(i)
if present:
appeared = True
elif appeared:
return self._values[i]
return None

def active_at(self, element_id: str) -> list:
"""Return the list of parameter values where the element is present."""
return [
self._values[i]
for i in range(len(self._values))
if element_id in self._compute(i)
]

def new_at(self, index: int) -> set[str]:
"""Return elements appearing at step index that were not in step index-1."""
current = self._compute(index)
if index == 0:
return set(current)
previous = self._compute(index - 1)
return set(current - previous)

def removed_at(self, index: int) -> set[str]:
"""Return elements present at step index-1 that are absent at step index."""
if index == 0:
return set()
current = self._compute(index)
previous = self._compute(index - 1)
return set(previous - current)

def subcomplex_at(self, index: int) -> bool:
"""Check if the slice at index is a valid subcomplex (closed under boundary)."""
return self._kc.is_subcomplex(set(self._compute(index)))
1 change: 1 addition & 0 deletions mkdocs.yml
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ nav:
- Algebraic Topology: api/analysis.md
- Clique Inference: api/clique.md
- Filtrations: api/filtration.md
- Parametric Sequences: api/parametric.md
- Diffs & Sequences: api/diff.md
- File I/O: api/io.md
- Codecs: api/codecs.md
Expand Down
Loading
Loading