Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use is_zip_file method in localstack

Best Python code snippet using localstack_python

events_iterator.py

Source:events_iterator.py

1import pandas as pd2import numpy as np3from os.path import splitext4import zipfile5import events_loader6class Fixed_Events_Numbers_iterator:7    def __init__(self, event_file_path, num_events=10000, start_index=0):8        self.iterator = pd.read_csv(event_file_path, delim_whitespace=False, header=None,9                                    names=['x', 'y', 't', 'pol'],10                                    dtype={'x': np.int16, 'y': np.int16, 't': np.float64, 'pol': np.int16},11                                    engine='python',12                                    skiprows=start_index + 1, chunksize=num_events, nrows=None, memory_map=True)13    def __iter__(self):14     return self15    def __next__(self):16        event_window = self.iterator.__next__().values17        return event_window18class Fixed_Frame_Duration_iterator:19    def __init__(self, event_file_path, duration_ms=60.0, start_index=0):20        file_extension = splitext(event_file_path)[1]21        assert(file_extension in ['.txt', '.zip', '.csv'])22        self.is_zip_file = (file_extension == '.zip')23        self.is_txt_file = (file_extension == '.txt')24        self.is_csv_file = (file_extension == '.csv')25        if self.is_zip_file:26            self.zip_file = zipfile.ZipFile(event_file_path)27            files_in_archive = self.zip_file.namelist()28            assert(len(files_in_archive) == 1) 29            self.event_file = self.zip_file.open(files_in_archive[0], 'r')30        elif self.is_txt_file:31            self.event_file = open(event_file_path, 'r')32        elif self.is_csv_file:33            with open(event_file_path) as file:34                self.event_file = np.loadtxt(file, delimiter="," , skiprows=start_index + 1)35            36        for i in range(1 + start_index):37            if self.is_zip_file or self.is_txt_file:38                self.event_file.readline()39            #if self.is_csv_file:40            #    self.event_file = self.event_file[0:1 + start_index]41                42        self.last_stamp = None43        self.duration_s = duration_ms / 1000.044        self.duration_ms = duration_ms45    def __iter__(self):46     return self  47    def __del__(self):48        if self.is_zip_file:49            self.zip_file.close()50        elif self.is_txt_file:51            self.event_file.close()52    def __next__(self):53            event_list = []54            event_window = []55            for line in self.event_file:               56                if self.is_zip_file:57                    line = line.decode("utf-8")58                if self.is_zip_file or self.is_txt_file:59                 x, y, t, pol = line.split(' ')60                 x, y, t, pol = int(x), int(y), float(t), int(pol)61                else:62                 idx, x, y, t, pol = line63                event_list.append([x, y, t, pol])64                if self.last_stamp is None:65                    self.last_stamp = t66                if t < self.last_stamp + self.duration_ms:67                    #self.last_stamp = t68                    ew = np.array(event_list)69                if t >= self.last_stamp + self.duration_ms:70                    self.last_stamp = t71                    event_list=[]72                    event_list.append([x, y, t, pol])73                    event_window.append(ew)74            return event_window75    76class Fixed_Events_and_Duration_iterator:77    def __init__(self, event_file_path, num_events=100, duration_ms=200.0, start_index=0):78        file_extension = splitext(event_file_path)[1]79        assert(file_extension in ['.txt', '.zip', '.csv'])80        self.is_zip_file = (file_extension == '.zip')81        self.is_txt_file = (file_extension == '.txt')82        self.is_csv_file = (file_extension == '.csv')83        if self.is_zip_file:84            self.zip_file = zipfile.ZipFile(event_file_path)85            files_in_archive = self.zip_file.namelist()86            assert(len(files_in_archive) == 1) 87            self.event_file = self.zip_file.open(files_in_archive[0], 'r')88        elif self.is_txt_file:89            self.event_file = open(event_file_path, 'r')90        elif self.is_csv_file:91            with open(event_file_path) as file:92                self.event_file = np.loadtxt(file, delimiter="," , skiprows=start_index + 1)93            94        for i in range(1 + start_index):95            if self.is_zip_file or self.is_txt_file:96                self.event_file.readline()97            #if self.is_csv_file:98            #    self.event_file = self.event_file[0:1 + start_index]99                100        self.num_events = num_events101        self.last_stamp = None102        self.duration_s = duration_ms / 1000.0103        self.duration_ms = duration_ms104    def __iter__(self):105     return self  106    def __del__(self):107        if self.is_zip_file:108            self.zip_file.close()109        elif self.is_txt_file:110            self.event_file.close()111    def __next__(self):112            event_list = []113            event_window = []114            for line in self.event_file:               115                if self.is_zip_file:116                    line = line.decode("utf-8")117                if self.is_zip_file or self.is_txt_file:118                 x, y, t, pol = line.split(' ')119                 x, y, t, pol = int(x), int(y), float(t), int(pol)120                else:121                 idx, x, y, t, pol = line122                event_list.append([x, y, t, pol])123                if self.last_stamp is None:124                    self.last_stamp = t125                if t < self.last_stamp + self.duration_ms:126                    #self.last_stamp = t127                    ew = np.array(event_list)128                if t >= self.last_stamp + self.duration_ms:129                    if ew.shape[0] < self.num_events:130                        ew = np.array(event_list)131                        continue132                    else:133                     self.last_stamp = t134                     event_list=[]135                     event_list.append([x, y, t, pol])136                     event_window.append(ew)137            return event_window...

