<commit_before>from __future__ import print_function, division, absolute_import#, unicode_literals # not casa compatible
from builtins import bytes, dict, object, range, map, input#, str # not casa compatible
from future.utils import itervalues, viewitems, iteritems, listvalues, listitems
from io import open

import pickle
import os.path
import glob
import shutil
import random
import distributed
from astropy import time
from time import sleep
import dask.utils
from evla_mcast.controller import Controller
from rfpipe import state, preferences, candidates
from realfast import pipeline, elastic, sdm_builder, heuristics

import logging
import matplotlib
import yaml
matplotlib.use('Agg')
ch = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s %(levelname)8s %(name)s | %(message)s')
ch.setFormatter(formatter)
logger = logging.getLogger('realfast_controller')

_vys_cfile_prod = '/home/cbe-master/realfast/lustre_workdir/vys.conf'  # production file
_vys_cfile_test = '/home/cbe-master/realfast/soft/vysmaw_apps/vys.conf'  # test file
_preffile = '/lustre/evla/test/realfast/realfast.yml'
_distributed_host = '192.168.201.101'  # for ib0 on cbe-node-01
_candplot_dir = '/users/claw/public_html/realfast/plots'
_candplot_url_prefix = 'http://www.aoc.nrao.edu/~claw/realfast/plots'

# standards should always have l=0 or m=0
# list of (seg, i0, dm, dt, amp, l, m)
_mock_standards = [(0, 1, 20, 0.01, 0.1, 1e-3, 0.),
                   (0, 1, 20, 0.01, 0.1, -1e-3, 0.),
                   (0, 1, 20, 0.01, 0.1, 0., 1e-3),
                   (0, 1, 20, 0.01, 0.1, 0., -1e-3)]


