SciFlow 0.2.0 → 0.3.0
raw patch · 10 files changed
+348/−211 lines, 10 filesdep +lensdep +optparse-applicativedep +split
Dependencies added: lens, optparse-applicative, split, th-lift
Files
- SciFlow.cabal +20/−8
- src/Scientific/Workflow.hs +26/−15
- src/Scientific/Workflow/Builder.hs +35/−19
- src/Scientific/Workflow/Builder/TH.hs +64/−48
- src/Scientific/Workflow/Main.hs +41/−0
- src/Scientific/Workflow/Serialization.hs +1/−1
- src/Scientific/Workflow/Serialization/Show.hs +4/−4
- src/Scientific/Workflow/Serialization/Yaml.hs +2/−1
- src/Scientific/Workflow/Types.hs +142/−115
- src/Scientific/Workflow/Utils.hs +13/−0
SciFlow.cabal view
@@ -1,14 +1,14 @@--- Initial SciFlow.cabal generated by cabal init. For further +-- Initial SciFlow.cabal generated by cabal init. For further -- documentation, see http://haskell.org/cabal/users-guide/ name: SciFlow-version: 0.2.0+version: 0.3.0 synopsis: Scientific workflow management system description: SciFlow is to help programmers design complex workflows with ease. .- Feature includes: + Feature includes: . 1. Use "labeled" arrows to connect individual steps and cache computational results.@@ -28,7 +28,7 @@ > > input :: Actor () Int > input = arr $ const 10- > + > > plus1 :: Actor Int Int > plus1 = arr (+1) >@@ -71,28 +71,40 @@ copyright: (c) 2015 Kai Zhang category: Control build-type: Simple--- extra-source-files: +-- extra-source-files: cabal-version: >=1.10 +Flag Debug+ Description: Enable debug support+ Default: False+ library exposed-modules: Scientific.Workflow+ Scientific.Workflow.Main Scientific.Workflow.Types Scientific.Workflow.Builder Scientific.Workflow.Builder.TH Scientific.Workflow.Serialization Scientific.Workflow.Serialization.Show Scientific.Workflow.Serialization.Yaml+ Scientific.Workflow.Utils --- other-modules: + if flag(debug)+ CPP-Options: -DDEBUG+ else+ CPP-Options: -DNDEBUG - -- other-extensions: - build-depends: + build-depends: base >=4.0 && <5.0 , bytestring , data-default-class+ , lens >=4.0 , mtl+ , optparse-applicative , shelly+ , split+ , th-lift , text , template-haskell , unordered-containers >=0.2
src/Scientific/Workflow.hs view
@@ -7,22 +7,33 @@ , mapA ) where -import Control.Arrow (Kleisli(..))-import Control.Monad.Reader (runReaderT, forM_)-import qualified Data.Text as T-import Shelly (mkdir_p, shelly, fromText)+import Control.Monad.State (foldM_, runStateT)+import qualified Data.Text as T+import Data.List (foldl')+import Shelly (fromText, mkdir_p,+ shelly) -import Scientific.Workflow.Builder-import Scientific.Workflow.Builder.TH-import Scientific.Workflow.Types-import Scientific.Workflow.Serialization.Yaml+import Scientific.Workflow.Builder+import Scientific.Workflow.Builder.TH+import Scientific.Workflow.Serialization.Yaml+import Scientific.Workflow.Types+import Scientific.Workflow.Utils -runWorkflow :: [Workflow] -> WorkflowOpt -> IO ()+runWorkflow :: [Workflow] -> RunOpt -> IO () runWorkflow wfs opt = do- shelly $ mkdir_p $ fromText $ T.pack $ _logDir opt- forM_ wfs $ \(Workflow wf) -> do- _ <- runReaderT (runProcessor wf ()) $ Config $ _logDir opt- return ()+ shelly $ mkdir_p $ fromText $ T.pack dir+ st <- do+ db <- mkNodesDB opt+ return $ if _runForce opt+ then case _runMode opt of+ Select xs -> foldl' (\d x -> writeNodeStatus x Unfinished d) db xs+ _ -> db+ else db -mapA :: Monad m => Kleisli m a b -> Kleisli m [a] [b]-mapA (Kleisli f) = Kleisli $ mapM f+ let config = WorkflowConfig (_runDir opt) (_runLogDir opt) st+ foldM_ f config wfs+ where+ dir = _runDir opt ++ "/" ++ _runLogDir opt+ f config (Workflow wf) = do+ (_, config') <- runStateT (runProcessor wf ()) config+ return config'
src/Scientific/Workflow/Builder.hs view
@@ -1,12 +1,17 @@ {-# LANGUAGE FlexibleInstances #-} module Scientific.Workflow.Builder where -import Control.Arrow (second)-import Control.Monad.State.Lazy (State, modify, foldM_)-import qualified Data.HashMap.Strict as M-import qualified Data.Text as T-import Data.Tuple (swap)+import Control.Arrow (second)+import Control.Monad.State.Lazy (State, foldM_, modify)+import qualified Data.HashMap.Strict as M+import qualified Data.Text as T+import Data.Tuple (swap)+import Language.Haskell.TH +import Scientific.Workflow.Types++-- | Factors are small subgraphs/units of workflows. Each factor is associated+-- with multiple inputs and a single output data Factor = S String | L String String | L2 (String,String) String@@ -15,16 +20,27 @@ | L5 (String,String,String,String,String) String | L6 (String,String,String,String,String,String) String +-- | State of Builder Monad, storing workflow structure data B = B- { _nodes :: [(String, String, T.Text)]+ { _nodes :: [(String, ExpQ, T.Text)] , _links :: [(String, Factor)] } type Builder = State B +-- | Objects that can be converted to ExpQ+class ToExpQ a where+ toExpQ :: a -> ExpQ++instance ToExpQ Name where+ toExpQ = varE++instance ToExpQ ExpQ where+ toExpQ = id+ -- | Declare a computational node-node :: String -> String -> T.Text -> Builder ()-node l f anno = modify $ \s -> s{_nodes = (l,f,anno) : _nodes s}+node :: ToExpQ a => String -> a -> T.Text -> Builder ()+node l f anno = modify $ \s -> s{_nodes = (l, toExpQ f, anno) : _nodes s} -- | many-to-one generalized link function link :: [String] -> String -> Builder ()@@ -45,7 +61,7 @@ singleton :: String -> Builder () singleton t = modify $ \s -> s{_links = (t, S t) : _links s} --- | Declare a path. +-- | Declare a path. path :: [String] -> Builder () path ns = foldM_ f (head ns) $ tail ns where@@ -75,8 +91,8 @@ data Graph = Graph { _children :: M.HashMap String [String]- , _parents :: M.HashMap String [String]- , _vertice :: [String]+ , _parents :: M.HashMap String [String]+ , _vertice :: [ID] } children :: String -> Graph -> [String]@@ -95,11 +111,11 @@ ps = M.fromListWith (++) $ map (second return . swap) es' vs' = concat vs es' = concat es- (vs,es) = unzip $ map f us- f (S a) = ([a], [])- f (L a t) = ([a,t], [(a,t)])- f (L2 (a,b) t) = ([a,b,t], [(a,t),(b,t)])- f (L3 (a,b,c) t) = ([a,b,c,t], [(a,t),(b,t),(c,t)])- f (L4 (a,b,c,d) t) = ([a,b,c,d,t], [(a,t),(b,t),(c,t),(d,t)])- f (L5 (a,b,c,d,e) t) = ([a,b,c,d,e,t], [(a,t),(b,t),(c,t),(d,t),(e,t)])- f (L6 (a,b,c,d,e,f) t) = ([a,b,c,d,e,f,t], [(a,t),(b,t),(c,t),(d,t),(e,t),(f,t)])+ (vs,es) = unzip $ map fn us+ fn (S a) = ([a], [])+ fn (L a t) = ([a,t], [(a,t)])+ fn (L2 (a,b) t) = ([a,b,t], [(a,t),(b,t)])+ fn (L3 (a,b,c) t) = ([a,b,c,t], [(a,t),(b,t),(c,t)])+ fn (L4 (a,b,c,d) t) = ([a,b,c,d,t], [(a,t),(b,t),(c,t),(d,t)])+ fn (L5 (a,b,c,d,e) t) = ([a,b,c,d,e,t], [(a,t),(b,t),(c,t),(d,t),(e,t)])+ fn (L6 (a,b,c,d,e,f) t) = ([a,b,c,d,e,f,t], [(a,t),(b,t),(c,t),(d,t),(e,t),(f,t)])
src/Scientific/Workflow/Builder/TH.hs view
@@ -4,65 +4,81 @@ import Language.Haskell.TH +import Control.Applicative ((<$>), (<*>)) import Control.Arrow ((>>>)) import Control.Monad.State+import Data.Default.Class import qualified Data.HashMap.Strict as M import Scientific.Workflow.Types+import Scientific.Workflow.Utils (fileExist) import Scientific.Workflow.Builder -mkWorkflow :: String -> Builder () -> Q [Dec]-mkWorkflow name st = do- nodeDec <- declareNodes nd- wfDec <- [d| $(varP $ mkName name) = $(fmap ListE $ mapM (`linkNodes` m) endNodes) |]++mkWorkflow :: String -- ^ the name of workflow+ -> Builder ()+ -> Q [Dec]+mkWorkflow name builder = do+-- st <- runIO $ readWorkflowState config $ fst $ unzip nd++ nodeDec <- defineNodes nd -- ^ define node functions++ -- construct workflow+ wfDec <- [d| $( varP $ mkName name ) = $( fmap ListE $ mapM (linkFrom table) leafNodes )+ |]+ return $ nodeDec ++ wfDec where- builder = execState st $ B [] []- endNodes = map (\x -> M.lookupDefault undefined x m) . leaves . fromFactors . snd . unzip . _links $ builder- m = M.fromList $ _links builder- nd = map (\(a,b,_) -> (a,b)) $ _nodes builder+ builderSt = execState builder $ B [] []+ leafNodes = map (flip (M.lookupDefault undefined) table) . leaves .+ fromFactors . snd . unzip . _links $ builderSt+ table = M.fromList $ _links builderSt+ nd = map (\(a,b,_) -> (a,b)) $ _nodes builderSt -declareNodes :: [(String, String)] -> Q [Dec]-declareNodes nodes = do d <- mapM f nodes- return $ concat d+defineNodes :: [(String, ExpQ)] -> Q [Dec]+defineNodes nodes = fmap concat $ mapM f nodes where- f (l, ar) = [d| $(varP $ mkName l) = proc l $(varE $ mkName ar) |]-{-# INLINE declareNodes #-}+ f (l, fn) = [d| $(varP $ mkName l) = proc l $(fn) |]+{-# INLINE defineNodes #-} -linkNodes :: Factor -> M.HashMap String Factor -> Q Exp-linkNodes nd m = [| Workflow $(go nd) |]+-- | Start linking processors from a given node+linkFrom :: M.HashMap String Factor+ -> Factor+ -> Q Exp+linkFrom table nd = [| Workflow $(go nd) |] where- lookup' x = M.lookupDefault (S x) x m+ expand x = go $ M.lookupDefault (S x) x table+ go (S a) = varE $ mkName a- go (L a t) = [| $(go $ lookup' a) >>> $(go $ S t) |]- go (L2 (a,b) t) =- [| zipS $(go $ lookup' a) - $(go $ lookup' b)- >>> $(go $ S t) |]- go (L3 (a,b,c) t) =- [| zipS3 $(go $ lookup' a)- $(go $ lookup' b)- $(go $ lookup' c)- >>> $(go $ S t) |]- go (L4 (a,b,c,d) t) =- [| zipS4 $(go $ lookup' a)- $(go $ lookup' b)- $(go $ lookup' c)- $(go $ lookup' d)- >>> $(go $ S t) |]- go (L5 (a,b,c,d,f) t) =- [| zipS5 $(go $ lookup' a)- $(go $ lookup' b)- $(go $ lookup' c)- $(go $ lookup' d)- $(go $ lookup' f)- >>> $(go $ S t) |]- go (L6 (a,b,c,d,f,e) t) =- [| zipS6 $(go $ lookup' a)- $(go $ lookup' b)- $(go $ lookup' c)- $(go $ lookup' d)- $(go $ lookup' f)- $(go $ lookup' e)- >>> $(go $ S t) |]-{-# INLINE linkNodes #-}+ go (L a t) = [| $(expand a) >>> $(go $ S t) |]+ go (L2 (a,b) t) = [| (,)+ <$> $(expand a)+ <*> $(expand b)+ >>> $(go $ S t) |]+ go (L3 (a,b,c) t) = [| (,,)+ <$> $(expand a)+ <*> $(expand b)+ <*> $(expand c)+ >>> $(go $ S t) |]+ go (L4 (a,b,c,d) t) = [| (,,,)+ <$> $(expand a)+ <*> $(expand b)+ <*> $(expand c)+ <*> $(expand d)+ >>> $(go $ S t) |]+ go (L5 (a,b,c,d,f) t) = [| (,,,,)+ <$> $(expand a)+ <*> $(expand b)+ <*> $(expand c)+ <*> $(expand d)+ <*> $(expand f)+ >>> $(go $ S t) |]+ go (L6 (a,b,c,d,f,e) t) = [| (,,,,,)+ <$> $(expand a)+ <*> $(expand b)+ <*> $(expand c)+ <*> $(expand d)+ <*> $(expand f)+ <*> $(expand e)+ >>> $(go $ S t) |]+{-# INLINE linkFrom #-}
+ src/Scientific/Workflow/Main.hs view
@@ -0,0 +1,41 @@+{-# LANGUAGE TemplateHaskell #-}+module Scientific.Workflow.Main where++import Options.Applicative+import Options.Applicative.Types+import Data.List.Split (splitOn)+import Language.Haskell.TH++import Scientific.Workflow++workflowOptions :: Parser RunOpt+workflowOptions = subparser $+ command "run" ( info (helper <*> runOptParser) $ fullDesc <> progDesc "run" )+ where+ runOptParser = RunOpt+ <$> strOption+ ( long "dir"+ <> value "./"+ <> short 'd' )+ <*> strOption+ ( long "log"+ <> value "wfCache"+ <> short 'l' )+ <*> option (Select . splitOn "," <$> readerAsk)+ ( long "nodes"+ <> value All )+ <*> switch+ ( long "force"+ <> short 'f' )++defaultMain :: Builder () -> Q [Dec]+defaultMain = defineWorkflow "main"++defineWorkflow :: String -> Builder () -> Q [Dec]+defineWorkflow name builder = do+ workflowDec <- mkWorkflow "workflow_main" builder+ mainDec <- [d| $(varP $ mkName name) = execParser+ (info (helper <*> workflowOptions) fullDesc) >>=+ runWorkflow $(varE $ mkName "workflow_main")+ |]+ return $ workflowDec ++ mainDec
src/Scientific/Workflow/Serialization.hs view
@@ -4,4 +4,4 @@ class Serializable a where serialize :: a -> B.ByteString- deserialize :: B.ByteString -> Maybe a+ deserialize :: B.ByteString -> a
src/Scientific/Workflow/Serialization/Show.hs view
@@ -1,11 +1,11 @@-{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE UndecidableInstances #-} module Scientific.Workflow.Serialization.Show where -import qualified Data.ByteString.Char8 as B+import qualified Data.ByteString.Char8 as B -import Scientific.Workflow.Serialization+import Scientific.Workflow.Serialization instance (Read a, Show a) => Serializable a where- serialize = B.pack . show + serialize = B.pack . show deserialize = read . B.unpack
src/Scientific/Workflow/Serialization/Yaml.hs view
@@ -5,9 +5,10 @@ ) where import Data.Yaml (FromJSON, ToJSON, encode, decode)+import Data.Maybe (fromJust) import Scientific.Workflow.Serialization instance (FromJSON a, ToJSON a) => Serializable a where serialize = encode- deserialize = decode+ deserialize = fromJust . decode
src/Scientific/Workflow/Types.hs view
@@ -1,19 +1,98 @@-{-# LANGUAGE GADTs #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE CPP #-} module Scientific.Workflow.Types where -import qualified Control.Category as C-import Control.Arrow (Kleisli(..), Arrow(..), first, second)-import Control.Monad.Reader (ReaderT, lift, reader, (>=>), MonadTrans)-import qualified Data.ByteString as B-import Data.Default.Class-import qualified Data.Text as T-import Shelly (shelly, test_f, fromText)+import Control.Applicative+import Control.Arrow (Arrow (..), Kleisli (..),+ first, second)+import qualified Control.Category as C+import Control.Lens (makeLenses, use, (%=))+import Control.Monad.State (MonadTrans, StateT, lift,+ (>=>))+import qualified Data.ByteString as B+import Data.Default.Class+import qualified Data.HashMap.Strict as M+import qualified Data.Text as T+import qualified Language.Haskell.TH.Lift as L+import Shelly (shelly, test_d, lsT, fromText) -import Scientific.Workflow.Serialization (Serializable(..))+import Scientific.Workflow.Serialization (Serializable (..)) +import Debug.Trace++--------------------------------------------------------------------------------+-- Workflow+--------------------------------------------------------------------------------++type ID = String++data NodeState = Finished+ | Unfinished+ | Skip+ deriving (Show)++type NodesDB = M.HashMap ID NodeState++data WorkflowConfig = WorkflowConfig+ { _baseDir :: !FilePath+ , _logDir :: !FilePath+ , _nodeStatus :: !NodesDB+ }++makeLenses ''WorkflowConfig++readNodeStatus :: ID -> NodesDB -> NodeState+readNodeStatus = M.lookupDefault Unfinished+{-# INLINE readNodeStatus #-}++writeNodeStatus :: ID -> NodeState -> NodesDB -> NodesDB+writeNodeStatus = M.insert+{-# INLINE writeNodeStatus #-}++data Mode = All+ | Select [ID]++L.deriveLift ''Mode++data RunOpt = RunOpt+ { _runDir :: !FilePath+ , _runLogDir :: !FilePath+ , _runMode :: !Mode+ , _runForce :: !Bool+ }++L.deriveLift ''RunOpt++instance Default RunOpt where+ def = RunOpt+ { _runDir = "./"+ , _runLogDir = "wfCache/"+ , _runMode = All+ , _runForce = False+ }++mkNodesDB :: RunOpt -> IO NodesDB+mkNodesDB opt = do+ fls <- shelly $ do+ e <- test_d $ fromText $ T.pack dir+ if e then lsT $ fromText $ T.pack dir+ else return []+ return $ M.fromList $+ zip (map (T.unpack . snd . T.breakOnEnd "/") fls) $ repeat Finished+ where+ dir = _runDir opt ++ "/" ++ _runLogDir opt ++ "/"+ data Workflow where Workflow :: IOProcessor () b -> Workflow +--------------------------------------------------------------------------------+-- Arrow+--------------------------------------------------------------------------------+ -- | labeled Arrow newtype Processor m a b = Processor { runProcessor :: a -> m b } @@ -37,126 +116,74 @@ -> Processor (t m) a b label (pre, suc) l (Kleisli f) = Processor $ \x -> do d <- pre l- v <- case d of- Nothing -> lift $ f x+ case d of+ Nothing -> do+ o <- lift $ f x+ suc l o+ return o Just v -> return v- suc l v- return v+{-# INLINE label #-} -type IOProcessor = Processor (ReaderT Config IO) -type Actor = Kleisli IO--actor :: (a -> IO b) -> Actor a b-actor = Kleisli+class Arrow a => Actor a b c where+ arrIO :: a b c -> Kleisli IO b c --- | Source produce an output without taking inputs-type Source i = IOProcessor () i+instance Actor (->) a b where+ arrIO = arr -proc :: Serializable b => String -> Kleisli IO a b -> IOProcessor a b-proc = label (recover, save)+instance Actor (Kleisli IO) a b where+ arrIO = id -source :: Serializable o => String -> o -> Source o-source l x = proc l $ arr $ const x+proc :: Actor ar a b => Serializable b => ID -> ar a b -> IOProcessor a b+proc l ar = label (recover, save) l $ arrIO ar -recover :: Serializable a => String -> ReaderT Config IO (Maybe a)-recover l = do- dir <- reader _baseDir- let file = dir ++ l- exist <- lift $ fileExist file- if exist- then do c <- lift $ B.readFile file- return $ deserialize c- else return Nothing+source :: Serializable o => ID -> o -> Source o+source l x = proc l $ const x -save :: Serializable a => String -> a -> ReaderT Config IO ()-save l x = do- dir <- reader _baseDir- lift $ B.writeFile (dir++l) $ serialize x+nullSource :: Source o+nullSource = label (const $ return $ Just undefined, undefined) ("" :: String) $ arr $ const undefined -fileExist :: FilePath -> IO Bool-fileExist x = shelly $ test_f $ fromText $ T.pack x+recover :: Serializable a => ID -> StateT WorkflowConfig IO (Maybe a)+recover l = do+ st <- readNodeStatus l <$> use nodeStatus --- | zip two sources-zipS :: Source a -> Source b -> Source (a,b)-zipS (Processor f) (Processor g) = Processor $ \_ -> do- a <- f ()- b <- g ()- return (a,b)+#ifdef DEBUG+ traceM $ "Process node: " ++ l ++ " . Status is: " ++ show st+#endif -zipS3 :: Source a -> Source b -> Source c -> Source (a,b,c)-zipS3 (Processor f) (Processor g) (Processor h) = Processor $ \_ -> do- a <- f ()- b <- g ()- c <- h ()- return (a,b,c)+ case st of+ Finished -> do+ dir1 <- use baseDir+ dir2 <- use logDir+ let file = dir1 ++ "/" ++ dir2 ++ "/" ++ l+ (Just . deserialize) <$> lift (B.readFile file)+ Unfinished -> return Nothing+ Skip -> return $ Just undefined+{-# INLINE recover #-} -zipS4 :: Source a- -> Source b- -> Source c- -> Source d- -> Source (a,b,c,d)-zipS4 (Processor f)- (Processor g)- (Processor h)- (Processor i)- = Processor $ \_ -> do- a <- f ()- b <- g ()- c <- h ()- d <- i ()- return (a,b,c,d)+save :: Serializable a => ID -> a -> StateT WorkflowConfig IO ()+save l x = do+ dir1 <- use baseDir+ dir2 <- use logDir+ lift $ B.writeFile (dir1 ++ "/" ++ dir2 ++ "/" ++ l) $ serialize x+ nodeStatus %= writeNodeStatus l Finished -zipS5 :: Source a- -> Source b- -> Source c- -> Source d- -> Source e- -> Source (a,b,c,d,e)-zipS5 (Processor f)- (Processor g)- (Processor h)- (Processor i)- (Processor j)- = Processor $ \_ -> do- a <- f ()- b <- g ()- c <- h ()- d <- i ()- e <- j ()- return (a,b,c,d,e)+#ifdef DEBUG+ traceM $ "Finish node: " ++ l ++ "\n"+#endif+{-# INLINE save #-} -zipS6 :: Source a- -> Source b- -> Source c- -> Source d- -> Source e- -> Source f- -> Source (a,b,c,d,e,f)-zipS6 (Processor f)- (Processor g)- (Processor h)- (Processor i)- (Processor j)- (Processor k)- = Processor $ \_ -> do- a <- f ()- b <- g ()- c <- h ()- d <- i ()- e <- j ()- f <- k ()- return (a,b,c,d,e,f)+type IOProcessor = Processor (StateT WorkflowConfig IO) -data Config = Config- { _baseDir :: !FilePath- }+-- | Source produce an output without inputs+type Source = IOProcessor () -data WorkflowOpt = WorkflowOpt- { _logDir :: !FilePath- }+instance Functor Source where+ fmap f (Processor g) = Processor $ fmap f . g -instance Default WorkflowOpt where- def = WorkflowOpt- { _logDir = "wfCache/"- }+instance Applicative Source where+ pure = Processor . const . return+ Processor f <*> Processor g = Processor $ \x -> do+ a <- f x+ b <- g x+ return $ a b
+ src/Scientific/Workflow/Utils.hs view
@@ -0,0 +1,13 @@+module Scientific.Workflow.Utils where++import Control.Arrow+import qualified Data.Text as T+import Shelly (fromText, shelly, test_f)++fileExist :: FilePath -> IO Bool+fileExist x = shelly $ test_f $ fromText $ T.pack x+{-# INLINE fileExist #-}++mapA :: Monad m => Kleisli m a b -> Kleisli m [a] [b]+mapA (Kleisli f) = Kleisli $ mapM f+{-# INLINE mapA #-}