brace.RealTimeGraphing.Graphing package

Submodules

brace.RealTimeGraphing.Graphing.AnimatedGraphManager module

class brace.RealTimeGraphing.Graphing.AnimatedGraphManager.AnimatedGraphManager(TRIAL_TIME: float = 0, FPS: int = 20, realTimeType: RealTimeType = RealTimeType.WRAPAROUND, backend: str = 'matplotlib')

Bases: object

Class that manages a set of animated plots and stores in the data internally. This serves as a consumer of a queue that takes in a NamedTuple that contains n number of measurements and a timestamp.

addNewDatastream(fig: Figure | GraphicsLayoutWidget, ax: ndarray[Axes] | ndarray[PlotItem], namedTupleType: type, propertyDataMapping: dict[str, Line2D] | dict[str, PlotDataItem], graphDownSampleRate: int = 20, MAX_TIME_RANGE: float = 10, func: Callable[[Figure, ndarray[Axes], dict], None] = None) → None

Registers a new NamedTuple type and the corresponding plots and lines to be graphed.

Parameters:

• fig (matplotlib.figure.Figure | pyqtgraph.GraphicsLayoutWidget) – Preconfigured figure from plt.subplots(). Or the GraphicsLayoutWidget that is in the GUI.

• ax (numpy.ndarray[matplotlib.axes.Axes] | numpy.ndarray[PlotItem]) – Axes preconfigured from plt.subplots() or handcreated for pyqtgraph. Most likely a 2D array used to manipulate the subplots.

• namedTupleType (type) – A NamedTuple implementation (type) that represents a datapoint for this particular datastream.

• propertyDataMapping (dict[str, Line2D] | dict[str, PlotDataItem]) – A dictionary mapping between the lines and the instance variables of the namedTupleType.

• graphDownSampleRate (int) – The number of samples in between a datapoint when real time graphing.

• MAX_TIME_RANGE (float) – The length of time in seconds that the graphing will sample (i.e. from the last datapoint, how far back it should start).

• func (Callable[[Figure, np.ndarray[Axes], dict], None]) – Callback function that handles the final drawing of the graph after finishing the real time graph.

Returns:

None

Return type:

None

clearQueues() → None

Empties the multiprocessing queue to remove old datapoints that may be in the queue to prevent leakages into future plots.

Returns:

None

Return type:

None

close() → None

Closes the queue for any processing. Commands other than setting the multiprocessing queue are not considered safe.

Returns:

None

Return type:

None

drawFinalGraph(**kwargs) → None

Draws the final graph with all of the datapoints added and with the x-axis and y-axis limits updated to fit the dataset.

Parameters:

kwargs (dict[str, Any]) – Dictionary of keywords that should be applied to the callback function.

Returns:

None

Return type:

None

getCounter() → None

getQueue() → Queue

Returns the multiprocessing queue that should be used for queuing NamedTuple messages.

Returns:

Multiprocessing queue that the AnimatedGraphManager should use for

reading datapoints. :rtype: multiprocessing.Queue

mplStart() → None

Main loop that reads the queue and updates the plot, within a specific time frame. This is a legacy plotting to be used with Matplotlib.

Returns:

None

Return type:

None

noAnimation() → None

This function simply parses all of the data until all the data is empty. This is most useful in plotting static plots, such as those with the Simulator.

Returns:

None

Return type:

None

qtDraw(fig: GraphicsLayoutWidget) → None

This function should be executed periodically on a timer (QTimer works well) during data streaming. This plots the datapoints of the figure given; not all figures are drawn as an optimization.

Parameters:

fig (GraphicsLayoutWidget) – The GraphicsLayoutWidget or Figure whose plots should be drawn using the currently available datapoints.

Returns:

None

Return type:

None

qtRun() → None

This function should be run in the background (thread-level parallelism) when data streaming is executed. This parses the datapoints, but does not draw them outright. For pyqtgraph plots.

Returns:

None

Return type:

None

reset() → None