class realfast_controller(Controller):

    def __init__(self, preffile=_preffile, inprefs={}, **kwargs):
        """ Creates controller object that can act on a scan configuration.
        Inherits a "run" method that starts asynchronous operation that calls
        handle_config. handle_sdm and handle_meta (with datasource "vys" or
        "sim") are also supported.

        kwargs can include:
        - tags, a comma-delimited string for cands to index (None -> "new"),
        - nameincludes, a string required to be in datasetId,
        - vys_timeout, factor over real-time for vys reading to wait,
        - mockprob, chance (range 0-1) that a mock is added,
        - saveproducts, boolean defining generation of mini-sdm,
        - indexresults, boolean defining push (meta)data to search index,
        - archiveproducts, boolean defining archiving mini-sdm,
        - throttle, boolean defining whether to slow pipeline submission,
        - read_overhead, throttle param requires multiple of vismem in a READERs memory,
        - read_totfrac, throttle param requires fraction of total READER memory be available,
        - spill_limit, throttle param limiting maximum size (in GB) of data spill directory,
        - searchintents, a list of intent names to search.
        """

        super(realfast_controller, self).__init__()

        self.inprefs = inprefs  # rfpipe preferences
        self.mockset = _mock_standards
        self.client = distributed.Client('{0}:{1}'
                                         .format(_distributed_host, '8786'))
        self.lock = dask.utils.SerializableLock()
        self.states = {}
        self.futures = {}
        self.futures_removed = {}

        # define attributes from yaml file
        prefs = {}
        self.preffile = preffile
        if self.preffile is not None:
            if os.path.exists(self.preffile):
                with open(self.preffile, 'r') as fp:
                    prefs = yaml.load(fp)['realfast']
                    logger.info("Parsed realfast preferences from {0}"
                                .format(self.preffile))
            else:
                logger.warn("realfast preffile {0} given, but not found"
                            .format(self.preffile))
        else:
            logger.info("No realfast preffile provided.")

        # get arguments from preffile, optional overload from kwargs
        for attr in ['tags', 'nameincludes', 'mockprob', 'vys_timeout',
                     'indexresults', 'saveproducts', 'archiveproducts',
                     'searchintents', 'throttle', 'read_overhead',
                     'read_totfrac', 'spill_limit']:
            setattr(self, attr, None)
            if attr in prefs:
                setattr(self, attr, prefs[attr])
            if attr in kwargs:
                setattr(self, attr, kwargs[attr])

    def __repr__(self):
        return ('realfast controller with {0} jobs'
                .format(len(self.futures)))

    @property
    def statuses(self):
        return ['{0}, {1}: {2}'.format(scanId, ftype,
                                       futures[ftype].status)
                for (scanId, futurelist) in iteritems(self.futures)
                for futures in futurelist
                for ftype in futures]

    @property
    def errors(self):
        return ['{0}, {1}: {2}'.format(scanId, ftype,
                                       futures[ftype].exception())
                for (scanId, futurelist) in iteritems(self.futures)
                for futures in futurelist
                for ftype in futures
                if futures[ftype].status == 'error']

    def handle_config(self, config, cfile=_vys_cfile_prod, segments=None):
        """ Triggered when obs comes in.
        Downstream logic starts here.
        Default vys config file uses production parameters.
        segments arg can be used to submit a subset of all segments.

        """

        summarize(config)

        if 'simulated_transient' not in self.inprefs:
            self.inject_transient(config.scanId)  # randomly inject mock transient

        if search_config(config, nameincludes=self.nameincludes,
                         searchintents=self.searchintents):
            logger.info('Config looks good. Generating rfpipe state...')

            self.set_state(config.scanId, config=config, inmeta={'datasource':
                                                                 'vys'})

            if self.indexresults:
                elastic.indexscan_config(config,
                                         preferences=self.states[config.scanId].prefs,
                                         datasource='vys')
            else:
                logger.info("Not indexing config or prefs.")

            self.start_pipeline(config.scanId, cfile=cfile, segments=segments)

        else:
            logger.info("Config not suitable for realfast. Skipping.")

        self.cleanup()

    def handle_sdm(self, sdmfile, sdmscan, bdfdir=None, segments=None):
        """ Parallel to handle_config, but allows sdm to be passed in.
        segments arg can be used to submit a subset of all segments.
        """

        # TODO: subscan assumed = 1
        sdmsubscan = 1
        scanId = '{0}.{1}.{2}'.format(os.path.basename(sdmfile.rstrip('/')),
                                      str(sdmscan), str(sdmsubscan))

        if 'simulated_transient' not in self.inprefs:
            self.inject_transient(scanId)  # randomly inject mock transient

        self.set_state(scanId, sdmfile=sdmfile, sdmscan=sdmscan, bdfdir=bdfdir,
                       inmeta={'datasource': 'sdm'})

        if self.indexresults:
            elastic.indexscan_sdm(sdmfile, sdmscan, sdmsubscan,
                                  preferences=self.states[scanId].prefs,
                                  datasource='sdm')
        else:
            logger.info("Not indexing sdm scan or prefs.")

        self.start_pipeline(scanId, segments=segments)

        self.cleanup()

    def handle_meta(self, inmeta, cfile=_vys_cfile_test, segments=None):
        """ Parallel to handle_config, but allows metadata dict to be passed in.
        Gets called explicitly.
        Default vys config file uses test parameters.
        inmeta datasource key can be 'vys' or 'sim' and is passed to rfpipe.
        segments arg can be used to submit a subset of all segments.
        """

        scanId = '{0}.{1}.{2}'.format(inmeta['datasetId'], str(inmeta['scan']),
                                      str(inmeta['subscan']))

        if 'simulated_transient' not in self.inprefs:
            self.inject_transient(scanId)  # randomly inject mock transient

        self.set_state(scanId, inmeta=inmeta)

        self.start_pipeline(scanId, cfile=cfile, segments=segments)

        self.cleanup()

    def set_state(self, scanId, config=None, inmeta=None, sdmfile=None,
                  sdmscan=None, bdfdir=None, prefsname=None):
        """ Given metadata source, define state for a scanId.
        """

        # TODO: define prefsname according to config and/or heuristics
        prefs = preferences.Preferences(**preferences.parsepreffile(self.preffile,
                                                                    name=prefsname,
                                                                    inprefs=self.inprefs))

        st = state.State(inmeta=inmeta, config=config, inprefs=prefs,
                         lock=self.lock,
                         sdmfile=sdmfile, sdmscan=sdmscan, bdfdir=bdfdir)

        logger.info('State set for scanId {0}. Requires {1:.1f} GB read and'
                    ' {2:.1f} GPU-sec to search.'
                    .format(st.metadata.scanId,
                            heuristics.total_memory_read(st),
                            heuristics.total_compute_time(st)))

        self.states[scanId] = st

    def start_pipeline(self, scanId, cfile=None, segments=None):
        """ Start pipeline conditional on cluster state.
        Sets futures and state after submission keyed by scanId.
        segments arg can be used to select or slow segment submission.
        """

        st = self.states[scanId]
        if not heuristics.valid_telcalfile(st):
            logger.warn("telcalfile {0} for scanId {1} is not available at "
                        "pipeline submission"
                        .format(st.gainfile, scanId))

        # submit, with optional throttling
        futures = []
        if self.throttle and self.read_overhead and self.read_totfrac and self.spill_limit:
            logger.info('Starting pipeline throttled by read_overhead {0}, '
                        'read_totfrac {1}, and spill_limit {2}'
                        .format(self.read_overhead, self.read_totfrac,
                                self.spill_limit))

            w_memlim = self.read_overhead*st.vismem*1e9

            tot_memlim = self.read_totfrac*sum([v['memory_limit']
                                                for v in itervalues(self.client.scheduler_info()['workers'])
                                                if 'READER' in v['resources']])

            t0 = time.Time.now().unix
            timeout = 0.8*st.metadata.inttime*st.metadata.nints  # bit shorter than scan
            elapsedtime = time.Time.now().unix - t0

            if segments is None:
                segments = range(st.nsegment)

            while (elapsedtime < timeout) and (len(futures) < len(segments)):
                # Submit if workers are not overloaded
                if (heuristics.worker_memory_ok(self.client, w_memlim) and
                   heuristics.total_memory_ok(self.client, tot_memlim) and
                   heuristics.spilled_memory_ok(limit=self.spill_limit)):
                    futures = pipeline.pipeline_scan_delayed(st,
                                                             segments=segments,
                                                             cl=self.client,
                                                             cfile=cfile,
                                                             vys_timeout=self.vys_timeout)
                else:
                    sleep(min(1, timeout/10))
                    self.cleanup()
                    elapsedtime = time.Time.now().unix - t0

            if elapsedtime > timeout:
                logger.info("Throttle timed out. ScanId {0} not submitted."
                            .format(st.metadata.scanId))

        else:
            logger.info('Starting pipeline...')
            futures = pipeline.pipeline_scan_delayed(st, segments=segments,
                                                     cl=self.client,
                                                     cfile=cfile,
                                                     vys_timeout=self.vys_timeout)

        if len(futures):
            self.futures[st.metadata.scanId] = futures

    def cleanup(self, badstatuslist=['cancelled', 'error', 'lost']):
        """ Clean up job list.
        Scans futures, removes finished jobs, and pushes results to relevant indices.
        badstatuslist can include 'cancelled', 'error', 'lost'.
        """

        removed = 0
        cindexed = 0
        sdms = 0

        scanIds = [scanId for scanId in self.futures]
        if len(scanIds):
            logger.info("Checking on jobs from {0} scanIds: {1}"
                        .format(len(scanIds), scanIds))

        for scanId in self.futures:
            # create list of futures (a dict per segment) that are done
            finishedlist = [futures for (scanId0, futurelist) in iteritems(self.futures)
                            for futures in futurelist
                            if (futures['candcollection'].status == 'finished') and
                               (scanId0 == scanId)]

            # one candcollection per segment
            for futures in finishedlist:
                candcollection = futures['candcollection'].result()
                if len(candcollection.array):
                    if self.indexresults:
                        nplots = moveplots(candcollection.prefs.workdir,
                                           scanId, destination=_candplot_dir)
