Stream class (gaze, eye states, IMU)#

class pyneon.Stream(source: DataFrame | Path | str)#

Bases: BaseTabular

Container for continuous data streams (gaze, eye states, IMU).

Data is indexed by timestamps in nanoseconds (timestamp [ns]).

Parameters:

sourcepandas.DataFrame or pathlib.Path or str

Source of the stream data. Can be either:

pandas.DataFrame: Must contain a timestamp [ns] column or index.
pathlib.Path or str: Path to a stream data file. Supported file formats:
- .csv: Pupil Cloud format CSV file
- .raw: Native format (requires .time and .dtype files in the same directory)

Note: Native format columns are automatically renamed to Pupil Cloud format for consistency. For example, gyro_x -> gyro x [deg/s].

Attributes:

filepathlib.Path or None: Path to the source file(s). None if initialized from DataFrame.
datapandas.DataFrame: Stream data with timestamp [ns] as index.
typestr: Inferred stream type based on data columns.

Methods

`annotate_events`(events[, overwrite, inplace])	Annotate stream data with event IDs based on event time intervals.
`apply_homographies`(homographies[, ...])	Compute gaze locations in surface coordinates using provided homographies based on gaze pixel coordinates and append them to the stream data.
`compute_azimuth_and_elevation`([method, ...])	Compute gaze azimuth and elevation angles (in degrees) from pixel coordinates based on gaze pixel coordinates and append them to the stream data.
`concat`(other[, float_kind, other_kind, ...])	Concatenate additional columns from another Stream to this Stream.
`copy`()	Create a deep copy of the instance.
`crop`([tmin, tmax, by, inplace])	Extract a subset of stream data within a specified temporal range.
`interpolate`([new_ts, float_kind, ...])	Interpolate the stream to new timestamps.
`interpolate_events`(events[, buffer, ...])	Interpolate data during the duration of events in the stream data.
`restrict`(other[, inplace])	Align this stream's temporal range to match another stream.
`save`(output_path)	Save the data to a CSV file.
`time_to_ts`(time)	Convert relative time(s) in seconds to the closest timestamp(s) in nanoseconds.
`window_average`(new_ts[, window_size, inplace])	Take the average over a time window to obtain smoothed data at new timestamps.

Examples

Load from Pupil Cloud CSV:

>>> from pyneon import Stream
>>> gaze = Stream("gaze.csv")

Load from native format:

>>> gaze = Stream("gaze ps1.raw") # Or "gaze_200hz.raw"

Create from DataFrame:

>>> df = pd.DataFrame({"timestamp [ns]": [...], "gaze x [px]": [...]})
>>> gaze = Stream(df)

data: DataFrame#

file: Path | None#

type: str#

property timestamps: ndarray#

Timestamps of the stream in nanoseconds.

Returns:

numpy.ndarray: Array of timestamps in nanoseconds (Unix time).

property ts: ndarray#

Alias for timestamps.

Returns:

numpy.ndarray: Array of timestamps in nanoseconds (Unix time).

property first_ts: int#

First timestamp of the stream in nanoseconds.

Returns:

int: First timestamp in nanoseconds (Unix time).

property last_ts: int#

Last timestamp of the stream in nanoseconds.

Returns:

int: Last timestamp in nanoseconds (Unix time).

property ts_diff: ndarray#

Difference between consecutive timestamps.

Returns:

numpy.ndarray: Array of time differences in nanoseconds.

property times: ndarray#

Timestamps converted to seconds relative to stream start.

Returns:

numpy.ndarray: Array of times in seconds, starting from 0.

property duration: float#

Duration of the stream in seconds.

Returns:

float: Total duration from first to last timestamp in seconds.

property sampling_freq_effective: float#

Effective/empirical sampling frequency of the stream in Hz.

Returns:

float: Effective sampling frequency in Hz.

