Ros.Topic.Util

Utility functions for working with Topics. These functions are primarily combinators for fusing two Topics in various ways.

Documentation

toList :: Topic IO a -> IO [a]

Produce an infinite list from a Topic.

fromList :: Monad m => [a] -> Topic m a

Produce a Topic from an infinite list.

tee :: Topic IO a -> IO (Topic IO a, Topic IO a)

Tee a Topic into two duplicate Topics. Each returned Topic will receive all the values of the original Topic while any side-effect produced by each step of the original Topic will occur only once.

This version of tee lazily pulls data from the original Topic when it is first required by a consumer of either of the returned Topics. This behavior is crucial when lazily consuming the data stream is preferred. For instance, using interruptible with tee will allow for a chunk of data to be abandoned before being fully consumed as long as neither consumer has forced its way too far down the stream.

This function is useful when two consumers must see all the same elements from a Topic. If the Topic was instead shared, then one consumer might get the first value from the Topic before the second consumer's buffer is created since buffer creation is lazy.

teeEager :: Topic IO a -> IO (Topic IO a, Topic IO a)

This version of tee eagerly pulls data from the original Topic as soon as it is available. This behavior is undesirable when lazily consuming the data stream is preferred. For instance, using interruptible with teeEager will likely not work well. However, teeEager may have slightly better performance than tee.

fan :: Int -> Topic IO a -> IO [Topic IO a]

Fan out one Topic out to a number of duplicate Topics, each of which will produce the same values. Side effects caused by the original Topic's production will occur only once. This is useful when a known number of consumers must see exactly all the same elements.

share :: Topic IO a -> IO (Topic IO a)

Make a Topic shareable among multiple consumers. Each consumer of a Topic gets its own read buffer automatically as soon as it starts pulling items from the Topic. Without calling one of share, tee, or fan on a Topic, the Topic's values will be split among all consumers (e.g. consumer A gets half the values produced by the Topic, while consumer B gets the other half with some unpredictable interleaving). Note that Topics returned by the Ros.Node.subscribe are already shared.

topicRate :: (Functor m, MonadIO m) => Double -> Topic m a -> Topic m a

The application topicRate rate t runs Topic t no faster than rate Hz.

partition :: (a -> Bool) -> Topic IO a -> IO (Topic IO a, Topic IO a)

Splits a Topic into two Topics: the elements of the first Topic all satisfy the given predicate, while none of the elements of the second Topic do.

consecutive :: Monad m => Topic m a -> Topic m (a, a)

Returns a Topic whose values are consecutive values from the original Topic.

(<+>) :: Topic IO a -> Topic IO b -> Topic IO (Either a b) infixl 7

Interleave two Topics. Items from each component Topic will be tagged with an Either constructor and added to the combined Topic as they become available.

everyNew :: Topic IO a -> Topic IO b -> Topic IO (a, b)

Returns a Topic that produces a new pair every time either of the component Topics produces a new value. The value of the other element of the pair will be the newest available value. The resulting Topic will produce a new value at the rate of the faster component Topic, and may contain duplicate consecutive elements.

bothNew :: Topic IO a -> Topic IO b -> Topic IO (a, b)

Returns a Topic that produces a new pair every time both of the component Topics have produced a new value. The composite Topic will produce pairs at the rate of the slower component Topic consisting of the most recent value from each Topic.

merge :: Topic IO a -> Topic IO a -> Topic IO a

Merge two Topics into one. The items from each component Topic will be added to the combined Topic as they become available.

finiteDifference :: (Functor m, Monad m) => (a -> a -> b) -> Topic m a -> Topic m b

Apply a function to each consecutive pair of elements from a Topic.

weightedMeanNormalized :: Monad m => n -> n -> (b -> b -> c) -> (n -> a -> b) -> (c -> a) -> Topic m a -> Topic m a