#                       TODO: can we use client here?
                        res = elastic.indexcands(candcollection, scanId,
                                                 tags=self.tags,
                                                 url_prefix=_candplot_url_prefix)
                        if res or nplots:
                            logger.info('Indexed {0} candidates and moved {1} '
                                        'plots to {2}'
                                        .format(res, nplots, _candplot_dir))
                        else:
                            logger.info('No candidates or plots found.')

                        cindexed += res
                    else:
                        logger.info("Not indexing cands.")

                    makesummaryplot(candcollection.prefs.workdir, scanId)

                else:
                    logger.debug('No candidates for a segment from scanId {0}'
                                 .format(scanId))

# TODO: index noises
#                if os.path.exists(st.noisefile):
#                   if self.indexresults:
#                       res = elastic.indexnoises(st.noisefile, scanId)
#                        nindexed += res
#                   else:
#                      logger.info("Not indexing noises.")
#                else:
#                    logger.info('No noisefile found, no noises indexed.')

                # optionally save and archive sdm/bdfs for segment
                if self.saveproducts and len(candcollection):<commit_msg>played with submitting len function but it is not reliably schedule<commit_after>from __future__ import print_function, division, absolute_import#, unicode_literals # not casa compatible
from builtins import bytes, dict, object, range, map, input#, str # not casa compatible
from future.utils import itervalues, viewitems, iteritems, listvalues, listitems
from io import open

