Flow
timelines.Flow(
sections=list(),
mode=FlowMode.default,
folded_length=0,
id='',
source_metadata=dict(),
_controller_ref=None,
)A computed flow (sequence of measure visitations).
A Flow represents one possible path through a score, accounting for repeats, jumps, and voltas. It can be: - Computed by ScoreFlowController from MeasureData - Loaded from .flow.csv ground truth - Compared using is_equivalent()
Flows are section-based, using sections (list of PlaythroughSection) for .flow.csv serialization and is_equivalent() comparison.
Flows computed by ScoreFlowController have a controller reference, allowing access to MeasureUnits via iter_units(). Flows loaded from CSV are “detached” and do not have controller access.
Note
MC ranges use the right-open interval convention [mc_start, mc_end), consistent with partitura and TimeToAlign’s TimeInterval model.
Attributes
| Name | Type | Description |
|---|---|---|
| sections | list[PlaythroughSection] |
The sequence of PlaythroughSection objects. |
| mode | FlowMode |
The FlowMode used to compute this flow. |
| folded_length | int |
Number of unique MCs (measures in printed score). |
| id | str |
Identifier for this flow (defaults to mode.value). |
| source_metadata | dict[str, Any] |
Optional metadata from the source MeasureData. |
Methods
| Name | Description |
|---|---|
| diagram | Show playthrough section sequence for this flow. |
| diff_diagram | Show side-by-side comparison of this flow with another. |
| diff_flows | Show differences between this flow and another using sequence alignment. |
| from_csv | Load Flow for specific mode from .flow.csv file. |
| from_dataframe | Create Flow from DataFrame with mc_start, mc_end, atomic_sections fields. |
| from_records | Create Flow from list of dicts with mc_start, mc_end, atomic_sections. |
| from_sections | Create a Flow from PlaythroughSections. |
| is_equivalent | Compare by zipped (mc_start, mc_end) ranges. |
| iter_units | Iterate over MeasureUnits via the controller. |
| to_atomic_sequence | Return flattened sequence of atomic section IDs. |
| to_csv_rows | Export as .flow.csv format rows. |
| to_dataframe | Convert to pandas DataFrame with section information. |
| to_mc_sequence | Return the sequence of MCs in traversal order. |
| to_records | Export as list of dicts (section-based). |
diagram
timelines.Flow.diagram(
width=70,
unicode=True,
show_mcs=False,
show_reasons=True,
)Show playthrough section sequence for this flow.
Parameters
| Name | Type | Description | Default |
|---|---|---|---|
| width | int |
Total width of the diagram in characters. | 70 |
| unicode | bool |
Use Unicode characters (True) or ASCII fallback (False). | True |
| show_mcs | bool |
Whether to expand MC sequences per section. | False |
| show_reasons | bool |
Whether to annotate why each section starts. | True |
Returns
| Name | Type | Description |
|---|---|---|
| 'Diagram' | Diagram object (displays as ASCII in terminal, rich HTML in Jupyter). |
diff_diagram
timelines.Flow.diff_diagram(other, width=80, unicode=True)Show side-by-side comparison of this flow with another.
Parameters
| Name | Type | Description | Default |
|---|---|---|---|
| other | 'Flow' | Another Flow to compare with. | required |
| width | int |
Total width of the diagram in characters. | 80 |
| unicode | bool |
Use Unicode characters (True) or ASCII fallback (False). | True |
Returns
| Name | Type | Description |
|---|---|---|
| 'Diagram' | Diagram object (displays as ASCII in terminal, rich HTML in Jupyter). |
diff_flows
timelines.Flow.diff_flows(other)Show differences between this flow and another using sequence alignment.
Uses difflib to produce a human-readable diff of atomic sequences.
Parameters
| Name | Type | Description | Default |
|---|---|---|---|
| other | 'Flow' | Another Flow to compare against. | required |
Returns
| Name | Type | Description |
|---|---|---|
str |
String showing the alignment/diff between the two flows. |
from_csv
timelines.Flow.from_csv(path, mode)Load Flow for specific mode from .flow.csv file.
Filters CSV to rows matching the given flow_mode.
Note
MC ranges use right-open interval convention [mc_start, mc_end).
Parameters
| Name | Type | Description | Default |
|---|---|---|---|
| path | 'Path | str' | Path to the .flow.csv file. | required |
| mode | FlowMode |
The FlowMode to load. | required |
Returns
| Name | Type | Description |
|---|---|---|
| 'Flow' | New Flow instance with sections populated. |
Raises
| Name | Type | Description |
|---|---|---|
ValueError |
If no entries found for the given mode. |
from_dataframe
timelines.Flow.from_dataframe(df, mode)Create Flow from DataFrame with mc_start, mc_end, atomic_sections fields.
Note
MC ranges use right-open interval convention [mc_start, mc_end).
Parameters
| Name | Type | Description | Default |
|---|---|---|---|
| df | 'pd.DataFrame' | DataFrame with fields: mc_start, mc_end, atomic_sections (or its flow-CSV spelling “atomic_segments”). | required |
| mode | FlowMode |
The FlowMode for this flow. | required |
Returns
| Name | Type | Description |
|---|---|---|
| 'Flow' | New Flow instance with sections populated. |
from_records
timelines.Flow.from_records(records, mode)Create Flow from list of dicts with mc_start, mc_end, atomic_sections.
Note
MC ranges use right-open interval convention [mc_start, mc_end).
Parameters
| Name | Type | Description | Default |
|---|---|---|---|
| records | list[dict] |
List of dicts, each with keys: - mc_start: int (inclusive) - mc_end: int (exclusive, right-open) - atomic_sections: str (semicolon-separated, e.g., “A;B”); the persisted flow-CSV format names this column “atomic_segments”, which is accepted as a fallback key | required |
| mode | FlowMode |
The FlowMode for this flow. | required |
Returns
| Name | Type | Description |
|---|---|---|
| 'Flow' | New Flow instance with sections populated. |
from_sections
timelines.Flow.from_sections(sections, mode, folded_length=None)Create a Flow from PlaythroughSections.
Parameters
| Name | Type | Description | Default |
|---|---|---|---|
| sections | list[PlaythroughSection] |
List of PlaythroughSection objects. | required |
| mode | FlowMode |
The FlowMode for this flow. | required |
| folded_length | int | None |
Number of unique MCs. If None, computed from sections. | None |
Returns
| Name | Type | Description |
|---|---|---|
| 'Flow' | New Flow instance with sections populated. |
is_equivalent
timelines.Flow.is_equivalent(other)Compare by zipped (mc_start, mc_end) ranges.
Two flows are equivalent if they have the same number of sections and each corresponding section has matching mc_start and mc_end.
Note: atomic_section_ids are NOT compared - only MC ranges matter.
Parameters
| Name | Type | Description | Default |
|---|---|---|---|
| other | 'Flow' | Another Flow to compare against. | required |
Returns
| Name | Type | Description |
|---|---|---|
bool |
True if flows are equivalent, False otherwise. |
iter_units
timelines.Flow.iter_units()Iterate over MeasureUnits via the controller.
This provides access to the folded score skeleton (one MeasureUnit per MeasureData row).
Yields
| Name | Type | Description |
|---|---|---|
MeasureUnit |
MeasureUnit objects in MC order. |
Raises
| Name | Type | Description |
|---|---|---|
ValueError |
If Flow is detached from controller (e.g., loaded from CSV). |
Examples
>>> flow = controller.compute_flow()
>>> for unit in flow.iter_units():
... print(f"MC {unit.mc}: next={unit.next}")to_atomic_sequence
timelines.Flow.to_atomic_sequence()Return flattened sequence of atomic section IDs.
This provides a canonical representation of the flow as a sequence of atomic section traversals. Useful for comparing flows and debugging.
Returns
| Name | Type | Description |
|---|---|---|
list[str] |
List of atomic section IDs in traversal order. |
Examples
>>> flow = Flow.from_sections([
... PlaythroughSection(1, 17, ("A", "B")),
... PlaythroughSection(17, 32, ("C",)),
... PlaythroughSection(6, 17, ("B",)),
... ], FlowMode.default)
>>> flow.to_atomic_sequence()
['A', 'B', 'C', 'B']to_csv_rows
timelines.Flow.to_csv_rows(source_file, software_version)Export as .flow.csv format rows.
Returns list of dicts with keys
flow_mode, source_file, software_version, mc_start, mc_end, atomic_sections
Parameters
| Name | Type | Description | Default |
|---|---|---|---|
| source_file | str |
The file that was parsed to produce this flow. | required |
| software_version | str |
Software name and version for reproducibility. | required |
Returns
| Name | Type | Description |
|---|---|---|
list[dict] |
List of dicts ready for CSV writing. |
to_dataframe
timelines.Flow.to_dataframe()Convert to pandas DataFrame with section information.
Returns
| Name | Type | Description |
|---|---|---|
| 'pd.DataFrame' | DataFrame with fields: mc_start, mc_end, atomic_sections | |
| 'pd.DataFrame' | for each PlaythroughSection. |
Note
This returns section-level data. For per-MC data, use iter_units() with controller access.
to_mc_sequence
timelines.Flow.to_mc_sequence()Return the sequence of MCs in traversal order.
Returns
| Name | Type | Description |
|---|---|---|
list[int] |
List of MC values in the order they are visited (right-open intervals). |
to_records
timelines.Flow.to_records()Export as list of dicts (section-based).
Returns
| Name | Type | Description |
|---|---|---|
list[dict] |
List of dicts with keys: mc_start, mc_end, atomic_sections. |