event_readers.py

Source:event_readers.py

1import pandas as pd2import zipfile3from os.path import splitext4import numpy as np5from .timers import Timer6class FixedSizeEventReader:7    """8    Reads events from a '.txt' or '.zip' file, and packages the events into9    non-overlapping event windows, each containing a fixed number of events.10    """11    def __init__(self, path_to_event_file, num_events=10000, start_index=0):12        print('Will use fixed size event windows with {} events'.format(num_events))13        print('Output frame rate: variable')14        self.iterator = pd.read_csv(path_to_event_file, delim_whitespace=True, header=None,15                                    names=['t', 'x', 'y', 'pol'],16                                    dtype={'t': np.float64, 'x': np.int16, 'y': np.int16, 'pol': np.int16},17                                    engine='c',18                                    skiprows=start_index + 1, chunksize=num_events, nrows=None, memory_map=True)19    def __iter__(self):20        return self21    def __next__(self):22        with Timer('Reading event window from file'):23            event_window = self.iterator.__next__().values24        return event_window25class FixedDurationEventReader:26    """27    Reads events from a '.txt' or '.zip' file, and packages the events into28    non-overlapping event windows, each of a fixed duration.29    **Note**: This reader is much slower than the FixedSizeEventReader.30              The reason is that the latter can use Pandas' very efficient cunk-based reading scheme implemented in C.31    """32    def __init__(self, path_to_event_file, duration_ms=50.0, start_index=0):33        print('Will use fixed duration event windows of size {:.2f} ms'.format(duration_ms))34        print('Output frame rate: {:.1f} Hz'.format(1000.0 / duration_ms))35        file_extension = splitext(path_to_event_file)[1]36        assert(file_extension in ['.txt', '.zip', '.csv'])37        self.is_zip_file = (file_extension == '.zip')38        self.is_csv_file = (file_extension == '.csv')39        if self.is_zip_file:  # '.zip'40            self.zip_file = zipfile.ZipFile(path_to_event_file)41            files_in_archive = self.zip_file.namelist()42            assert(len(files_in_archive) == 1)  # make sure there is only one text file in the archive43            self.event_file = self.zip_file.open(files_in_archive[0], 'r')44        else:45            self.event_file = open(path_to_event_file, 'r')46        # ignore header + the first start_index lines47        for i in range(1 + start_index):48            self.event_file.readline()49        self.last_stamp = None50        self.duration_s = duration_ms / 1000.051    def __iter__(self):52        return self53    def __del__(self):54        if self.is_zip_file:55            self.zip_file.close()56        self.event_file.close()57    def __next__(self):58        with Timer('Reading event window from file'):59            event_list = []60            for line in self.event_file:61                if self.is_zip_file:62                    line = line.decode("utf-8")63                if self.is_csv_file:64                    t, x, y, pol = line.split(',')65                else:66                    t, x, y, pol = line.split(' ')67                if self.is_csv_file:68                    # Temp workaround -> because in my .csv files timestamp is not . separated -> it is uint, not float69                    t, x, y, pol = float(t) / 1000000.0, int(x), int(y), int(pol)70                else:71                    t, x, y, pol = float(t), int(x), int(y), int(pol)72                event_list.append([t, x, y, pol])73                if self.last_stamp is None:74                    self.last_stamp = t75                if t > self.last_stamp + self.duration_s:76                    self.last_stamp = t77                    event_window = np.array(event_list)78                    return event_window...

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.