The import window for a long run, chosen on a preview.
A stream that would take a large share of the machine — a quarter of
its memory, io.rattlesnake.LARGE_STREAM_SHARE — is not imported
whole without asking (Brandon, 2026-09-18: a 22.5 GB run fits a 64 GB
machine and not a 32 GB one). This dialog is the question: one
channel's envelope over the whole run to read the run by, the
truncate reading's own span over it to choose the stretch, the
channels to keep, and what the choice costs in samples and gigabytes
against the machine's memory. Import reads exactly that — the same
import_file(path, start=, stop=, channels=) a script says, so the
journal replays the import as it was made.
The preview is one channel, and the window chosen on it applies to
every channel: right for a run, and a trust that the run is
stationary across channels, which the chooser is there to check.
Classes:
| Name |
Description |
StreamWindowDialog |
How much of a run to import, chosen on one channel's envelope.
|
Functions:
| Name |
Description |
ask_stream_window |
Put the dialog up; the import options chosen, or None to skip the file.
|
Classes
StreamWindowDialog(path: str | PathLike, summary: Mapping[str, Any], colors: Mapping[str, str], parent: QWidget | None = None)
Bases: QDialog
How much of a run to import, chosen on one channel's envelope.
options() is the answer in the importer's own vocabulary: empty
for the whole run, start/stop for a window, channels for a
subset — only what differs from importing whole, so the journal
line stays the plain one when nothing was cut.
Methods:
| Name |
Description |
channels |
The checked channels, in table order.
|
options |
The import's keyword arguments: only what differs from whole.
|
Source code in src/visualdynamics/gui/stream_window.py
| def __init__(self, path: str | os.PathLike, summary: Mapping[str, Any],
colors: Mapping[str, str], parent: QWidget | None = None) -> None:
super().__init__(parent)
import pyqtgraph as pg
self.path: str = str(path)
self.summary: dict[str, Any] = dict(summary)
self.colors: dict[str, str] = dict(colors)
self.rate: float = float(summary['sample_rate'])
streams = list(summary['streams'])
#: the stream being previewed — the largest, since it is the one
#: that raised the question
self.stream: dict[str, Any] = max(streams, key=lambda s: s['bytes'])
#: the last preview read, for whoever asks what was drawn
self.preview: dict[str, Any] | None = None
self.overlay: TruncationOverlay | None = None
self._curves: list[Any] = []
base = os.path.basename(self.path)
self.setWindowTitle(f'Import a part of {base}')
memory = int(summary.get('memory') or 0)
total = sum(s['bytes'] for s in streams)
words = (f'{base} holds {len(summary["channels"])} channels for '
f'{self.stream["seconds"]:.4g} s at {self.rate:g} Hz: '
f'{total / GB:.2f} GB of time data')
if memory:
words += f' on a machine with {memory / GB:.0f} GB'
words += ('. Choose the stretch and the channels to import; the '
'preview is one channel over the whole run.')
self.header: QLabel = QLabel(words)
self.header.setWordWrap(True)
self.channel_box: QComboBox = QComboBox()
self.channel_box.addItems(list(summary['channels']))
self.channel_box.setToolTip('Which channel the preview shows')
self.stream_box: QComboBox = QComboBox()
for stream in streams:
self.stream_box.addItem(
f'{stream["key"]} ({stream["seconds"]:.4g} s)', stream['key'])
self.stream_box.setCurrentIndex(streams.index(self.stream))
self.stream_box.setVisible(len(streams) > 1)
self.plot: Any = pg.PlotWidget()
self.plot.setBackground(background_brush(self.colors))
foreground = pg.mkPen(self.colors['plot_foreground'])
for edge in ('left', 'bottom'):
axis = self.plot.getAxis(edge)
axis.setPen(foreground)
axis.setTextPen(foreground)
self.plot.setLabel('bottom', 'time [s]')
self.plot.setMinimumSize(560, 260)
self.panel: TruncatePanel = TruncatePanel()
self.panel.title.setText('Window')
# the dialog's Import is the apply; a second button that made a
# record from a run not yet imported would be a promise
self.panel.apply_button.hide()
# the usual answer for a long run is its end — the stretch at
# full level after the ramp — so the end is one number away
# (Brandon, 2026-09-19): the seconds before the end to import,
# by `rattlesnake.last_window`, the rule a script's `last=`
# reads too. Zero is the field standing by, shown as a dash.
self.last_box: QDoubleSpinBox = QDoubleSpinBox()
self.last_box.setRange(0.0, 1e9)
self.last_box.setDecimals(1)
self.last_box.setSuffix(' s')
self.last_box.setSpecialValueText('—')
self.last_box.setKeyboardTracking(False)
self.last_box.setToolTip('Import only this many seconds before the '
'end of the run; a run shorter than that '
'is imported whole')
self.channel_list: QListWidget = QListWidget()
self.channel_list.setToolTip('Uncheck a channel to leave it out')
for dof in summary['channels']:
item = QListWidgetItem(dof)
item.setFlags(item.flags() | Qt.ItemFlag.ItemIsUserCheckable)
item.setCheckState(Qt.CheckState.Checked)
self.channel_list.addItem(item)
self.cost: QLabel = QLabel()
self.cost.setWordWrap(True)
buttons = QDialogButtonBox()
self.import_button: Any = buttons.addButton(
'Import', QDialogButtonBox.ButtonRole.AcceptRole)
buttons.addButton(QDialogButtonBox.StandardButton.Cancel)
buttons.accepted.connect(self.accept)
buttons.rejected.connect(self.reject)
chooser = QHBoxLayout()
chooser.addWidget(QLabel('Preview channel'))
chooser.addWidget(self.channel_box)
chooser.addWidget(self.stream_box)
chooser.addStretch(1)
last_row = QHBoxLayout()
last_row.addWidget(QLabel('Last'))
last_row.addWidget(self.last_box)
last_row.addWidget(QLabel('of the run'))
last_row.addStretch(1)
side = QVBoxLayout()
side.addWidget(self.panel)
side.addLayout(last_row)
side.addWidget(QLabel('Channels'))
side.addWidget(self.channel_list, 1)
middle = QHBoxLayout()
middle.addWidget(self.plot, 1)
middle.addLayout(side)
layout = QVBoxLayout(self)
layout.addWidget(self.header)
layout.addLayout(chooser)
layout.addLayout(middle, 1)
layout.addWidget(self.cost)
layout.addWidget(buttons)
self.channel_box.currentIndexChanged.connect(lambda _i: self._draw_preview())
self.stream_box.currentIndexChanged.connect(self._stream_chosen)
self.panel.changed.connect(self._panel_edited)
self.last_box.valueChanged.connect(self._last_edited)
self.channel_list.itemChanged.connect(lambda _item: self._restate())
self._show_stream(self.stream)
|
Methods:
The checked channels, in table order.
Source code in src/visualdynamics/gui/stream_window.py
| def channels(self) -> list[str]:
"""The checked channels, in table order."""
return [self.channel_list.item(i).text()
for i in range(self.channel_list.count())
if self.channel_list.item(i).checkState() == Qt.CheckState.Checked]
|
options() -> dict[str, Any]
The import's keyword arguments: only what differs from whole.
Source code in src/visualdynamics/gui/stream_window.py
| def options(self) -> dict[str, Any]:
"""The import's keyword arguments: only what differs from whole."""
out: dict[str, Any] = {}
first, last = self._window()
span = self.truncation()
if (first, last) != (0, int(self.stream['samples'])):
out['start'] = float(span.start)
out['stop'] = float(span.stop)
kept = self.channels()
if len(kept) < len(self.summary['channels']):
out['channels'] = kept
return out
|
Functions:
ask_stream_window(parent: QWidget | None, path: str | PathLike, summary: Mapping[str, Any], colors: Mapping[str, str]) -> dict[str, Any] | None
Put the dialog up; the import options chosen, or None to skip the file.
Source code in src/visualdynamics/gui/stream_window.py
| def ask_stream_window(parent: QWidget | None, path: str | os.PathLike,
summary: Mapping[str, Any],
colors: Mapping[str, str]) -> dict[str, Any] | None:
"""Put the dialog up; the import options chosen, or None to skip the file."""
dialog = StreamWindowDialog(path, summary, colors, parent)
if dialog.exec() == QDialog.DialogCode.Accepted:
return dialog.options()
return None
|