Resets the datalines in this AnimatedGraphManager for all data. Flags for streaming are also reset. This should be performed every stream session.

Returns:

None

Return type:

None

setQueue(queue: Queue) → None

Supplies a multiprocessing queue (other than the one created automatically in the AnimatedGraphManager) that should be used for reading datapoints in this AnimatedGraphManager.

Parameters:

queue – The multiprocessing queue to read from.

Returns:

None

Return type:

None

setupAnimation(fig: Figure) → None

Sets up an matplotlib animation for the figure. Blitting is done to increase performance.

Parameters:

fig (Figure) – The matplotlib figure to be animated.

Returns:

None

Return type:

None

class brace.RealTimeGraphing.Graphing.AnimatedGraphManager.FigureAxes(fig: GraphicsLayoutWidget | Figure, ax: ndarray[Axes] | ndarray[PlotItem], func: Callable[[Figure, ndarray[Axes], dict], None] = None)

Bases: object

exception brace.RealTimeGraphing.Graphing.AnimatedGraphManager.MismatchedTupleException

Bases: Exception

Raised for differences between the size of the tuple from the producer and the total number of datasets in the graph being drawn.

class brace.RealTimeGraphing.Graphing.AnimatedGraphManager.RealTimeType(*values)

Bases: IntEnum

Enum for classifying two types of real time drawing.

SLIDING_WINDOW = 1

WRAPAROUND = 0

brace.RealTimeGraphing.Graphing.DataStream module

class brace.RealTimeGraphing.Graphing.DataStream.DataLine(line: Line2D | PlotDataItem, propertyName: str, backend: str = 'matplotlib')

Bases: object

Class that plots the lines on the graph identifying with NamedTuple fields.

clearDataLine() → None

Clears out all the data in this line including all caches and stored data.

Returns:

None

Return type:

None

getLine() → Line2D | PlotDataItem

Gets the line element for this DataLine that draws in the plot.

Returns:

The line element that draws in the plots.

Return type:

Line2D | PlotDataItem

getPropertyName() → str

Returns the field name of the line, that is based on the NamedTuple name.

Returns:

The field name of NamedTuple that belongs to this line.

Return type:

str

updateLine(startingIndex: int, sampleNumber: int) → None

Updates the line based on a starting index and goes to the end of the array, sampling 1:sampleNumber for the time series datasets. The last element is plotted. This type of plotting is generally used for the “wraparound” type plots.

Parameters:

• startingIndex (int) – The index of the data that should be started for plotting.

• sampleNumber (int) – How many elements to skip before the next sample is plotted.

Returns:

None

Return type:

None

updateLine2(startingIndex: int, sampleNumber: int) → None

Updates the line based on a starting index and goes to the end of the array, sampling 1:sampleNumber for the time series datasets, caching it frame-by-frame for performance. The last element is plotted. This type of plotting is generally used for the “sliding window” type plots.

Parameters:

• startingIndex (int) – The index of the data that should be started for plotting.

• sampleNumber (int) – How many elements to skip before the next sample is plotted.

Returns:

None

Return type:

None

updatePoints(t: ndarray[float], y: ndarray[float]) → None

Updates the internally listed datapoints and the lines for each related dataset.

Parameters:

• t (np.ndarray[float] | list[float]) – A list of data indicating the time component that should be paired with the y-axis data.

• y (np.ndarray[float] | list[float]) – : A list of data that should be paired with the time series which is stored and can be plotted against the axis lines.

Returns:

None

Return type:

None

class brace.RealTimeGraphing.Graphing.DataStream.DataStream(fig: GraphicsLayoutWidget | Figure, axes: list[Axes] | list[PlotItem], lines: dict[str, Line2D] | dict[str, PlotDataItem], namedTupleType: type, graphDownSampleRate: int = 20, MAX_TIME_RANGE: float = 10, backend: str = 'matplotlib')

Bases: object

A class that bridges the figure drawing from the lines and the NamedTuple that is assigned to the lines.

clearData() → None

