kafka-device 0.1.5.0 → 0.2.1.0
raw patch · 6 files changed
+242/−46 lines, 6 filesdep +linear
Dependencies added: linear
Files
- kafka-device.cabal +9/−5
- src/Main.hs +4/−7
- src/MainKeyboard.hs +4/−6
- src/Network/UI/Kafka.hs +40/−18
- src/Network/UI/Kafka/Interpretation.hs +179/−0
- src/Network/UI/Kafka/Keyboard.hs +6/−10
kafka-device.cabal view
@@ -1,12 +1,12 @@ name: kafka-device-version: 0.1.5.0+version: 0.2.1.0 synopsis: UI device events via a Kafka message broker description: This package contains functions for passing UI device events to topics on a Kafka message broker \<<https://kafka.apache.org/>\>. Also see \<<https://hackage.haskell.org/package/kafka-device-joystick/>\>, \<<https://hackage.haskell.org/package/kafka-device-glut/>\>, \<<https://hackage.haskell.org/package/kafka-device-spacenav/>\>, \<<https://hackage.haskell.org/package/kafka-device-leap/>\>. license: MIT license-file: LICENSE author: Brian W Bush <consult@brianwbush.info> maintainer: Brian W Bush <consult@brianwbush.info>-copyright: (c) 2016 Brian W Bush+copyright: (c) 2016-17 Brian W Bush category: Hardware build-type: Simple cabal-version: >= 1.10@@ -23,6 +23,7 @@ library exposed-modules: Network.UI.Kafka+ Network.UI.Kafka.Interpretation Network.UI.Kafka.Keyboard Network.UI.Kafka.Types build-depends: base >= 4.8 && < 5@@ -30,10 +31,11 @@ , binary , bytestring , cereal+ , linear , milena , mtl hs-source-dirs: src- ghc-options: -Wall+ ghc-options: -O2 -Wall default-language: Haskell2010 executable kafka-device@@ -43,10 +45,11 @@ , binary , bytestring , cereal+ , linear , milena , mtl hs-source-dirs: src- ghc-options: -Wall -threaded+ ghc-options: -O2 -Wall -threaded default-language: Haskell2010 executable kafka-device-keyboard@@ -56,8 +59,9 @@ , binary , bytestring , cereal+ , linear , milena , mtl hs-source-dirs: src- ghc-options: -Wall -threaded+ ghc-options: -O2 -Wall -threaded default-language: Haskell2010
src/Main.hs view
@@ -1,6 +1,6 @@ {-|-Module : Main-Copyright : (c) 2016 Brian W Bush+Module : $Header$+Copyright : (c) 2016-17 Brian W Bush License : MIT Maintainer : Brian W Bush <consult@brianwbush.info> Stability : Experimental@@ -16,8 +16,7 @@ ) where -import Data.String (IsString(fromString))-import Network.UI.Kafka (consumerLoop)+import Network.UI.Kafka (TopicConnection(TopicConnection), consumerLoop) import System.Environment (getArgs) @@ -34,9 +33,7 @@ putStrLn $ "Kafka topic: " ++ topic (_, loop) <- consumerLoop- (fromString client)- (fromString host, toEnum $ read port)- (fromString topic)+ (TopicConnection client (host, read port) topic) $ curry print result <- loop either print return result
src/MainKeyboard.hs view
@@ -1,6 +1,6 @@ {-|-Module : Main-Copyright : (c) 2016 Brian W Bush+Module : $Header$+Copyright : (c) 2016-17 Brian W Bush License : MIT Maintainer : Brian W Bush <consult@brianwbush.info> Stability : Experimental@@ -16,7 +16,7 @@ ) where -import Data.String (IsString(fromString))+import Network.UI.Kafka (TopicConnection(TopicConnection)) import Network.UI.Kafka.Keyboard (keyboardLoop) import System.Environment (getArgs) @@ -35,9 +35,7 @@ putStrLn $ "Sensor name: " ++ sensor (_, loop) <- keyboardLoop- (fromString client)- (fromString host, toEnum $ read port)- (fromString topic)+ (TopicConnection client (host, read port) topic) sensor result <- loop either print return result
src/Network/UI/Kafka.hs view
@@ -1,6 +1,6 @@ {-|-Module : Network.UI.Kafka-Copyright : (c) 2016 Brian W Bush+Module : $Header$+Copyright : (c) 2016-17 Brian W Bush License : MIT Maintainer : Brian W Bush <consult@brianwbush.info> Stability : Experimental@@ -10,9 +10,14 @@ -} +{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE RecordWildCards #-}++ module Network.UI.Kafka ( -- * Types- Sensor+ TopicConnection(..)+, Sensor , LoopAction , ExitAction -- * Consumption@@ -24,19 +29,38 @@ ) where +import Control.Arrow ((***)) import Control.Concurrent (MVar, newEmptyMVar, isEmptyMVar, threadDelay, tryPutMVar) import Control.Monad (void, when) import Control.Monad.Except (liftIO)+import Data.Aeson.Types (FromJSON, ToJSON) import Data.Binary (decode, encode) import Data.ByteString.Char8 (pack, unpack) import Data.ByteString.Lazy (fromStrict, toStrict)+import Data.String (IsString(fromString))+import GHC.Generics (Generic) import Network.UI.Kafka.Types (Event)-import Network.Kafka (KafkaAddress, KafkaClientError, KafkaClientId, KafkaTime(..), TopicAndMessage(..), getLastOffset, mkKafkaState, runKafka, withAddressHandle)+import Network.Kafka (KafkaClientError, KafkaTime(..), TopicAndMessage(..), getLastOffset, mkKafkaState, runKafka, withAddressHandle) import Network.Kafka.Consumer (fetch', fetchMessages, fetchRequest) import Network.Kafka.Producer (makeKeyedMessage, produceMessages)-import Network.Kafka.Protocol (FetchResponse(..), KafkaBytes(..), Key(..), Message(..), TopicName, Value(..))+import Network.Kafka.Protocol (FetchResponse(..), KafkaBytes(..), Key(..), Message(..), Value(..)) +-- | Connection information for a Kafka topic.+data TopicConnection =+ TopicConnection+ {+ client :: String -- ^ A name for the Kafka client.+ , address :: (String, Int) -- ^ The host name and port for the Kafka broker.+ , topic :: String -- ^ The Kafka topic name.+ }+ deriving (Eq, Generic, Read, Show)++instance FromJSON TopicConnection++instance ToJSON TopicConnection++ -- | A name for a sensor. type Sensor = String @@ -56,15 +80,15 @@ -- | Consume events for a Kafka topic.-consumerLoop :: KafkaClientId -- ^ A Kafka client identifier for the consumer.- -> KafkaAddress -- ^ The address of the Kafka broker.- -> TopicName -- ^ The Kafka topic name.+consumerLoop :: TopicConnection -- ^ The Kafka topic name and connection information. -> ConsumerCallback -- ^ The consumer callback. -> IO (ExitAction, LoopAction) -- ^ Action to create the exit and loop actions.-consumerLoop clientId address topic consumer =+consumerLoop TopicConnection{..} consumer = do exitFlag <- newEmptyMVar :: IO (MVar ()) let+ topic' = fromString topic+ address' = fromString *** fromIntegral $ address fromMessage :: Message -> (Sensor, Event) fromMessage message = let@@ -73,7 +97,7 @@ (unpack k, decode $ fromStrict v) loop offset = do- result <- withAddressHandle address $ \handle -> fetch' handle =<< fetchRequest offset 0 topic+ result <- withAddressHandle address' $ \handle -> fetch' handle =<< fetchRequest offset 0 topic' let (_, [(_, _, offset', _)]) : _ = _fetchResponseFields result messages = fromMessage . _tamMessage <$> fetchMessages result@@ -87,9 +111,9 @@ return ( void $ tryPutMVar exitFlag ()- , fmap void $ runKafka (mkKafkaState clientId address)+ , fmap void $ runKafka (mkKafkaState (fromString client) address') $ do- offset <- getLastOffset LatestTime 0 topic+ offset <- getLastOffset LatestTime 0 topic' loop offset ) @@ -99,13 +123,11 @@ -- | Produce events for a Kafka topic.-producerLoop :: KafkaClientId -- ^ A Kafka client identifier for the producer.- -> KafkaAddress -- ^ The address of the Kafka broker.- -> TopicName -- ^ The Kafka topic name.+producerLoop :: TopicConnection -- ^ The Kafka topic name and connection information. -> Sensor -- ^ The name of the sensor producing the event. -> ProducerCallback -- ^ The producer callback. -> IO (ExitAction, LoopAction) -- ^ Action to create the exit and loop actions.-producerLoop clientId address topic sensor producer =+producerLoop TopicConnection{..} sensor producer = do exitFlag <- newEmptyMVar :: IO (MVar ()) let@@ -115,7 +137,7 @@ void $ produceMessages $ map- (TopicAndMessage topic . makeKeyedMessage (pack sensor) . toStrict . encode)+ (TopicAndMessage (fromString topic) . makeKeyedMessage (pack sensor) . toStrict . encode) events running <- liftIO $ isEmptyMVar exitFlag when running@@ -123,5 +145,5 @@ return ( void $ tryPutMVar exitFlag ()- , void <$> runKafka (mkKafkaState clientId address) loop+ , void <$> runKafka (mkKafkaState (fromString client) (fromString *** fromIntegral $ address)) loop )
+ src/Network/UI/Kafka/Interpretation.hs view
@@ -0,0 +1,179 @@+{-|+Module : $Header$+Copyright : (c) 2016-17 Brian W Bush+License : MIT+Maintainer : Brian W Bush <consult@brianwbush.info>+Stability : Experimental+Portability : Stable++Interpret user-interfaces events on Kafka topics.+-}++++{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}++{-# OPTIONS_GHC -fno-warn-orphans #-}+++module Network.UI.Kafka.Interpretation (+-- * Types+ Interpretation(..)+, AxisInterpretation(..)+, AnalogHandler+, ButtonHandler+-- * Event handling+, interpretationLoop+) where+++import Control.Concurrent.MVar (MVar, newMVar, readMVar, swapMVar)+import Control.Monad (unless, void)+import Data.Aeson.Types (FromJSON, ToJSON)+import GHC.Generics (Generic)+import Linear.Conjugate (Conjugate)+import Linear.Epsilon (Epsilon)+import Linear.Quaternion (Quaternion(..), axisAngle, rotate)+import Linear.V3 (V3(..))+import Linear.Vector ((^+^), basis)+import Network.UI.Kafka (ExitAction, LoopAction, Sensor, TopicConnection, producerLoop)+import Network.UI.Kafka.Types (Button(..), Event(ButtonEvent, LocationEvent, OrientationEvent), Toggle(..))+++instance FromJSON a => FromJSON (V3 a)++instance ToJSON a => ToJSON (V3 a)+++instance FromJSON a => FromJSON (Quaternion a)++instance ToJSON a => ToJSON (Quaternion a)+++-- | Instructions for interpreting user-interface events from Kafka.+data Interpretation a =+ TrackInterpretation+ {+ kafka :: TopicConnection -- ^ How to connect to Kafka.+ , sensor :: Sensor -- ^ The name of the sensor.+ , path :: FilePath -- ^ The path the the device.+ , xAxis :: AxisInterpretation a -- ^ Interpretation for the x-axis.+ , yAxis :: AxisInterpretation a -- ^ Interpretation for the y-axis.+ , zAxis :: AxisInterpretation a -- ^ Interpretation for the z-axis.+ , phiAxis :: AxisInterpretation a -- ^ Interpretation for the Euler angle about the x-axis.+ , thetaAxis :: AxisInterpretation a -- ^ Interpretation for the Euler angle about the y-axis.+ , psiAxis :: AxisInterpretation a -- ^ Interpretation for the Euler angle about the z-axis.+ , location :: V3 a -- ^ The initial location, (x, y, z).+ , orientation :: V3 a -- ^ The initial orientation, (phi, theta, psi).+ , flying :: Bool -- ^ Whether to interpret displacements as being relative to the current orientation.+ , resetButton :: Maybe Int -- ^ The button number for resetting the location and orientation to their initial values.+ }+ deriving (Eq, Generic, Read, Show)++instance FromJSON a => FromJSON (Interpretation a)++instance ToJSON a => ToJSON (Interpretation a)+++-- | Instructions for interpreting an axis.+data AxisInterpretation a =+ AxisInterpretation+ {+ axisNumber :: Int -- ^ The axis number.+ , threshold :: Maybe a -- ^ The threshold for forwarding events.+ , increment :: a -- ^ How much to increment the axis per unit of change.+ , lowerBound :: Maybe a -- ^ The minimum value for the axis position.+ , upperBound :: Maybe a -- ^ The maximum value for the axis position.+ }+ deriving (Eq, Generic, Read, Show)++instance FromJSON a => FromJSON (AxisInterpretation a)++instance ToJSON a => ToJSON (AxisInterpretation a)+++-- | Translate a SpaceNavigator event on Linux into events for Kafka.+translate :: (Conjugate a, Epsilon a, Num a, Ord a, RealFloat a)+ => MVar (State a) -- ^ Reference to the current location and orientation.+ -> AnalogHandler a b -- ^ How to handle raw analog events.+ -> ButtonHandler a b -- ^ How to handle raw button events.+ -> Interpretation a -- ^ The interpretation.+ -> b -- ^ A raw event.+ -> IO [Event] -- ^ The corresponding events for Kafka.+translate state analogHandler buttonHandler TrackInterpretation{..} event =+ do+ (location0, orientation0) <- readMVar state+ let+ adjust number setting AxisInterpretation{..} =+ if number == axisNumber && maybe True (abs setting >) threshold+ then setting * increment+ else 0+ clamp AxisInterpretation{..} =+ maybe id ((maximum .) . (. return) . (:)) lowerBound+ . maybe id ((minimum .) . (. return) . (:)) upperBound+ (location1@(V3 x y z), orientation1@(Quaternion q0 (V3 qx qy qz))) =+ case (buttonHandler event, analogHandler event) of+ (Just (number, pressed), _) -> if pressed && Just number == resetButton+ then (location, eulerToQuaternion orientation)+ else (location0, orientation0)+ (_, Just (number, setting)) -> let+ euler = adjust number setting <$> V3 phiAxis thetaAxis psiAxis+ axes = V3 xAxis yAxis zAxis+ delta = adjust number setting <$> axes+ in+ (+ clamp+ <$> axes+ <*> location0+ ^+^ (if flying then (orientation0 `rotate`) else id) delta+ , eulerToQuaternion euler+ * orientation0+ )+ (_, _) -> (location0, orientation0)+ [x', y', z', q0', qx', qy', qz'] = realToFrac <$> [x, y, z, q0, qx, qy, qz]+ unless (location0 == location1 && orientation0 == orientation1)+ . void+ $ swapMVar state (location1, orientation1)+ return+ . (if location0 /= location1 then (LocationEvent ( x' , y' , z' ) :) else id)+ . (if orientation0 /= orientation1 then (OrientationEvent (q0', qx', qy', qz') :) else id)+ $ case buttonHandler event of+ Just (number, pressed) -> [ButtonEvent (IndexButton number, if pressed then Down else Up)]+ Nothing -> []++-- | Convert from Euler angles to a quaternion.+eulerToQuaternion :: (Epsilon a, Num a, RealFloat a) => V3 a -> Quaternion a+eulerToQuaternion (V3 phi theta psi) =+ let+ [ex, ey, ez] = basis+ in+ ez `axisAngle` psi * ey `axisAngle` theta * ex `axisAngle` phi+++-- | Location and orientation.+type State a = (V3 a, Quaternion a)+++-- | How to handle raw analog events.+type AnalogHandler a b = b -> Maybe (Int, a)+++-- | How to handle raw button events.+type ButtonHandler a b = b -> Maybe (Int, Bool)+++-- | Repeatedly interpret events.+interpretationLoop :: (Conjugate a, Epsilon a, Num a, Ord a, RealFloat a)+ => AnalogHandler a b -- ^ How to handle raw analog events.+ -> ButtonHandler a b -- ^ How to handle raw button events.+ -> Interpretation a -- ^ The interpretation.+ -> IO b -- ^ Action for getting the next raw event.+ -> IO (ExitAction, LoopAction) -- ^ Action to create the exit and loop actions.+interpretationLoop analogHandler buttonHandler interpretation@TrackInterpretation{..} action =+ do+ state <- newMVar (location, eulerToQuaternion orientation)+ producerLoop kafka sensor+ $ translate state analogHandler buttonHandler interpretation+ =<< action
src/Network/UI/Kafka/Keyboard.hs view
@@ -1,6 +1,6 @@ {-|-Module : Network.UI.Kafka.Keyboard-Copyright : (c) 2016 Brian W Bush+Module : $Header$+Copyright : (c) 2016-17 Brian W Bush License : MIT Maintainer : Brian W Bush <consult@brianwbush.info> Stability : Experimental@@ -16,24 +16,20 @@ ) where -import Network.Kafka (KafkaAddress, KafkaClientId)-import Network.Kafka.Protocol (TopicName)-import Network.UI.Kafka (ExitAction, LoopAction, Sensor, producerLoop)+import Network.UI.Kafka (ExitAction, LoopAction, Sensor, TopicConnection, producerLoop) import Network.UI.Kafka.Types (Event(KeyEvent)) import System.IO (BufferMode(NoBuffering), hSetBuffering, hSetEcho, stdin) -- | Produce keyboard events on a Kafka topic from standard input.-keyboardLoop :: KafkaClientId -- ^ A Kafka client identifier for the producer.- -> KafkaAddress -- ^ The address of the Kafka broker.- -> TopicName -- ^ The Kafka topic name.+keyboardLoop :: TopicConnection -- ^ The Kafka topic name and connection information. -> Sensor -- ^ The name of the sensor producing events. -> IO (ExitAction, LoopAction) -- ^ Action to create the exit and loop actions.-keyboardLoop client address topic sensor =+keyboardLoop topicConnection sensor = do hSetBuffering stdin NoBuffering hSetEcho stdin False- producerLoop client address topic sensor+ producerLoop topicConnection sensor $ fmap (: []) $ KeyEvent <$> getChar