KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _aggregate_count method in lisa

Best Python code snippet using lisa_python

inventory.py

Source:inventory.py

...21        'placeholders': devices.placeholders()22    }23    return jsonify(response)24class InventoryAggregation(Aggregation):25    def _aggregate_count(self, pipeline, value_if_none=0) -> int:26        response = self._aggregate(pipeline)27        try:28            return response[0]['count']29        except IndexError:30            return value_if_none31class InventoryDeviceAggregation(InventoryAggregation):32    FINAL_EVENTS = ['devices:Ready', 'devices:Dispose']33    def __init__(self):34        super().__init__('devices')35    def devices_not_fully_processed(self):36        pipeline = [37            {38                '$match': {39                    '@type': {'$nin': list(Component.types)},40                }41            }42        ]43        pipeline += self._not_containing_events(self.FINAL_EVENTS)44        pipeline += [45            {46                '$group': {47                    '_id': None,48                    'count': {'$sum': 1}49                }50            },51            {52                '$project': {53                    '_id': False,54                    'count': True55                }56            }57        ]58        return self._aggregate_count(pipeline)59    def devices_more_than_without_being_processed(self, more_than: timedelta = timedelta(weeks=1)):60        EVENTS = map_(DeviceEventDomain.GENERIC_TYPES, lambda _, key: Naming.new_type(key, 'devices'))61        EVENTS += ['devices:Add', 'devices:Migrate']62        pipeline = [63            {64                '$match': {65                    '@type': {'$nin': list(Component.types)},66                    '_created': {'$lte': datetime.today() - more_than}67                }68            },69        ]70        pipeline += self._not_containing_events(EVENTS)71        pipeline += [72            {73                '$group': {74                    '_id': None,75                    'count': {'$sum': 1}76                }77            },78            {79                '$project': {80                    '_id': False,81                    'count': True82                }83            }84        ]85        return self._aggregate_count(pipeline)86    def hard_drives_with_errors(self) -> int:87        pipeline = [88            {89                '$match': {90                    '@type': 'HardDrive'91                }92            },93            {94                '$project': {95                    'erased': {96                        '$filter': {97                            'input': '$events',98                            'as': 'event',99                            'cond': {'$in': ['$$event.@type', ['devices:EraseBasic', 'devices:EraseSectors']]}100                        }101                    },102                    'disposed': {103                        '$filter': {104                            'input': '$events',105                            'as': 'event',106                            'cond': {'$in': ['$$event.@type', ['devices:Dispose']]}107                        }108                    }109                }110            },111            {112                '$match': {113                    'erased': {'$gte': ['$size', 1]},114                    'disposed': {'$size': 0},115                }116            },117            {118                '$group': {119                    '_id': None,120                    'count': {'$sum': 1}121                }122            }123        ]124        return self._aggregate_count(pipeline)125    def placeholders(self) -> int:126        pipeline = [127            {128                '$match': {129                    'placeholder': True130                }131            },132            {133                '$group': {134                    '_id': None,135                    'count': {'$sum': 1}136                }137            }138        ]139        return self._aggregate_count(pipeline)140    @staticmethod141    def _not_containing_events(events_type: list):142        return [143            {144                '$project': {145                    'contains': {146                        '$filter': {147                            'input': '$events',148                            'as': 'event',149                            'cond': {'$in': ['$$event.@type', events_type]}150                        }151                    }152                }153            },...

get_ligation_efficiency.py

Source:get_ligation_efficiency.py

1from collections import Counter2import gzip3import operator4import pysam5import re6import sys7# Program for checking barcoding success rate8def main():9    lig = LigationEfficiency()10    lig.count_barcodes(sys.argv[1])11    lig.print_to_stdout()12class LigationEfficiency:13    def __init__(self):14        self._aggregate_count = Counter()15        self._position_count = Counter()16        self._pattern = re.compile('\[([a-zA-Z0-9_\-]+)\]')17        self._total = 018    def count_barcodes(self, filename):19        if filename.lower().endswith(".bam"):20            self.count_barcodes_in_bam_file(filename)21        elif filename.lower().endswith((".fastq", ".fq")):22            self.count_barcodes_in_fastq_file(filename)23        elif filename.lower().endswith((".fastq.gz", ".fq.gz")):24            self.count_barcodes_in_fastqgz_file(filename)25    def count_barcodes_in_bam_file(self, bamfile):26        with pysam.AlignmentFile(bamfile, "rb") as f:27            for read in f.fetch(until_eof=True):28                self.count_barcodes_in_name(read.query_name)29                self._total += 130    def count_barcodes_in_fastq_file(self, fastqfile):31        with open(fastqfile, "r") as f:32            for line in f:33                self.count_barcodes_in_name(line)34                next(f)35                next(f)36                next(f)37                self._total += 138    def count_barcodes_in_fastqgz_file(self, fastqgzfile):39        with gzip.open(fastqgzfile, "r") as f:40            for line in f:41                self.count_barcodes_in_name(line)42                next(f)43                next(f)44                next(f)45                self._total += 146    def count_barcodes_in_name(self, name):47        barcodes = self._pattern.findall(name)48        num_found = 049        pos = 050        for barcode in barcodes:51            if barcode != "NOT_FOUND":52                num_found += 153                self._position_count[pos] += 154            pos += 155        self._aggregate_count[num_found] += 156    def print_to_stdout(self):57        counts = sorted(self._aggregate_count.items(),58                        key=operator.itemgetter(0))59        for num_barcodes, count in counts:60            pct = "{0:.1f}%".format(100.0 * count / self._total)61            barcode = "barcode" if num_barcodes == 1 else "barcodes"62            print (str(count) + " (" + pct + ") reads found with " +63                str(num_barcodes) + " " + barcode + ".")64        print65        counts = sorted(self._position_count.items(),66                        key=operator.itemgetter(0))67        for position, count in counts:68            pct = "{0:.1f}%".format(100.0 * count / self._total)69            print (str(count) + " (" + pct + ") barcodes found in position " +70                str(position + 1) + ".")71if __name__ == "__main__":...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.