Bug Fixes (Issue #453) (#476)

            warnings.warn('Use of filters and query_string are mutually exclusive.')

        with self.session_scope() as session:

            query = session.query(query_obj)

            # Now apply any filters that might be passed in.

            for attr, value in filters.items():

                query = query.filter(getattr(query_obj, attr)==value)

            # Execute the query to get the rows to be processed

            res = query.all()

            # Support either an SQL query string, or a simple dict based query

            if query_string:

                res = session.execute(query_string).fetchall()

                    query = query.filter(getattr(query_obj, attr)==value)

                # Execute the query to get the rows to be processed

                res = query.all()

                res = query.order_by(query_obj.id).all()

            if len(res) == 0:

                raise ValueError('Query returned zero results.')

                msg['config'] = self.config  # Hacky for now, just passs the whole config dict

                self.redis_queue.rpush(self.processing_queue,

                                    json.dumps(msg, cls=JsonEncoder))

            assert len(res) == self.queue_length

        return len(res)

    def apply(self, function, on='edge', args=(), walltime='01:00:00', chunksize=1000, filters={}, query_string='', **kwargs):

    def apply(self, function, on='edge', args=(), walltime='01:00:00', chunksize=1000, arraychunk=25, filters={}, query_string='', **kwargs):

        """

        A mirror of the apply function from the standard CandidateGraph object. This implementation

        dispatches the job to the cluster as an independent operation instead of applying an arbitrary function

                    This number may be have an actualy higher or lower limited based on

                    how the cluster has been configured.

        arraychunk : int

                     The number of concurrent jobs to run per job array. e.g. chunksize=100 and

                     arraychunk=25 gives the job array 1-100%25

        filters : dict

                  Of simple filters to apply on database rows where the key is the attribute and

                  the value used to check equivalency (e.g., attribute == value).

                     time=walltime,

                     partition=self.config['cluster']['queue'],

                     output=self.config['cluster']['cluster_log_dir']+f'/autocnet.{function}-%j')

        submitter.submit(array='1-{}%25'.format(job_counter), chunksize=chunksize)

        submitter.submit(array='1-{}%{}'.format(job_counter,arraychunk), chunksize=chunksize)

        return job_counter

    def generic_callback(self, msg):

        # Ingest isis control net as a df and do some massaging

        data = cnet.from_isis(path)

        data_to_update = data[['id', 'serialnumber', 'measureJigsawRejected', 'sampleResidual', 'lineResidual', 'samplesigma', 'linesigma', 'adjustedCovar', 'apriorisample', 'aprioriline']]

        data_to_update['adjustedCovar'] = data_to_update['adjustedCovar'].apply(lambda row : list(row))

        data_to_update['id'] = data_to_update['id'].apply(lambda row : int(row))

        data_to_update.loc[:,'adjustedCovar'] = data_to_update['adjustedCovar'].apply(lambda row : list(row))

        data_to_update.loc[:,'id'] = data_to_update['id'].apply(lambda row : int(row))

        # Generate a temp table, update the real table, then drop the temp table

        data_to_update.to_sql('temp_measures', self.engine, if_exists='replace', index_label='serialnumber', index = False)

        source_node = NetworkNode(node_id=sourceid, image_path=res.path)

        source_node.parent = ncg

        print(f'Attempting to subpixel register {len(measures)} measures for point {pointid}')

        print(f'Attempting to subpixel register {len(measures)-1} measures for point {pointid}')

        resultlog = []

        for measure in measures[1:]:

            destination_node.parent = ncg

            print('geom_match image:', res.path)

            try:

                new_x, new_y, dist, metric,  _ = geom_match(source_node.geodata, destination_node.geodata,

                                                        source.sample, source.line,

                                                        template_kwargs=subpixel_template_kwargs,

                                                        phase_kwargs=iterative_phase_kwargs, size_x=100, size_y=100)

                                                        phase_kwargs=iterative_phase_kwargs,

                                                        size_x=100, size_y=100)

            except Exception as e:

                print(f'geom_match failed on measure {measure.id} with exception -> {e}')

                measure.ignore = True

                currentlog['status'] = f"Failed to register measure {measure.id}"

                resultlog.append(currentlog)

                continue

            if new_x == None or new_y == None:

                measure.ignore = True # Unable to geom match

                currentlog['status'] = 'Failed to geom match.'

                resultlog.append(currentlog)

                continue

            cost = cost_func(dist, template_metric)

            if cost <= threshold:

                measure.ignore = True # Threshold criteria not met

                currentlog['status'] = f'Cost failed. Distance shifted: {dist}. Metric: {template_metric}.'

                resultlog.append(currentlog)

                continue

            if iterative_phase_kwargs:

                measure.template_metric = metric[0]

                measure.template_metric = metric

                measure.template_shift = dist

            cost = cost_func(measure.template_shift, measure.template_metric)

            if cost <= threshold:

                measure.ignore = True # Threshold criteria not met

                currentlog['status'] = f'Cost failed. Distance shifted: {measure.template_shift}. Metric: {measure.template_metric}.'

                resultlog.append(currentlog)

                continue

            # Update the measure

            measure.sample = new_x

            measure.line = new_y