property sampling_freq_nominal: int | None#: Nominal sampling frequency in Hz as specified by Pupil Labs (see https://pupil-labs.com/products/neon/specs). None for custom or unknown stream types.

property is_uniformly_sampled: bool#

Whether the stream is uniformly sampled.

Returns:

bool: True if all consecutive timestamp differences are approximately equal.

time_to_ts(time: Number | ndarray) → ndarray#

Convert relative time(s) in seconds to the closest timestamp(s) in nanoseconds.

Parameters:

timenumbers.Number or numpy.ndarray: Time(s) in seconds relative to stream start.

Returns:

numpy.ndarray: Corresponding timestamp(s) in nanoseconds.

crop(tmin: Number | None = None, tmax: Number | None = None, by: Literal['timestamp', 'time', 'sample'] = 'timestamp', inplace: bool = False) → Stream | None#

Extract a subset of stream data within a specified temporal range.

The by parameter determines how tmin and tmax are interpreted: - "timestamp": Absolute Unix timestamps in nanoseconds - "time": Relative time in seconds from the stream’s first sample - "sample": Zero-based sample indices

Both bounds are inclusive. If either bound is omitted, it defaults to the stream’s natural boundary (earliest or latest sample).

Parameters:

tminnumbers.Number, optional: Lower bound of the range to extract (inclusive). If None, starts from the stream’s beginning. Defaults to None.
tmaxnumbers.Number, optional: Upper bound of the range to extract (inclusive). If None, extends to the stream’s end. Defaults to None.
by{“timestamp”, “time”, “sample”}, optional: Unit used to interpret tmin and tmax. Defaults to "timestamp".
inplacebool, optional: If True, replace current data. Otherwise returns a new instance. Defaults to False.

Returns:

Stream or None: A new Stream instance with modified data if inplace=False, otherwise None.

Raises:

ValueError: If both tmin and tmax are None, if bounds are negative, or if no data falls within the specified range.

Examples

Crop to the first 0.5 seconds of data:

>>> stream_500ms = stream.crop(tmin=0, tmax=0.5, by="time")

Crop using absolute timestamps:

>>> cropped = stream.crop(tmin=start_ts, tmax=end_ts, by="timestamp")

Extract samples 100 through 200:

>>> samples = stream.crop(tmin=100, tmax=200, by="sample")

restrict(other: Stream, inplace: bool = False) → Stream | None#

Align this stream’s temporal range to match another stream.

This method crops the data to include only samples between the first and last timestamps of the reference stream. It is equivalent to calling crop(tmin=other.first_ts, tmax=other.last_ts, by="timestamp").

Useful for ensuring temporal alignment across multiple data streams, particularly when streams have different start or end times.

Parameters:

otherStream: Reference stream whose temporal boundaries define the cropping range.
inplacebool, optional: If True, replace current data. Otherwise returns a new instance. Defaults to False.

Returns:

Stream or None: A new Stream instance with modified data if inplace=False, otherwise None.

Examples

Align IMU data to match the temporal extent of gaze data:

>>> imu_aligned = imu.restrict(gaze)

Interpolate the stream to new timestamps. Useful for temporal synchronization (e.g., stream-to-stream, stream-to-video) or resampling to a uniform rate.

Data columns of float type are interpolated using the method specified by float_kind, while other columns use the method specified by other_kind. This distinction allows for appropriate interpolation methods based on data type.

Parameters:

new_tsnumpy.ndarray, optional: Target timestamps (in nanoseconds) for the resampled data. If None, timestamps are auto-generated at uniform intervals based on sampling_freq_nominal. Defaults to None.
float_kindstr or int, optional: Kind of interpolation applied to columns of float type. See scipy.interpolate.interp1d for details. Defaults to “linear”.
other_kindstr or int, optional: Kind of interpolation applied to columns of other types. See scipy.interpolate.interp1d for details. Only “nearest”, “nearest-up”, “previous”, and “next” are recommended. Defaults to “nearest”.
max_gap_msint, optional: Maximum allowed distance (in milliseconds) to both adjacent original timestamps (left and right). A requested new timestamp will be ignored if its distance to the immediate left OR right original timestamp is greater than or equal to max_gap_ms (no interpolation will be performed at that timestamp). Defaults to 500.
inplacebool, optional: If True, replace current data. Otherwise returns a new instance. Defaults to False.

