<commit_before>import * as c from 'colors/safe'
import { Observable, ReplaySubject } from 'rxjs'
import { TransformStream } from './transform-streams'
import { InputStream } from './input-streams'
import { OutputStream } from './output-streams'
import * as shared from './shared'
import Subscription from './subscription'
import logger from './logger'

export interface Pipe {
  name: string,
  in: InputStream,
  out: OutputStream
}

export interface Transformers {
  [name: string]: Array<TransformStream>
}

export interface Subscriptions {
  [id: string]: Subscription
}

export type LogLevel = 'NONE' | 'DEBUG' | 'INFO'

export interface Config {
  LOG_LEVEL?: LogLevel
}

export class MutatorIO {
  transformers: Transformers = {}
  subscriptions: Subscriptions = {}

  constructor (public pipes: Array<Pipe>, public config: Config = {}) {
    logger.level = config.LOG_LEVEL.toLowerCase() || 'info'
  }

  removeTransformer (pipeName: string, index: number): boolean {
    if (!this.transformers[pipeName][index]) {
      return false
    }
    this.transformers[pipeName].slice(index, 1)
    return true
  }

  transform (pipeName: string, transform: TransformStream): Subscription {
    this.transformers[pipeName] = this.transformers[pipeName] || []
    const lastIndex = this.transformers[pipeName].push(transform) - 1

    const subscription = new Subscription(this, pipeName, lastIndex)
    transform.subscriptionId = subscription.id
    this.subscriptions[subscription.id] = subscription
    return subscription
  }

  start (): void {
    interface pipeResult extends Array<Object> {
      0: string,
      1: Observable<Object>
    }

    const streams = Object.keys(this.transformers)
      .reduce((acc, pipeName): Array<pipeResult> => {
        const currentPipe = this.pipes.find((pipe) => pipe.name === pipeName)

        const currentPipeResult = this.transformers[pipeName]
          .map((transformer): pipeResult => {
            const inStream = currentPipe.in.create()
            const outStream = currentPipe.out.create()

            return [
              pipeName,
              inStream
                .do((msg) => logger.debug(c.yellow('pre-transformation'), msg))
                .flatMap((msg) => shared.wrapToObservable(transformer.call(this, msg)))
                .do((msg) => logger.debug(c.yellow('post-transformation'), msg))
                .flatMap((msg) => shared.wrapToObservable(outStream(msg))
                  .catch((err) => {
                    logger.error(err)
                    return inStream
                  })
                ),
              this.subscriptions[transformer.subscriptionId]
            ]
          })

        return acc.concat(currentPipeResult)
      }, [])

    streams.map(([pipeName, stream, subscription]) => {
      logger.info(`${c.rainbow('•••')} Listening on pipe ${pipeName} ${c.rainbow('•••')}`)
      subscription.disposable = stream
        .subscribe(
          (msg) => logger.info(msg),
          (err) => logger.error(err),
          () => logger.info(`${c.rainbow('•••')} ${pipeName} pipe closed ${c.rainbow('•••')}`)
        )
    })
  }
}
<commit_msg>Added further type checking :ballot_box_with_check:<commit_after>import * as c from 'colors/safe'
import { Observable, ReplaySubject } from 'rxjs'
import { TransformStream } from './transform-streams'
import { InputStream } from './input-streams'
import { OutputStream } from './output-streams'
import * as shared from './shared'
import Subscription from './subscription'
import logger from './logger'

export interface Pipe {
  name: string,
  in: InputStream,
  out: OutputStream
}

export interface Transformers {
  [name: string]: Array<TransformStream>
}

export interface Subscriptions {
  [id: string]: Subscription
}

export type LogLevel = 'NONE' | 'DEBUG' | 'INFO'

export interface Config {
  LOG_LEVEL?: LogLevel
}

export class MutatorIO {
  transformers: Transformers = {}
  subscriptions: Subscriptions = {}

  constructor (public pipes: Array<Pipe>, public config: Config = {}) {
    logger.level = config.LOG_LEVEL.toLowerCase() || 'info'
  }

  removeTransformer (pipeName: string, index: number): boolean {
    if (!this.transformers[pipeName][index]) {
      return false
    }
    this.transformers[pipeName].slice(index, 1)
    return true
  }

  transform (pipeName: string, transform: TransformStream): Subscription {
    this.transformers[pipeName] = this.transformers[pipeName] || []
    const lastIndex = this.transformers[pipeName].push(transform) - 1

    const subscription = new Subscription(this, pipeName, lastIndex)
    transform.subscriptionId = subscription.id
    this.subscriptions[subscription.id] = subscription
    return subscription
  }

  start (): void {
    interface pipeResult extends Array<Object> {
      0: string,
      1: Observable<Object>
      2: Subscription
    }

    const streams = Object.keys(this.transformers)
      .reduce((acc, pipeName): Array<pipeResult> => {
        const currentPipe = this.pipes.find((pipe) => pipe.name === pipeName)

        const currentPipeResult = this.transformers[pipeName]
          .map((transformer): pipeResult => {
            const inStream = currentPipe.in.create()
            const outStream = currentPipe.out.create()

            return [
              pipeName,
              inStream
                .do((msg) => logger.debug(c.yellow('pre-transformation'), msg))
                .flatMap((msg) => shared.wrapToObservable(transformer.call(this, msg)))
                .do((msg) => logger.debug(c.yellow('post-transformation'), msg))
                .flatMap((msg) => shared.wrapToObservable(outStream(msg))
                  .catch((err) => {
                    logger.error(err)
                    return inStream
                  })
                ),
              this.subscriptions[transformer.subscriptionId]
            ]
          })

        return acc.concat(currentPipeResult)
      }, [])

    streams.map(([pipeName, stream, subscription]: pipeResult) => {
      logger.info(`${c.rainbow('•••')} Listening on pipe ${pipeName} ${c.rainbow('•••')}`)
      subscription.disposable = stream
        .subscribe(
          (msg) => logger.info(msg),
          (err) => logger.error(err),
          () => logger.info(`${c.rainbow('•••')} ${pipeName} pipe closed ${c.rainbow('•••')}`)
        )
    })
  }
}

{'fraction_non_alphanumeric': 0.1355081555834379, 'fraction_numerical': 0.0014115432873274781, 'mean_word_length': 2.751176470588235, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 37, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3695627', 'n_tokens_mistral': 1855, 'n_tokens_neox': 1806, 'n_words': 399}