SciFlow 0.3.0 → 0.4.0
raw patch · 12 files changed
+301/−517 lines, 12 filesdep +containersdep +fgldep +graphvizdep −unordered-containers
Dependencies added: containers, fgl, graphviz
Dependencies removed: unordered-containers
Files
- SciFlow.cabal +8/−65
- src/Scientific/Workflow.hs +9/−27
- src/Scientific/Workflow/Builder.hs +152/−89
- src/Scientific/Workflow/Builder/TH.hs +0/−84
- src/Scientific/Workflow/DB.hs +40/−0
- src/Scientific/Workflow/Main.hs +0/−41
- src/Scientific/Workflow/Serialization.hs +0/−7
- src/Scientific/Workflow/Serialization/Show.hs +0/−11
- src/Scientific/Workflow/Serialization/Yaml.hs +0/−14
- src/Scientific/Workflow/Types.hs +64/−166
- src/Scientific/Workflow/Utils.hs +0/−13
- src/Scientific/Workflow/Visualize.hs +28/−0
SciFlow.cabal view
@@ -2,67 +2,11 @@ -- documentation, see http://haskell.org/cabal/users-guide/ name: SciFlow-version: 0.3.0+version: 0.4.0 synopsis: Scientific workflow management system description: SciFlow is to help programmers design complex workflows with ease.- .- Feature includes:- .- 1. Use "labeled" arrows to connect individual steps- and cache computational results.- .- 2. Use monad and template haskell to automate the process- of building DAGs.- .- Here is a trivial example. Since we use template haskell,- we need to divide this small program into two files.- .- > -- File 1: MyModule.hs- >- > module MyModule where- >- > import Control.Arrow- > import Scientific.Workflow- >- > input :: Actor () Int- > input = arr $ const 10- >- > plus1 :: Actor Int Int- > plus1 = arr (+1)- >- > mul2 :: Actor Int Int- > mul2 = arr (*2)- >- > combine :: Actor (Int, Int) Int- > combine = arr $ \(a,b) -> a + b- >- > -- builder monad- > builder :: Builder ()- > builder = do- > node "id000" "input" "this is input"- > node "id001" "plus1" "add 1 to the input"- > node "id002" "mul2" "double the input"- > node "id003" "combine" "combine two input"- >- > "id000" ~> "id001"- > "id000" ~> "id002"- > link2 ("id001", "id002") "id003"- >- > --------------------------------------------- > -- File 2: main.hs- >- > import Scientific.Workflow- > import MyModule- > import Data.Default- >- > -- assemble workflow using template haskell- > $(mkWorkflow "myWorkflow" builder)- >- > main = do result <- runWorkflow myWorkflow def- > print result- . license: MIT license-file: LICENSE@@ -79,16 +23,13 @@ Default: False library+ ghc-options: -Wall 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+ Scientific.Workflow.DB+ Scientific.Workflow.Types+ Scientific.Workflow.Visualize if flag(debug) CPP-Options: -DDEBUG@@ -99,6 +40,8 @@ base >=4.0 && <5.0 , bytestring , data-default-class+ , fgl+ , graphviz , lens >=4.0 , mtl , optparse-applicative@@ -107,7 +50,7 @@ , th-lift , text , template-haskell- , unordered-containers >=0.2+ , containers , yaml hs-source-dirs: src
src/Scientific/Workflow.hs view
@@ -1,39 +1,21 @@ module Scientific.Workflow- ( module Scientific.Workflow.Types+ ( runWorkflow+ , getWorkflowState , module Scientific.Workflow.Builder- , module Scientific.Workflow.Builder.TH- , module Scientific.Workflow.Serialization.Yaml- , runWorkflow- , mapA+ , module Scientific.Workflow.Types ) where -import Control.Monad.State (foldM_, runStateT)-import qualified Data.Text as T-import Data.List (foldl')-import Shelly (fromText, mkdir_p,- shelly)+import Control.Monad.State import Scientific.Workflow.Builder-import Scientific.Workflow.Builder.TH-import Scientific.Workflow.Serialization.Yaml import Scientific.Workflow.Types-import Scientific.Workflow.Utils -runWorkflow :: [Workflow] -> RunOpt -> IO ()-runWorkflow wfs opt = do- 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-- let config = WorkflowConfig (_runDir opt) (_runLogDir opt) st+runWorkflow :: [Workflow] -> State RunOpt () -> IO ()+runWorkflow wfs setOpt = do+ config <- getWorkflowState $ _dbPath opt foldM_ f config wfs where- dir = _runDir opt ++ "/" ++ _runLogDir opt+ opt = execState setOpt defaultRunOpt f config (Workflow wf) = do- (_, config') <- runStateT (runProcessor wf ()) config+ (_, config') <- runStateT (wf ()) config return config'
src/Scientific/Workflow/Builder.hs view
@@ -1,121 +1,184 @@+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-}-module Scientific.Workflow.Builder where+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE CPP #-} -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+module Scientific.Workflow.Builder+ ( node+ , link+ , (~>)+ , path+ , Builder+ , buildWorkflow+ , buildWorkflowPart+ , getWorkflowState+ , mkDAG+ ) where -import Scientific.Workflow.Types+import Control.Lens ((^.), (%~), _1, _2, _3, at, (.=))+import Control.Monad.State+import qualified Data.Text as T+import Data.Graph.Inductive.Graph+ ( mkGraph+ , lab+ , labNodes+ , outdeg+ , lpre+ , labnfilter+ , gmap+ , suc+ , subgraph )+import Data.Graph.Inductive.PatriciaTree (Gr)+import Data.List (sortBy)+import qualified Data.Map as M+import Data.Maybe (fromJust)+import Data.Ord (comparing)+import qualified Data.Map as M --- | 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- | L3 (String,String,String) String- | L4 (String,String,String,String) String- | L5 (String,String,String,String,String) String- | L6 (String,String,String,String,String,String) String+import Language.Haskell.TH+import qualified Language.Haskell.TH.Lift as T --- | State of Builder Monad, storing workflow structure-data B = B- { _nodes :: [(String, ExpQ, T.Text)]- , _links :: [(String, Factor)]- }+import Scientific.Workflow.Types+import Scientific.Workflow.DB -type Builder = State B+import Debug.Trace (traceM) --- | Objects that can be converted to ExpQ-class ToExpQ a where- toExpQ :: a -> ExpQ+instance T.Lift T.Text where+ lift t = [| T.pack $(T.lift $ T.unpack t) |] -instance ToExpQ Name where- toExpQ = varE -instance ToExpQ ExpQ where- toExpQ = id+type EdgeOrd = Int+type Node = (PID, (ExpQ, Attribute))+type Edge = (PID, PID, EdgeOrd) +type Builder = State ([Node], [Edge])+ -- | Declare a computational node-node :: ToExpQ a => String -> a -> T.Text -> Builder ()-node l f anno = modify $ \s -> s{_nodes = (l, toExpQ f, anno) : _nodes s}+node :: ToExpQ q => PID -> q -> State Attribute () -> Builder ()+node p fn setAttr = modify $ _1 %~ (newNode:)+ where+ attr = execState setAttr defaultAttribute+ newNode = (p, (toExpQ fn, attr))+{-# INLINE node #-} -- | many-to-one generalized link function-link :: [String] -> String -> Builder ()-link [] t = singleton t-link [a] t = link1 a t-link [a,b] t = link2 (a,b) t-link [a,b,c] t = link3 (a,b,c) t-link [a,b,c,d] t = link4 (a,b,c,d) t-link [a,b,c,d,e] t = link5 (a,b,c,d,e) t-link [a,b,c,d,e,f] t = link6 (a,b,c,d,e,f) t-link _ _ = error "I can't have so many links, yet!"+link :: [PID] -> PID -> Builder ()+link xs t = modify $ _2 %~ (zip3 xs (repeat t) [0..] ++)+{-# INLINE link #-} -- | (~>) = link.-(~>) :: [String] -> String -> Builder ()+(~>) :: [PID] -> PID -> Builder () (~>) = link+{-# INLINE (~>) #-} -- | singleton-singleton :: String -> Builder ()-singleton t = modify $ \s -> s{_links = (t, S t) : _links s}---- | Declare a path.-path :: [String] -> Builder ()+path :: [PID] -> Builder () path ns = foldM_ f (head ns) $ tail ns where- f a t = link1 a t >> return t+ f a t = link [a] t >> return t+{-# INLINE path #-} --- | one-to-one link-link1 :: String -> String -> Builder ()-link1 a t = modify $ \s -> s{_links = (t, L a t) : _links s}+buildWorkflow :: String+ -> Builder ()+ -> Q [Dec]+buildWorkflow wfName b = mkWorkflow wfName $ mkDAG b --- | two-to-one link-link2 :: (String, String) -> String -> Builder ()-link2 (a,b) t = modify $ \s -> s{_links = (t, L2 (a,b) t) : _links s}+buildWorkflowPart :: State RunOpt ()+ -> String+ -> Builder ()+ -> Q [Dec]+buildWorkflowPart setOpt wfName b = do+ st <- runIO $ getWorkflowState $ opt^.dbPath+ mkWorkflow wfName $ trimDAG st $ mkDAG b+ where+ opt = execState setOpt defaultRunOpt --- | tree-to-one link-link3 :: (String, String, String) -> String -> Builder ()-link3 (a,b,c) t = modify $ \s -> s{_links = (t, L3 (a,b,c) t) : _links s}+getWorkflowState :: FilePath -> IO WorkflowState+getWorkflowState dir = do+ db <- openDB dir+ ks <- getKeys db+ pSt <- mapM (flip isFinished db) ks+ let pSts = M.fromList $ zipWith (\k s ->+ if s then (k, Finished) else (k, Scheduled)) ks pSt+ return $ WorkflowState db pSts+{-# INLINE getWorkflowState #-} -link4 :: (String, String, String, String) -> String -> Builder ()-link4 (a,b,c,d) t = modify $ \s -> s{_links = (t, L4 (a,b,c,d) t) : _links s}+-- | Objects that can be converted to ExpQ+class ToExpQ a where+ toExpQ :: a -> ExpQ -link5 :: (String, String, String, String, String) -> String -> Builder ()-link5 (a,b,c,d,e) t = modify $ \s -> s{_links = (t, L5 (a,b,c,d,e) t) : _links s}+instance ToExpQ Name where+ toExpQ = varE -link6 :: (String, String, String, String, String, String) -> String -> Builder ()-link6 (a,b,c,d,e,f) t = modify $ \s -> s{_links = (t, L6 (a,b,c,d,e,f) t) : _links s}+instance ToExpQ ExpQ where+ toExpQ = id +type DAG = Gr Node EdgeOrd -data Graph = Graph- { _children :: M.HashMap String [String]- , _parents :: M.HashMap String [String]- , _vertice :: [ID]- }+mkDAG :: Builder () -> DAG+mkDAG b = mkGraph ns' es'+ where+ ns' = map (\x -> (pid2nid $ fst x, x)) ns+ es' = map (\(fr, t, o) -> (pid2nid fr, pid2nid t, o)) es+ (ns, es) = execState b ([], [])+ pid2nid p = M.findWithDefault (error "mkDAG") p m+ where+ m = M.fromListWithKey err $ zip (map fst ns) [0..]+ err k _ _ = error $ "multiple instances for: " ++ T.unpack k+{-# INLINE mkDAG #-} -children :: String -> Graph -> [String]-children x = M.lookupDefault [] x . _children+trimDAG :: WorkflowState -> DAG -> DAG+trimDAG st dag = gmap revise gr+ where+ revise c | done (c^._3._1) && null (c^._1) = _3._2._1 %~ e $ c+ | otherwise = c+ where e x = [| const undefined >=> $(x) |]+ gr = labnfilter f dag+ where+ f (i, (x,_)) = (not . done) x || any (not . done) children+ where children = map (fst . fromJust . lab dag) $ suc dag i+ done x = getStatus x == Finished+ getStatus x = M.findWithDefault Scheduled x $ st^.procStatus+{-# INLINE trimDAG #-} -parents :: String -> Graph -> [String]-parents x = M.lookupDefault [] x . _parents+mkWorkflow :: String -> DAG -> Q [Dec]+mkWorkflow wfName dag = do+ decNode <- concat <$> mapM (f . snd) ns+ decWf <- [d| $(varP $ mkName wfName) = $(fmap ListE $ mapM linking leafNodes)+ |]+ return $ decNode ++ decWf+ where+ f (p, (fn, _)) = [d| $(varP $ mkName $ T.unpack p) = mkProc p $(fn) |]+ ns = labNodes dag+ leafNodes = filter ((==0) . outdeg dag . fst) ns+ linking nd = [| Workflow $(go nd) |]+ where+ go n = connect inNodes n+ where+ inNodes = map (\(x,_) -> (x, fromJust $ lab dag x)) $+ sortBy (comparing snd) $ lpre dag $ fst n+ define n = varE $ mkName (T.unpack $ (snd n) ^. _1)+ connect [] t = define t+ connect [s1] t = [| $(go s1) >=> $(define t) |]+ connect xs t = [| $(foldl g e0 $ tail xs) >=> $(define t) |]+ where+ e0 = [| (fmap.fmap) $(conE (tupleDataName $ length xs)) $(go $ head xs) |]+ g acc x = [| ((<*>) . fmap (<*>)) $(acc) $(go x) |]+{-# INLINE mkWorkflow #-} -leaves :: Graph -> [String]-leaves g = filter (\x -> null $ children x g) $ _vertice g+mkProc :: Serializable b => PID -> (a -> IO b) -> (Processor a b)+mkProc p f = \input -> do+ st <- get+ case M.findWithDefault Scheduled p (st^.procStatus) of+ Finished -> lift $ readData p $ st^.db+ Scheduled -> do+#ifdef DEBUG+ traceM $ "Running node: " ++ T.unpack p+#endif+ result <- lift $ f input+ lift $ saveData p result $ st^.db -fromFactors :: [Factor] -> Graph-fromFactors us = Graph cs ps vs'- where- cs = M.fromListWith (++) $ map (second return) es'- ps = M.fromListWith (++) $ map (second return . swap) es'- vs' = concat vs- es' = concat es- (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)])+ (procStatus . at p) .= Just Finished+ return result+{-# INLINE mkProc #-}
− src/Scientific/Workflow/Builder/TH.hs
@@ -1,84 +0,0 @@-{-# LANGUAGE TemplateHaskell #-}-{-# LANGUAGE OverloadedStrings #-}-module Scientific.Workflow.Builder.TH where--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 -- ^ 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- 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--defineNodes :: [(String, ExpQ)] -> Q [Dec]-defineNodes nodes = fmap concat $ mapM f nodes- where- f (l, fn) = [d| $(varP $ mkName l) = proc l $(fn) |]-{-# INLINE defineNodes #-}---- | Start linking processors from a given node-linkFrom :: M.HashMap String Factor- -> Factor- -> Q Exp-linkFrom table nd = [| Workflow $(go nd) |]- where- expand x = go $ M.lookupDefault (S x) x table-- go (S a) = varE $ mkName a- 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/DB.hs view
@@ -0,0 +1,40 @@+{-# LANGUAGE OverloadedStrings #-}+module Scientific.Workflow.DB+ ( openDB+ , readData+ , saveData+ , isFinished+ , getKeys+ ) where++import Scientific.Workflow.Types+import Shelly (fromText, lsT, shelly, test_f, mkdir_p)+import qualified Data.ByteString as B+import qualified Data.Text as T++openDB :: FilePath -> IO WorkflowDB+openDB dir = do+ shelly $ mkdir_p $ fromText $ T.pack dir+ return $ WorkflowDB dir+{-# INLINE openDB #-}++readData :: Serializable r => PID -> WorkflowDB -> IO r+readData p (WorkflowDB dir) = deserialize <$> B.readFile fl+ where fl = dir ++ "/" ++ T.unpack p+{-# INLINE readData #-}++saveData :: Serializable r => PID -> r -> WorkflowDB -> IO ()+saveData p result (WorkflowDB dir) = B.writeFile fl $ serialize result+ where fl = dir ++ "/" ++ T.unpack p+{-# INLINE saveData #-}++isFinished :: PID -> WorkflowDB -> IO Bool+isFinished p (WorkflowDB dir) = shelly $ test_f $ fromText $ T.pack fl+ where fl = dir ++ "/" ++ T.unpack p+{-# INLINE isFinished #-}++getKeys :: WorkflowDB -> IO [PID]+getKeys (WorkflowDB dir) = f <$> shelly (lsT $ fromText $ T.pack dir)+ where+ f = map (snd . T.breakOnEnd "/")+{-# INLINE getKeys #-}
− src/Scientific/Workflow/Main.hs
@@ -1,41 +0,0 @@-{-# 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
@@ -1,7 +0,0 @@-module Scientific.Workflow.Serialization where--import qualified Data.ByteString as B--class Serializable a where- serialize :: a -> B.ByteString- deserialize :: B.ByteString -> a
− src/Scientific/Workflow/Serialization/Show.hs
@@ -1,11 +0,0 @@-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE UndecidableInstances #-}-module Scientific.Workflow.Serialization.Show where--import qualified Data.ByteString.Char8 as B--import Scientific.Workflow.Serialization--instance (Read a, Show a) => Serializable a where- serialize = B.pack . show- deserialize = read . B.unpack
− src/Scientific/Workflow/Serialization/Yaml.hs
@@ -1,14 +0,0 @@-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE UndecidableInstances #-}-module Scientific.Workflow.Serialization.Yaml- (Serializable(..)- ) 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 = fromJust . decode
src/Scientific/Workflow/Types.hs view
@@ -1,189 +1,87 @@-{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE UndecidableInstances #-} {-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE GADTs #-} {-# LANGUAGE TemplateHaskell #-}-{-# LANGUAGE CPP #-}-module Scientific.Workflow.Types where -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)+module Scientific.Workflow.Types+ ( WorkflowDB(..)+ , Workflow(..)+ , PID+ , ProcState(..)+ , WorkflowState(..)+ , db+ , procStatus+ , Processor+ , RunOpt(..)+ , defaultRunOpt+ , dbPath+ , Serializable(..)+ , Attribute+ , defaultAttribute+ , label+ , note+ , def+ ) where -import Scientific.Workflow.Serialization (Serializable (..))+import Control.Lens (makeLenses)+import Control.Monad.State+import qualified Data.ByteString as B+import qualified Data.Map as M+import qualified Data.Text as T+import Data.Maybe (fromJust)+import Data.Yaml (FromJSON, ToJSON, encode, decode) -import Debug.Trace+class Serializable a where+ serialize :: a -> B.ByteString+ deserialize :: B.ByteString -> a ------------------------------------------------------------------------------------ Workflow---------------------------------------------------------------------------------+instance (FromJSON a, ToJSON a) => Serializable a where+ serialize = encode+ deserialize = fromJust . decode -type ID = String+data WorkflowDB = WorkflowDB FilePath -data NodeState = Finished- | Unfinished- | Skip- deriving (Show)+type PID = T.Text -type NodesDB = M.HashMap ID NodeState+data ProcState = Finished+ | Scheduled+ deriving (Eq) -data WorkflowConfig = WorkflowConfig- { _baseDir :: !FilePath- , _logDir :: !FilePath- , _nodeStatus :: !NodesDB+data WorkflowState = WorkflowState+ { _db :: WorkflowDB+ , _procStatus :: M.Map PID ProcState } -makeLenses ''WorkflowConfig--readNodeStatus :: ID -> NodesDB -> NodeState-readNodeStatus = M.lookupDefault Unfinished-{-# INLINE readNodeStatus #-}--writeNodeStatus :: ID -> NodeState -> NodesDB -> NodesDB-writeNodeStatus = M.insert-{-# INLINE writeNodeStatus #-}+makeLenses ''WorkflowState -data Mode = All- | Select [ID]+type Processor a b = a -> StateT WorkflowState IO b -L.deriveLift ''Mode+data Workflow where+ Workflow :: (Processor () o) -> Workflow data RunOpt = RunOpt- { _runDir :: !FilePath- , _runLogDir :: !FilePath- , _runMode :: !Mode- , _runForce :: !Bool+ { _dbPath :: FilePath } -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 }--instance Monad m => C.Category (Processor m) where- id = Processor return- (Processor f) . (Processor g) = Processor $ g >=> f--instance Monad m => Arrow (Processor m) where- arr f = Processor (return . f)- first (Processor f) = Processor (\ ~(b,d) -> f b >>= \c -> return (c,d))- second (Processor f) = Processor (\ ~(d,b) -> f b >>= \c -> return (d,c))---- | Label is a pair of side effects-type Label m l o = (l -> m (Maybe o), l -> o -> m ())---- | Turn a Kleisli arrow into a labeled arrow-label :: (MonadTrans t, Monad m, Monad (t m))- => Label (t m) l b- -> l- -> Kleisli m a b- -> Processor (t m) a b-label (pre, suc) l (Kleisli f) = Processor $ \x -> do- d <- pre l- case d of- Nothing -> do- o <- lift $ f x- suc l o- return o- Just v -> return v-{-# INLINE label #-}---class Arrow a => Actor a b c where- arrIO :: a b c -> Kleisli IO b c--instance Actor (->) a b where- arrIO = arr--instance Actor (Kleisli IO) a b where- arrIO = id--proc :: Actor ar a b => Serializable b => ID -> ar a b -> IOProcessor a b-proc l ar = label (recover, save) l $ arrIO ar--source :: Serializable o => ID -> o -> Source o-source l x = proc l $ const x--nullSource :: Source o-nullSource = label (const $ return $ Just undefined, undefined) ("" :: String) $ arr $ const undefined--recover :: Serializable a => ID -> StateT WorkflowConfig IO (Maybe a)-recover l = do- st <- readNodeStatus l <$> use nodeStatus--#ifdef DEBUG- traceM $ "Process node: " ++ l ++ " . Status is: " ++ show st-#endif-- 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 #-}--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+makeLenses ''RunOpt -#ifdef DEBUG- traceM $ "Finish node: " ++ l ++ "\n"-#endif-{-# INLINE save #-}+defaultRunOpt :: RunOpt+defaultRunOpt = RunOpt+ { _dbPath = "wfDB" } -type IOProcessor = Processor (StateT WorkflowConfig IO)+data Attribute = Attribute+ { _label :: T.Text -- ^ short description+ , _note :: T.Text -- ^ long description+ } --- | Source produce an output without inputs-type Source = IOProcessor ()+defaultAttribute :: Attribute+defaultAttribute = Attribute+ { _label = ""+ , _note = ""+ } -instance Functor Source where- fmap f (Processor g) = Processor $ fmap f . g+makeLenses ''Attribute -instance Applicative Source where- pure = Processor . const . return- Processor f <*> Processor g = Processor $ \x -> do- a <- f x- b <- g x- return $ a b+def :: State a ()+def = return ()
− src/Scientific/Workflow/Utils.hs
@@ -1,13 +0,0 @@-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 #-}
+ src/Scientific/Workflow/Visualize.hs view
@@ -0,0 +1,28 @@+{-# LANGUAGE OverloadedStrings #-}+module Scientific.Workflow.Visualize+ ( renderBuilder+ ) where++import Control.Lens+import Scientific.Workflow.Types+import Shelly (fromText, lsT, shelly, test_f, mkdir_p)+import qualified Data.ByteString as B+import qualified Data.Text as T+import qualified Data.Text.Lazy as TL++import Data.GraphViz+import Data.GraphViz.Printing+import Data.GraphViz.Attributes.Complete++import Scientific.Workflow.Types (note)+import Scientific.Workflow.Builder++renderBuilder :: Builder () -> TL.Text+renderBuilder b = renderDot . toDot $ graphToDot param dag+ where+ fmtnode (_, (p, (_, attr))) = [Label $ StrLabel $ TL.fromStrict lab]+ where+ lab | T.null (attr^.label) = p+ | otherwise = attr^.label+ dag = mkDAG b+ param = nonClusteredParams{fmtNode = fmtnode}