import pickle
import os.path
import glob
import shutil
import random
import distributed
from astropy import time
from time import sleep
import dask.utils
from evla_mcast.controller import Controller
from rfpipe import state, preferences, candidates
from realfast import pipeline, elastic, sdm_builder, heuristics

import logging
import matplotlib
import yaml
matplotlib.use('Agg')
ch = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s %(levelname)8s %(name)s | %(message)s')
ch.setFormatter(formatter)
logger = logging.getLogger('realfast_controller')

_vys_cfile_prod = '/home/cbe-master/realfast/lustre_workdir/vys.conf'  # production file
_vys_cfile_test = '/home/cbe-master/realfast/soft/vysmaw_apps/vys.conf'  # test file
_preffile = '/lustre/evla/test/realfast/realfast.yml'
_distributed_host = '192.168.201.101'  # for ib0 on cbe-node-01
_candplot_dir = '/users/claw/public_html/realfast/plots'
_candplot_url_prefix = 'http://www.aoc.nrao.edu/~claw/realfast/plots'

# standards should always have l=0 or m=0
# list of (seg, i0, dm, dt, amp, l, m)
_mock_standards = [(0, 1, 20, 0.01, 0.1, 1e-3, 0.),
                   (0, 1, 20, 0.01, 0.1, -1e-3, 0.),
                   (0, 1, 20, 0.01, 0.1, 0., 1e-3),
                   (0, 1, 20, 0.01, 0.1, 0., -1e-3)]