Returns:

Stream or None: A new Stream instance with modified data if inplace=False, otherwise None.

Notes

Timestamps in new_ts that fall outside the original data range will have NaN values in the interpolated stream.
Column data types are preserved after interpolation.
Uses scipy.interpolate.interp1d internally.

Examples

Interpolate gaze data to uniform 200 Hz sampling:

>>> gaze_uniform = gaze.interpolate()

Align gaze data to IMU timestamps:

>>> gaze_on_imu = gaze.interpolate(new_ts=imu.ts)

annotate_events(events: Events, overwrite: bool = False, inplace: bool = False) → Stream | None#

Annotate stream data with event IDs based on event time intervals.

Parameters:

eventsEvents: Events instance containing the events to annotate. The events must have a valid id_name attribute, as well as start timestamp [ns] and end timestamp [ns] columns.
overwritebool, optional: If True, overwrite existing event ID annotations in the stream data. Defaults to False.
inplacebool, optional: If True, replace current data. Otherwise returns a new instance. Defaults to False.

Returns:

Stream or None: A new Stream instance with modified data if inplace=False, otherwise None.

Raises:

ValueError: If no event ID column is known for the Events instance.
KeyError: If the expected event ID column is not found in the Events data.

Interpolate data during the duration of events in the stream data. Particularly useful for repairing blink artifacts in eye states or gaze data. Similar to mne.preprocessing.eyetracking.interpolate_blinks().

Parameters:

eventsEvents: Events instance containing the events to interpolate. The events must have start timestamp [ns] and end timestamp [ns] columns.
buffernumbers.Number or tuple[numbers.Number, numbers.Number], optional: The time before and after an event (in seconds) to consider invalid. If a single number is provided, the same buffer is applied to both before and after the event. Defaults to 0.05.
float_kindstr or int, optional: Kind of interpolation applied to columns of float type. See scipy.interpolate.interp1d for details. Defaults to “linear”.
other_kindstr or int, optional: Kind of interpolation applied to columns of other types. See scipy.interpolate.interp1d for details. Only “nearest”, “nearest-up”, “previous”, and “next” are recommended. Defaults to “nearest”.
max_gap_msint, optional: Maximum allowed distance (in milliseconds) to both adjacent original timestamps (left and right). A requested new timestamp will be ignored if its distance to the immediate left OR right original timestamp is greater than or equal to max_gap_ms (no interpolation will be performed at that timestamp). Defaults to 500.
inplacebool, optional: If True, replace current data. Otherwise returns a new instance. Defaults to False.

Returns:

Stream or None: A new Stream instance with modified data if inplace=False, otherwise None.

Examples

Interpolate eye states data during blinks with a 50 ms buffer before and after:

>>> eye_states = eye_states.interpolate_events(blinks, buffer=0.05)

window_average(new_ts: ndarray, window_size: int | None = None, inplace: bool = False) → Stream | None#

Take the average over a time window to obtain smoothed data at new timestamps.

Parameters:

new_tsnumpy.ndarray

An array of new timestamps (in nanoseconds) at which to evaluate the averaged signal. Must be coarser than the source sampling, i.e.:

>>> np.median(np.diff(new_ts)) > np.median(np.diff(data.index))

window_sizeint, optional

The size of the time window (in nanoseconds) over which to compute the average around each new timestamp. If None (default), the window size is set to the median interval between the new timestamps, i.e., np.median(np.diff(new_ts)). The window size must be larger than the median interval between the original data timestamps, i.e., window_size > np.median(np.diff(data.index)).