Compute a running "average" of a Topic using a user-provided normalization function applied to the sum of products. The arguments are a constat alpha that is used to scale the current average, a constant invAlpha used to scale the newest value, a function for adding two scaled values, a function for scaling input values, a function for normalizing the sum of scaled values, and finally the stream to average. Parameterizing over all the arithmetic to this extent allows for the use of denormalizing scaling factors, as might be used to keep all arithmetic integral. An example would be scaling the average by the integer 7, the new value by the integer 1, then normalizing by dividing the sum of scaled values by 8.

simpsonsRule :: (Monad m, Fractional n) => (a -> a -> a) -> (n -> a -> a) -> Topic m a -> Topic m a

Perform numerical integration of a Topic using Simpson's rule applied at three consecutive points. This requires a function for adding values from the Topic, and a function for scaling values by a fractional number.

weightedMean :: (Monad m, Num n) => n -> (a -> a -> a) -> (n -> a -> a) -> Topic m a -> Topic m a

Compute a running "average" of a Topic. The application weightedMean alpha plus scale t sums the product of alpha and the current average with the product of 1 - alpha and the newest value produced by Topic t. The addition and scaling operations are performed using the supplied plus and scale functions.

weightedMean2 :: Monad m => n -> n -> (a -> a -> a) -> (n -> a -> a) -> Topic m a -> Topic m a

Compute a running "average" of a Topic. The application weightedMean2 alpha invAlpha plus scale t sums the product of alpha and the current average with the product of invAlpha and the newest value produced by Topic t. The addition and scaling operations are performed using the supplied plus and scale functions.

filterBy :: Monad m => Topic m (a -> Bool) -> Topic m a -> Topic m a

Use a Topic of functions to filter a Topic of values. The application filterBy t1 t2 causes each function from Topic t1 to be applied to values produced by t2 until it returns True. At that point, the filterBy application produces the accepted value of the t2 and moves on to the next function from t1 which is applied to the rest of t2 in the same manner.

gate :: (Applicative m, Monad m) => Topic m a -> Topic m b -> Topic m a

Produce elements of the first Topic no faster than elements of the second Topic are produced.

concats :: (Monad m, Foldable f) => Topic m (f a) -> Topic m a

Flatten a Topic of Foldable values. For example, turn a Topic m [a] of finite lists into a Topic a by taking each element from each list in sequence.

interruptible :: Foldable t => Topic IO (t a) -> Topic IO a

Flatten a Topic of Foldable values such that old values are discarded as soon as the original Topic produces a new Foldable.

forkTopic :: Topic IO a -> IO (Topic IO a)

Pull elements from a Topic in a new thread. This allows IO Topics to run at different rates even if they are consumed by a single thread.

slidingWindow :: (Monad m, Monoid a) => Int -> Topic m a -> Topic m a

Sliding window over a Monoid. slidingWindow n t slides a window of width n along Topic t. As soon as at least n elements have been produced by t, the output Topic starts producing the mconcat of the elements in the window.

slidingWindowG :: (Monad m, AdditiveGroup a) => Int -> Topic m a -> Topic m a

Sliding window over an AdditiveGroup. slidingWindowG n t slides a window of width n along Topic t. As soon as at least n elements have been produced by t, the output Topic starts producing the total sum of the elements of the window. This function is more efficient than slidingWindow because the group inverse operation is used to remove elements falling behind the window from the running sum.

topicOn :: (Applicative m, Monad m) => (a -> b) -> (a -> c -> d) -> m (b -> m c) -> Topic m a -> Topic m d

A way of pushing a monadic action into and along a Topic. The application topicOn proj inj trans t extracts a function from trans that is then applied to the result of applying proj to each value of Topic t. The result of that application is supplied to the result of applying inj to the same values from t to produce a value for the output Topic. A typical use case is projecting out a field from the original Topic t using proj so that it may be modified by trans and then injected back into the original structure using inj.

subsample :: Monad m => Int -> Topic m b -> Topic m b

subsample n t subsamples topic t by dropping n elements for every element produced by the result topic.