Clears out the data in all the datalines and in the time list. Indexing for the plots is reset back to the beginning.

Returns:

None

Return type:

None

draw() → None

Draws the plot in the wraparound configuration. If the last datapoint has a time that reaches beyond a multiple of the MAX_TIME_RANGE, then it will refresh and adjust the x-axis limits as needed so that the plotting can continue from the furthest left onwards.

Returns:

None

Return type:

None

drawFinal(padding: float = 0.0) → None

Draws the plot in the final draw configuration. This contains all of the datapoints and updates the x-axis and y-axis limits so that it fits the minimum and maximum values of each of the datasets.

Parameters:

padding (float) – The percentage padding that should be included in the plots after the final draw. Default is 0.00, indicating hard fit.

Returns:

None

Return type:

None

drawSlidingWindow() → None

Draws the plot in the sliding window-type configuration. Determines the window where the furthest right refers to the most immediate datapoint that was added, and the furthest left is the datapoint from t[-1] - MAX_TIME_RANGE forward. If there are no datapoints from that range nothing is drawn, but the x-axis limits are still set to this window.

Returns:

None

Return type:

None

getDatalines() → list[DataLine]

Gets the point/line relations between the plots.

Returns:

The list of DataLines for this DataStream object

Return type:

list[DataLine]

getLines() → list[Line2D] | list[PlotDataItem]

Gets the individual list of lines that should have data plotted.

Returns:

List of lines that plot datapoints.

Return type:

list[Line2D] | list[PlotDataItem]

updatePoints(property: str, tPoints: list[float], dataPoints: list[float]) → None

Updates the data points with new data (low level) for the dataLines

Parameters:

• property (str) – The NamedTuple field name that should be updated with the new data points.

• tPoints (list[float]) – The time list of points to update.

• dataPoints (list[float]) – The y-axis points that should be aligned with the tPoints.

Returns:

None

Return type:

None

updateTime(timePoints: list[float]) → None

Updates the time points with new data (high level)

Parameters:

timePoints (list[float]) – list of time points to append to the end.

Returns:

None

Return type:

None

brace.RealTimeGraphing.Graphing.IDataProducer module

class brace.RealTimeGraphing.Graphing.IDataProducer.IDataProducer(UPDATE_RATE_PER_SECOND: int, TRIAL_TIME: float = 0, startTime: Synchronized = None)

Bases: ABC

A class for data producers that run on an update rate that create NamedTuple datapoints. These are generally then read in by the AnimatedGraphManager. These datapoints can be synchronized in time with a shared start time.

DATA_TOPIC = 'datatypes'

getSharedStartTime() → Synchronized

Gets the shared startTime for this IDataProducer. This should be used to get the primary IDataProducer startTime that should be shared for all the secondaries.

Returns:

The shared start time that should be shared between the IDataProducers.

Return type:

multiprocessing.Value

setSharedStartTime(updateStartTime: float, timeSynchronizationCondition: Condition = None) → None

This should be started in the running loop of each IDataProducer. Sets the startTime value only if this startTime is the primary to synchronize the startTime values. If it is the secondary datastream, then it will pause execution until the synchronized time is set from the primary.

Parameters:

• updateStartTime (float) – A double-precision floating point number representing the start time before the while loop. This value is used to synchronize startTime in multiple processes to achieve more accurate timing.

• timeSynchronizationCondition (multiprocessing.synchronize.Condition) – Multiprocessing condition that pauses execution of the subprocess until the primary sets up the startTime.

Returns:

None

Return type:

None

abstractmethod start(timeSynchronizationCondition: Condition = None, **kwargs) → None

Main update function that runs through data logic and produces output data. Output data in the form of a NamedTuple is added into a shared queue between producer and consumer.

Parameters:

• timeSynchronizationCondition (multiprocessing.synchronize.Condition) – Condition that pauses secondaries flow until the time is set by a primary.

• kwargs (dict[str, Any]) – Keyword arguments that should be passed in for all other IDataProducers implementing this function.

Returns:

None

Return type:

None

Module contents