pipeline

Documentation

Overview

pipeline is a package used by the bookpipeline command, which handles the core functionality, using channels heavily to coordinate jobs. Note that it is considered an "internal" package, not intended for external use, and no guarantee is made of the stability of any interfaces provided.

Index

• Constants
• func Analyse(conn Downloader) func(chan string, chan string, chan error, *log.Logger)
• func CheckImages(dir string) error
• func DetectQueueType(dir string, conn Queuer) string
• func DownloadAll(dir string, name string, conn DownloadLister) error
• func DownloadAnalyses(dir string, name string, conn Downloader) error
• func DownloadBestPages(dir string, name string, conn Downloader, pluspngs bool) error
• func DownloadPdfs(dir string, name string, conn Downloader) error
• func GetMailSettings() (mailSettings, error)
• func Ocr(training string, tesscmd string) func(chan string, chan string, chan error, *log.Logger)
• func OcrPage(msg bookpipeline.Qmsg, conn Pipeliner, ...) error
• func Preprocess(thresholds []float64) func(chan string, chan string, chan error, *log.Logger)
• func ProcessBook(msg bookpipeline.Qmsg, conn Pipeliner, ...) error
• func SaveLogs(conn Uploader, starttime int64, hostname string) error
• func UploadImages(dir string, bookname string, conn Uploader) error
• func Wipe(towipe chan string, up chan string, errc chan error, logger *log.Logger)
• type DownloadLister
• type Downloader
• type Lister
• type MinPipeliner
• type NullWriter
  • func (w NullWriter) Write(p []byte) (n int, err error)
• type Pipeliner
• type Queuer
• type UploadQueuer
• type Uploader

Constants

const HeartbeatSeconds = 60

Functions

func Analyse

func Analyse(conn Downloader) func(chan string, chan string, chan error, *log.Logger)

func CheckImages

func CheckImages(dir string) error

CheckImages checks that all files in a directory are images that can be decoded (skipping dotfiles)

func DetectQueueType

func DetectQueueType(dir string, conn Queuer) string

DetectQueueType detects which queue to use based on the preponderance of files of a particular extension in a directory

func DownloadAll

func DownloadAll(dir string, name string, conn DownloadLister) error

func DownloadAnalyses

func DownloadAnalyses(dir string, name string, conn Downloader) error

func DownloadBestPages

func DownloadBestPages(dir string, name string, conn Downloader, pluspngs bool) error

func DownloadPdfs

func DownloadPdfs(dir string, name string, conn Downloader) error

func GetMailSettings

func GetMailSettings() (mailSettings, error)

func Ocr

func Ocr(training string, tesscmd string) func(chan string, chan string, chan error, *log.Logger)

func OcrPage

func OcrPage(msg bookpipeline.Qmsg, conn Pipeliner, process func(chan string, chan string, chan error, *log.Logger), fromQueue string, toQueue string) error

OcrPage OCRs a page based on a message. It may make sense to roll this back into processBook (on which it is based) once working well.

func Preprocess

func Preprocess(thresholds []float64) func(chan string, chan string, chan error, *log.Logger)

func ProcessBook

func ProcessBook(msg bookpipeline.Qmsg, conn Pipeliner, process func(chan string, chan string, chan error, *log.Logger), match *regexp.Regexp, fromQueue string, toQueue string) error

func SaveLogs

func SaveLogs(conn Uploader, starttime int64, hostname string) error

func UploadImages

func UploadImages(dir string, bookname string, conn Uploader) error

UploadImages uploads all files (except those which start with a ".") from a directory (recursively) into conn.WIPStorageId(), prefixed with the given bookname and a slash

func Wipe

func Wipe(towipe chan string, up chan string, errc chan error, logger *log.Logger)

Types

type DownloadLister

type DownloadLister interface {
	Download(bucket string, key string, fn string) error
	ListObjects(bucket string, prefix string) ([]string, error)
	Log(v ...interface{})
	WIPStorageId() string
}

type Downloader

type Downloader interface {
	Download(bucket string, key string, fn string) error
	Log(v ...interface{})
	WIPStorageId() string
}

type Lister

type Lister interface {
	ListObjects(bucket string, prefix string) ([]string, error)
	Log(v ...interface{})
	WIPStorageId() string
}

type MinPipeliner

type MinPipeliner interface {
	Pipeliner
	MinimalInit() error
}

type NullWriter

type NullWriter bool

null writer to enable non-verbose logging to be discarded

func (NullWriter) Write

func (w NullWriter) Write(p []byte) (n int, err error)

type Pipeliner

type Pipeliner interface {
	AddToQueue(url string, msg string) error
	AnalyseQueueId() string
	CheckQueue(url string, timeout int64) (bookpipeline.Qmsg, error)
	DelFromQueue(url string, handle string) error
	Download(bucket string, key string, fn string) error
	GetLogger() *log.Logger
	Init() error
	ListObjects(bucket string, prefix string) ([]string, error)
	Log(v ...interface{})
	OCRPageQueueId() string
	PreQueueId() string
	QueueHeartbeat(msg bookpipeline.Qmsg, qurl string, duration int64) (bookpipeline.Qmsg, error)
	Upload(bucket string, key string, path string) error
	WipeQueueId() string
	WIPStorageId() string
}

type Queuer

type Queuer interface {
	AddToQueue(url string, msg string) error
	AnalyseQueueId() string
	CheckQueue(url string, timeout int64) (bookpipeline.Qmsg, error)
	DelFromQueue(url string, handle string) error
	Log(v ...interface{})
	OCRPageQueueId() string
	PreQueueId() string
	QueueHeartbeat(msg bookpipeline.Qmsg, qurl string, duration int64) (bookpipeline.Qmsg, error)
	WipeQueueId() string
}

type UploadQueuer

type UploadQueuer interface {
	Log(v ...interface{})
	Upload(bucket string, key string, path string) error
	WIPStorageId() string
	PreQueueId() string
	WipeQueueId() string
	OCRPageQueueId() string
	AnalyseQueueId() string
	CheckQueue(url string, timeout int64) (bookpipeline.Qmsg, error)
	AddToQueue(url string, msg string) error
	DelFromQueue(url string, handle string) error
	QueueHeartbeat(msg bookpipeline.Qmsg, qurl string, duration int64) (bookpipeline.Qmsg, error)
}

type Uploader

type Uploader interface {
	Log(v ...interface{})
	Upload(bucket string, key string, path string) error
	WIPStorageId() string
}