class realfast_controller(Controller):

    def __init__(self, preffile=_preffile, inprefs={}, **kwargs):
        """ Creates controller object that can act on a scan configuration.
        Inherits a "run" method that starts asynchronous operation that calls
        handle_config. handle_sdm and handle_meta (with datasource "vys" or
        "sim") are also supported.

        kwargs can include:
        - tags, a comma-delimited string for cands to index (None -> "new"),
        - nameincludes, a string required to be in datasetId,
        - vys_timeout, factor over real-time for vys reading to wait,
        - mockprob, chance (range 0-1) that a mock is added,
        - saveproducts, boolean defining generation of mini-sdm,
        - indexresults, boolean defining push (meta)data to search index,
        - archiveproducts, boolean defining archiving mini-sdm,
        - throttle, boolean defining whether to slow pipeline submission,
        - read_overhead, throttle param requires multiple of vismem in a READERs memory,
        - read_totfrac, throttle param requires fraction of total READER memory be available,
        - spill_limit, throttle param limiting maximum size (in GB) of data spill directory,
        - searchintents, a list of intent names to search.
        """

        super(realfast_controller, self).__init__()

        self.inprefs = inprefs  # rfpipe preferences
        self.mockset = _mock_standards
        self.client = distributed.Client('{0}:{1}'
                                         .format(_distributed_host, '8786'))
        self.lock = dask.utils.SerializableLock()
        self.states = {}
        self.futures = {}
        self.futures_removed = {}

        # define attributes from yaml file
        prefs = {}
        self.preffile = preffile
        if self.preffile is not None:
            if os.path.exists(self.preffile):
                with open(self.preffile, 'r') as fp:
                    prefs = yaml.load(fp)['realfast']
                    logger.info("Parsed realfast preferences from {0}"
                                .format(self.preffile))
            else:
                logger.warn("realfast preffile {0} given, but not found"
                            .format(self.preffile))
        else:
            logger.info("No realfast preffile provided.")

        # get arguments from preffile, optional overload from kwargs
        for attr in ['tags', 'nameincludes', 'mockprob', 'vys_timeout',
                     'indexresults', 'saveproducts', 'archiveproducts',
                     'searchintents', 'throttle', 'read_overhead',
                     'read_totfrac', 'spill_limit']:
            setattr(self, attr, None)
            if attr in prefs:
                setattr(self, attr, prefs[attr])
            if attr in kwargs:
                setattr(self, attr, kwargs[attr])

    def __repr__(self):
        return ('realfast controller with {0} jobs'
                .format(len(self.futures)))

    @property
    def statuses(self):
        return ['{0}, {1}: {2}'.format(scanId, ftype,
                                       futures[ftype].status)
                for (scanId, futurelist) in iteritems(self.futures)
                for futures in futurelist
                for ftype in futures]

    @property
    def errors(self):
        return ['{0}, {1}: {2}'.format(scanId, ftype,
                                       futures[ftype].exception())
                for (scanId, futurelist) in iteritems(self.futures)
                for futures in futurelist
                for ftype in futures
                if futures[ftype].status == 'error']

    def handle_config(self, config, cfile=_vys_cfile_prod, segments=None):
        """ Triggered when obs comes in.
        Downstream logic starts here.
        Default vys config file uses production parameters.
        segments arg can be used to submit a subset of all segments.

        """

        summarize(config)

        if 'simulated_transient' not in self.inprefs:
            self.inject_transient(config.scanId)  # randomly inject mock transient

        if search_config(config, nameincludes=self.nameincludes,
                         searchintents=self.searchintents):
            logger.info('Config looks good. Generating rfpipe state...')

            self.set_state(config.scanId, config=config, inmeta={'datasource':
                                                                 'vys'})

            if self.indexresults:
                elastic.indexscan_config(config,
                                         preferences=self.states[config.scanId].prefs,
                                         datasource='vys')
            else:
                logger.info("Not indexing config or prefs.")

            self.start_pipeline(config.scanId, cfile=cfile, segments=segments)

        else:
            logger.info("Config not suitable for realfast. Skipping.")

        self.cleanup()

    def handle_sdm(self, sdmfile, sdmscan, bdfdir=None, segments=None):
        """ Parallel to handle_config, but allows sdm to be passed in.
        segments arg can be used to submit a subset of all segments.
        """

        # TODO: subscan assumed = 1
        sdmsubscan = 1
        scanId = '{0}.{1}.{2}'.format(os.path.basename(sdmfile.rstrip('/')),
                                      str(sdmscan), str(sdmsubscan))

        if 'simulated_transient' not in self.inprefs:
            self.inject_transient(scanId)  # randomly inject mock transient

        self.set_state(scanId, sdmfile=sdmfile, sdmscan=sdmscan, bdfdir=bdfdir,
                       inmeta={'datasource': 'sdm'})

        if self.indexresults:
            elastic.indexscan_sdm(sdmfile, sdmscan, sdmsubscan,
                                  preferences=self.states[scanId].prefs,
                                  datasource='sdm')
        else:
            logger.info("Not indexing sdm scan or prefs.")

        self.start_pipeline(scanId, segments=segments)

        self.cleanup()

    def handle_meta(self, inmeta, cfile=_vys_cfile_test, segments=None):
        """ Parallel to handle_config, but allows metadata dict to be passed in.
        Gets called explicitly.
        Default vys config file uses test parameters.
        inmeta datasource key can be 'vys' or 'sim' and is passed to rfpipe.
        segments arg can be used to submit a subset of all segments.
        """

        scanId = '{0}.{1}.{2}'.format(inmeta['datasetId'], str(inmeta['scan']),
                                      str(inmeta['subscan']))

        if 'simulated_transient' not in self.inprefs:
            self.inject_transient(scanId)  # randomly inject mock transient

        self.set_state(scanId, inmeta=inmeta)

        self.start_pipeline(scanId, cfile=cfile, segments=segments)

        self.cleanup()

    def set_state(self, scanId, config=None, inmeta=None, sdmfile=None,
                  sdmscan=None, bdfdir=None, prefsname=None):
        """ Given metadata source, define state for a scanId.
        """

        # TODO: define prefsname according to config and/or heuristics
        prefs = preferences.Preferences(**preferences.parsepreffile(self.preffile,
                                                                    name=prefsname,
                                                                    inprefs=self.inprefs))

        st = state.State(inmeta=inmeta, config=config, inprefs=prefs,
                         lock=self.lock,
                         sdmfile=sdmfile, sdmscan=sdmscan, bdfdir=bdfdir)

        logger.info('State set for scanId {0}. Requires {1:.1f} GB read and'
                    ' {2:.1f} GPU-sec to search.'
                    .format(st.metadata.scanId,
                            heuristics.total_memory_read(st),
                            heuristics.total_compute_time(st)))

        self.states[scanId] = st

    def start_pipeline(self, scanId, cfile=None, segments=None):
        """ Start pipeline conditional on cluster state.
        Sets futures and state after submission keyed by scanId.
        segments arg can be used to select or slow segment submission.
        """

        st = self.states[scanId]
        if not heuristics.valid_telcalfile(st):
            logger.warn("telcalfile {0} for scanId {1} is not available at "
                        "pipeline submission"
                        .format(st.gainfile, scanId))

        # submit, with optional throttling
        futures = []
        if self.throttle and self.read_overhead and self.read_totfrac and self.spill_limit:
            logger.info('Starting pipeline throttled by read_overhead {0}, '
                        'read_totfrac {1}, and spill_limit {2}'
                        .format(self.read_overhead, self.read_totfrac,
                                self.spill_limit))

            w_memlim = self.read_overhead*st.vismem*1e9

            tot_memlim = self.read_totfrac*sum([v['memory_limit']
                                                for v in itervalues(self.client.scheduler_info()['workers'])
                                                if 'READER' in v['resources']])

            t0 = time.Time.now().unix
            timeout = 0.8*st.metadata.inttime*st.metadata.nints  # bit shorter than scan
            elapsedtime = time.Time.now().unix - t0

            if segments is None:
                segments = range(st.nsegment)

            while (elapsedtime < timeout) and (len(futures) < len(segments)):
                # Submit if workers are not overloaded
                if (heuristics.worker_memory_ok(self.client, w_memlim) and
                   heuristics.total_memory_ok(self.client, tot_memlim) and
                   heuristics.spilled_memory_ok(limit=self.spill_limit)):
                    futures = pipeline.pipeline_scan_delayed(st,
                                                             segments=segments,
                                                             cl=self.client,
                                                             cfile=cfile,
                                                             vys_timeout=self.vys_timeout)
                else:
                    sleep(min(1, timeout/10))
                    self.cleanup()
                    elapsedtime = time.Time.now().unix - t0

            if elapsedtime > timeout:
                logger.info("Throttle timed out. ScanId {0} not submitted."
                            .format(st.metadata.scanId))

        else:
            logger.info('Starting pipeline...')
            futures = pipeline.pipeline_scan_delayed(st, segments=segments,
                                                     cl=self.client,
                                                     cfile=cfile,
                                                     vys_timeout=self.vys_timeout)

        if len(futures):
            self.futures[st.metadata.scanId] = futures

    def cleanup(self, badstatuslist=['cancelled', 'error', 'lost']):
        """ Clean up job list.
        Scans futures, removes finished jobs, and pushes results to relevant indices.
        badstatuslist can include 'cancelled', 'error', 'lost'.
        """

        removed = 0
        cindexed = 0
        sdms = 0

        scanIds = [scanId for scanId in self.futures]
        if len(scanIds):
            logger.info("Checking on jobs from {0} scanIds: {1}"
                        .format(len(scanIds), scanIds))

        for scanId in self.futures:
            # create list of futures (a dict per segment) that are done
            finishedlist = [futures for (scanId0, futurelist) in iteritems(self.futures)
                            for futures in futurelist
                            if (futures['candcollection'].status == 'finished') and
                               (scanId0 == scanId)]

            # one candcollection per segment
            for futures in finishedlist:
                candcollection = futures['candcollection'].result()
                ncands = len(candcollection)