inplacebool, optional

If True, replace current data. Otherwise returns a new instance. Defaults to False.

Returns:

Stream or None: A new Stream instance with modified data if inplace=False, otherwise None.

compute_azimuth_and_elevation(method: Literal['linear'] = 'linear', overwrite: bool = False, inplace: bool = False) → Stream | None#

Compute gaze azimuth and elevation angles (in degrees) from pixel coordinates based on gaze pixel coordinates and append them to the stream data.

The stream data must contain the required gaze columns: gaze x [px] and gaze y [px].

Parameters:

method{“linear”}, optional: Method to compute gaze angles. Defaults to “linear”.
overwritebool, optional: Only applicable if azimuth and elevation columns already exist. If True, overwrite existing columns. If False, raise an error. Defaults to False.
inplacebool, optional: If True, replace current data. Otherwise returns a new instance. Defaults to False.

Returns:

Stream or None: A new Stream instance with modified data if inplace=False, otherwise None.

Raises:

ValueError: If required gaze columns are not present in the data.

apply_homographies(homographies: Stream, max_gap_ms: Number = 500, overwrite: bool = False, inplace: bool = False) → Stream | None#

Compute gaze locations in surface coordinates using provided homographies based on gaze pixel coordinates and append them to the stream data.

Since homographies are estimated per video frame and might not be available for every frame, they need to be resampled/interpolated to the timestamps of the gaze data before application.

The stream data must contain the required gaze columns: gaze x [px] and gaze y [px]. The output stream will contain two new columns: gaze x [surface coord] and gaze y [surface coord].

Parameters:

homographiesStream

Stream indexed by “timestamp [ns]” with columns “homography (0,0)” through “homography (2,2)”, corresponding to the 9 elements of the estimated 3x3 homography matrix for each retained frame.

Returned by pyneon.find_homographies().

max_gap_msint, optional

Maximum allowed distance (in milliseconds) to both adjacent original timestamps (left and right). A requested new timestamp will be ignored if its distance to the immediate left OR right original timestamp is greater than or equal to max_gap_ms (no interpolation will be performed at that timestamp). Defaults to 500.

overwritebool, optional

Only applicable if surface gaze columns already exist. If True, overwrite existing columns. If False, raise an error. Defaults to False.

inplacebool, optional

If True, replace current data. Otherwise returns a new instance. Defaults to False.

Returns:

Stream or None: A new Stream instance with modified data if inplace=False, otherwise None.

concat(other: Stream, float_kind: str | int = 'linear', other_kind: str | int = 'nearest', max_gap_ms: Number | None = 500, inplace: bool = False) → Stream | None#

Concatenate additional columns from another Stream to this Stream. The other Stream will be interpolated to the timestamps of this Stream to achieve temporal alignment. See interpolate() for details.

Parameters:

otherStream: The other stream to concatenate.
float_kindstr or int, optional: Kind of interpolation applied to columns of float type. See scipy.interpolate.interp1d for details. Defaults to “linear”.
other_kindstr or int, optional: Kind of interpolation applied to columns of other types. See scipy.interpolate.interp1d for details. Only “nearest”, “nearest-up”, “previous”, and “next” are recommended. Defaults to “nearest”.
max_gap_msint, optional: Maximum allowed distance (in milliseconds) to both adjacent original timestamps (left and right). A requested new timestamp will be ignored if its distance to the immediate left OR right original timestamp is greater than or equal to max_gap_ms (no interpolation will be performed at that timestamp). Defaults to 500.
inplacebool, optional: If True, replace current data. Otherwise returns a new instance. Defaults to False.

Returns:

Stream or None: A new Stream instance with modified data if inplace=False, otherwise None.

Notes

To concatenate multiple Stream throughout the recording, you can also use Recording.concat_streams().

Stream class (gaze, eye states, IMU)#

This Page