<commit_before>#! /usr/bin/env python

# Main controller. <NAME> June 2015
#
# Based on frb_trigger_controller.py by <NAME>, 2015/02
#
# Listen for OBS packets having a certain 'triggered archiving'
# intent, and perform some as-yet-unspecified action when these
# are recieved.
#

import datetime
import os
import asyncore
import click
import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

from realfast import queue_monitor, rtutils, mcaf_library

# set up
rtparams_default = os.path.join(os.path.join(os.path.split(os.path.split(mcaf_library.__file__)[0])[0], 'conf'), 'rtpipe_cbe.conf') # install system puts conf files here. used by queue_rtpipe.py
telcaldir = '/home/mchammer/evladata/telcal'  # then yyyy/mm
workdir = os.getcwd()     # assuming we start in workdir
redishost = os.uname()[1]  # assuming we start on redis host

class FRBController(object):
    """Listens for OBS packets and tells FRB processing about any
    notable scans."""

    def __init__(self, intent='', project='', production=False, verbose=False, rtparams=''):
        # Mode can be project, intent
        self.intent = intent
        self.project = project
        self.production = production
        self.verbose = verbose
        self.rtparams = rtparams

    def add_sdminfo(self, sdminfo):
        config = mcaf_library.MCAST_Config(sdminfo=sdminfo)

        # !!! Wrapper here to deal with potential subscans?

        # Check if MCAST message is simply telling us the obs is finished
        if config.obsComplete:
            logger.info("Received finalMessage=True; This observation has completed.")
            # if completing the desired SB, then do a final rsync
            if self.project in config.projectID:
                logger.info("Final rsync to make workdir copy of SDM %sd complete." % (config.sdmLocation.rstrip('/')))
                rtutils.rsync(config.sdmLocation.rstrip('/'), workdir)  # final transfer to ensure complete SDM in workdir

        elif self.intent in config.intentString and self.project in config.projectID:
            logger.info("Scan %d has desired intent (%s) and project (%s)" % (config.scan, self.intent, self.project))
            logger.debug("BDF is in %s\n" % (config.bdfLocation))

            # If we're not in listening mode, prepare data and submit to queue system
            if self.production:
                filename = config.sdmLocation.rstrip('/')
                scan = int(config.scan)

                assert len(filename) and isinstance(filename, str), 'Filename empty or not a string?'

                # check that SDM is usable by rtpipe. Currently checks spw order and duplicates.
                if rtutils.check_spw(filename, scan):
                    logger.info("Processing sdm %s, scan %d..." % (os.path.basename(filename), scan))

                    # 1) copy data into place
                    rtutils.rsync(filename, workdir)
                    filename = os.path.join(workdir, os.path.basename(filename))   # new full-path filename
                    assert 'mchammer' not in filename, 'filename %s is SDM original!'

                    # 2) find telcalfile (use timeout to wait for it to be written)
                    telcalfile = rtutils.gettelcalfile(telcaldir, filename, timeout=60)

                    # 3) submit search job and add tail job to monitoring queue
                    if telcalfile:
                        logger.info('Submitting job to rtutils.search with args: %s %s %s %s %s %s %s %s' % ('default', filename, self.rtparams, '', str([scan]), telcalfile, redishost, os.path.dirname(config.bdfLocation.rstrip('/'))))
                        lastjob = rtutils.search('default', filename, self.rtparams, '', [scan], telcalfile=telcalfile, redishost=redishost, bdfdir=os.path.dirname(config.bdfLocation.rstrip('/')))
                        rtutils.addjob(lastjob.id)
                    else:<commit_msg>Now does not submit scans with missing BDFs to queue<commit_after>#! /usr/bin/env python

# Main controller. <NAME> June 2015
#
# Based on frb_trigger_controller.py by <NAME>, 2015/02
#
# Listen for OBS packets having a certain 'triggered archiving'
# intent, and perform some as-yet-unspecified action when these
# are recieved.
#

import datetime
import os
import asyncore
import click
import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

from realfast import queue_monitor, rtutils, mcaf_library

# set up
rtparams_default = os.path.join(os.path.join(os.path.split(os.path.split(mcaf_library.__file__)[0])[0], 'conf'), 'rtpipe_cbe.conf') # install system puts conf files here. used by queue_rtpipe.py
telcaldir = '/home/mchammer/evladata/telcal'  # then yyyy/mm
workdir = os.getcwd()     # assuming we start in workdir
redishost = os.uname()[1]  # assuming we start on redis host

class FRBController(object):
    """Listens for OBS packets and tells FRB processing about any
    notable scans."""

    def __init__(self, intent='', project='', production=False, verbose=False, rtparams=''):
        # Mode can be project, intent
        self.intent = intent
        self.project = project
        self.production = production
        self.verbose = verbose
        self.rtparams = rtparams

    def add_sdminfo(self, sdminfo):
        config = mcaf_library.MCAST_Config(sdminfo=sdminfo)

        # !!! Wrapper here to deal with potential subscans?

        # Check if MCAST message is simply telling us the obs is finished
        if config.obsComplete:
            logger.info("Received finalMessage=True; This observation has completed.")
            # if completing the desired SB, then do a final rsync
            if self.project in config.projectID:
                logger.info("Final rsync to make workdir copy of SDM %sd complete." % (config.sdmLocation.rstrip('/')))
                rtutils.rsync(config.sdmLocation.rstrip('/'), workdir)  # final transfer to ensure complete SDM in workdir

        elif self.intent in config.intentString and self.project in config.projectID:
            logger.info("Scan %d has desired intent (%s) and project (%s)" % (config.scan, self.intent, self.project))
            logger.debug("BDF is in %s\n" % (config.bdfLocation))

            # If we're not in listening mode, prepare data and submit to queue system
            if self.production:
                filename = config.sdmLocation.rstrip('/')
                scan = int(config.scan)

                assert len(filename) and isinstance(filename, str), 'Filename empty or not a string?'

                # check that SDM is usable by rtpipe. Currently checks spw order and duplicates.
                if rtutils.check_spw(filename, scan):
                    logger.info("Processing sdm %s, scan %d..." % (os.path.basename(filename), scan))

                    # 1) copy data into place
                    rtutils.rsync(filename, workdir)
                    filename = os.path.join(workdir, os.path.basename(filename))   # new full-path filename
                    assert 'mchammer' not in filename, 'filename %s is SDM original!'

                    # 2) find telcalfile (use timeout to wait for it to be written)
                    telcalfile = rtutils.gettelcalfile(telcaldir, filename, timeout=60)

                    # 3) submit search job and add tail job to monitoring queue
                    if telcalfile:
                        if os.path.exists(config.bdfLocation):
                            logger.info('Submitting job to rtutils.search with args: %s %s %s %s %s %s %s %s' % ('default', filename, self.rtparams, '', str([scan]), telcalfile, redishost, os.path.dirname(config.bdfLocation.rstrip('/'))))
                            lastjob = rtutils.search('default', filename, self.rtparams, '', [scan], telcalfile=telcalfile, redishost=redishost, bdfdir=os.path.dirname(config.bdfLocation.rstrip('/')))
                            rtutils.addjob(lastjob.id)
                        else:
                            logger.info('***WARNING: BDF is missing for this scan. Will not submit to queue.')
                    else:
{'fraction_non_alphanumeric': 0.10477918935269208, 'fraction_numerical': 0.004355716878402904, 'mean_word_length': 2.546117546117546, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, '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': '5415333', 'n_tokens_mistral': 2413, 'n_tokens_neox': 2295, 'n_words': 731}