#                ncands = self.client.submit(len, futures['candcollection'], priority=3).result() # not dependably scheduled
                if ncands:
#                    candcollection = futures['candcollection'].result()
                    if self.indexresults:
                        nplots = moveplots(candcollection.prefs.workdir,
                                           scanId, destination=_candplot_dir)
#                       TODO: can we use client here?
                        res = elastic.indexcands(candcollection, scanId,
                                                 tags=self.tags,
                                                 url_prefix=_candplot_url_prefix)
                        if res or nplots:
                            logger.info('Indexed {0} candidates and moved {1} '
                                        'plots to {2}'
                                        .format(res, nplots, _candplot_dir))
                        else:
                            logger.info('No candidates or plots found.')

                        cindexed += res
                    else:
                        logger.info("Not indexing cands found in scanId {0}"
                                    .format(scanId))

                    makesummaryplot(candcollection.prefs.workdir, scanId)

                else:
                    logger.debug('No candidates for a segment from scanId {0}'
                                 .format(scanId))

# TODO: index noises
#                if os.path.exists(st.noisefile):
#                   if self.indexresults:
#                       res = elastic.indexnoises(st.noisefile, scanId)
#                        nindexed += res
#                   else:
#                      logger.info("Not indexing noises.")
#                else:
#                    logger.info('No noisefile found, no noises indexed.')

                # optionally save and archive sdm/bdfs for segment
                if self.saveproducts and ncands:
{'fraction_non_alphanumeric': 0.09011964492473948, 'fraction_numerical': 0.008040653544320082, 'mean_word_length': 1.6588848982384128, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 91, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6308896', 'n_tokens_mistral': 8609, 'n_tokens_neox': 8114, 'n_words': 2189}