<commit_before>from weathernetwork.server.interface import IServerSideProxy
from weathernetwork.common.fileformats import PCWetterstationFormatFile
from weathernetwork.server.weathermessage import WeatherMessage
from weathernetwork.common.delayedexception import DelayedException
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import os
from zipfile import ZipFile
import time
import datetime
from datetime import timedelta
import threading
from multiprocessing import Event, Process, Queue
import signal
import glob

class FileSystemObserver(FileSystemEventHandler):
    def __init__(self, data_directory):
        self._data_directory = data_directory
        self._received_file_queue = []
        self._exception_event = Event()
        self._processed_files_lock = []
        self._processed_files = dict()


    def stop(self):
        self._exception_event.set()


    def set_received_file_queue(self, received_file_queue):
        self._received_file_queue = received_file_queue


    def _remove_outdated_processed_files(self, curr_time, max_keeping_time = 10):
        """Removes too old processed message IDs from the list."""
        removal_list = []
        for file in self._processed_files.keys():
            if (curr_time - self._processed_files[file]) > timedelta(seconds = max_keeping_time):
                removal_list.append(file)

        for item in removal_list:
            del self._processed_files[item]


    def on_modified(self, event):
        """A file has been modified"""
        if not event.is_directory:
            # ensure that this file has not yet been processed (workaround for the watchdog library reporting possibly multiple "modified" events for a single file)
            full_path = event.src_path              
            curr_time = datetime.datetime.utcnow()   
            with self._processed_files_lock:
                self._remove_outdated_processed_files(curr_time)                
                already_processed = full_path in self._processed_files
                self._processed_files[full_path] = curr_time

            if not already_processed:
                self._received_file_queue.put(full_path)


    def process(self, received_file_queue, exception_handler):
        try:
            signal.signal(signal.SIGINT, signal.SIG_IGN) # required for correct handling of Crtl + C in a multiprocess environment
            self._processed_files_lock = threading.Lock()
            filesystem_observer = Observer()
            self._received_file_queue = received_file_queue
        
            filesystem_observer.schedule(self, self._data_directory, recursive=False)
            filesystem_observer.start()
        
            # wait until stop is signalled
            self._exception_event.wait()

            filesystem_observer.stop()
            filesystem_observer.join()
        except Exception as e:
            exception_handler(DelayedException(e))


    def feed_modified_file(self, full_path):
        """Feeds a data file manually into the queue of received files."""
        self._received_file_queue.put(full_path)


class FTPServerBrokerProcess(object):
    def __init__(self, data_directory, data_file_extension, temp_data_directory):
        self._data_directory = data_directory        
        self._data_file_extension = data_file_extension
        self._temp_data_directory = temp_data_directory


    def process(self, received_file_queue, request_queue, exception_handler):
        try:
            signal.signal(signal.SIGINT, signal.SIG_IGN)
            while (True):
                full_path = received_file_queue.get()
                if full_path is None:
                    break;
                message_ID, station_ID, data = self._on_new_data(full_path)
                if message_ID:
                    request_queue.put((message_ID, station_ID, data))
        except Exception as e:
            exception_handler(DelayedException(e))


    def _on_new_data(self, full_path):
        """A new file has been received via FTP (i.e. modified)"""
        # extract all required information from the name of the transferred data file
        file, file_extension = os.path.splitext(full_path)
        file_path, message_ID = os.path.split(file)

        if file_extension.upper() == self._data_file_extension:
            splitted_message_ID = message_ID.split( '_' )
            if len(splitted_message_ID) > 0:
                station_ID = splitted_message_ID[-1]
                data = self._read_data_files(message_ID, station_ID)
                return message_ID, station_ID, data

        return None, None, None


    def _read_data_files(self, message_ID, station_ID):
        wait_time = 0.05    # wait time between data file reading trials (in seconds)
        max_trials = 10     # reading trials for the data file before a fatal failure is assumed

        new_data_file_list = []
        try:
            # extract data from the ZIP-file(s)
            file_still_blocked = True
            counter = 0
            while file_still_blocked:
                try:
                    zip_file = ZipFile(self._data_directory + '/' + message_ID + self._data_file_extension, 'r')
                except PermissionError:
                    # workaround because the watchdog library cannot monitor the file close event and the file may still be open when the "modified" event is signalled
                    counter += 1
                    if counter > max_trials:
                        reraise
                    time.sleep(wait_time)
                else:
                    file_still_blocked = False
                    with zip_file:
                        new_data_file_list = zip_file.namelist()
                        zip_file.extractall(self._temp_data_directory)

            # read a list of WeatherData objects from the unzipped data files  
            for curr_file_name in new_data_file_list:
                file = PCWetterstationFormatFile(self._temp_data_directory, curr_file_name) # TODO: the data from different files needs to be merged
                data = file.read()
                data = data[0]
        finally:
            # delete the temporary data files
            PCWetterstationFormatFile.deletedatafiles(self._temp_data_directory, new_data_file_list)

        return data


class FTPBroker(object):
    """Broker for weather data transmission based on FTP"""

    def __init__(self, request_queue, data_directory, data_file_extension, temp_data_directory, exception_handler):<commit_msg>Implemented correct reading of ZIP-files containing multiple data files.<commit_after>from weathernetwork.server.interface import IServerSideProxy
from weathernetwork.common.fileformats import PCWetterstationFormatFile
from weathernetwork.server.weathermessage import WeatherMessage
from weathernetwork.common.delayedexception import DelayedException
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import os
from zipfile import ZipFile
import time
import datetime
from datetime import timedelta
import threading
from multiprocessing import Event, Process, Queue
import signal
import glob

class FileSystemObserver(FileSystemEventHandler):
    def __init__(self, data_directory):
        self._data_directory = data_directory
        self._received_file_queue = []
        self._exception_event = Event()
        self._processed_files_lock = []
        self._processed_files = dict()


    def stop(self):
        self._exception_event.set()


    def set_received_file_queue(self, received_file_queue):
        self._received_file_queue = received_file_queue


    def _remove_outdated_processed_files(self, curr_time, max_keeping_time = 10):
        """Removes too old processed message IDs from the list."""
        removal_list = []
        for file in self._processed_files.keys():
            if (curr_time - self._processed_files[file]) > timedelta(seconds = max_keeping_time):
                removal_list.append(file)

        for item in removal_list:
            del self._processed_files[item]


    def on_modified(self, event):
        """A file has been modified"""
        if not event.is_directory:
            # ensure that this file has not yet been processed (workaround for the watchdog library reporting possibly multiple "modified" events for a single file)
            full_path = event.src_path              
            curr_time = datetime.datetime.utcnow()   
            with self._processed_files_lock:
                self._remove_outdated_processed_files(curr_time)                
                already_processed = full_path in self._processed_files
                self._processed_files[full_path] = curr_time

            if not already_processed:
                self._received_file_queue.put(full_path)


    def process(self, received_file_queue, exception_handler):
        try:
            signal.signal(signal.SIGINT, signal.SIG_IGN) # required for correct handling of Crtl + C in a multiprocess environment
            self._processed_files_lock = threading.Lock()
            filesystem_observer = Observer()
            self._received_file_queue = received_file_queue
        
            filesystem_observer.schedule(self, self._data_directory, recursive=False)
            filesystem_observer.start()
        
            # wait until stop is signalled
            self._exception_event.wait()

            filesystem_observer.stop()
            filesystem_observer.join()
        except Exception as e:
            exception_handler(DelayedException(e))


    def feed_modified_file(self, full_path):
        """Feeds a data file manually into the queue of received files."""
        self._received_file_queue.put(full_path)


class FTPServerBrokerProcess(object):
    def __init__(self, data_directory, data_file_extension, temp_data_directory):
        self._data_directory = data_directory        
        self._data_file_extension = data_file_extension
        self._temp_data_directory = temp_data_directory


    def process(self, received_file_queue, request_queue, exception_handler):
        try:
            signal.signal(signal.SIGINT, signal.SIG_IGN)
            while (True):
                full_path = received_file_queue.get()
                if full_path is None:
                    break;
                message_ID, station_ID, data = self._on_new_data(full_path)
                if message_ID:
                    request_queue.put((message_ID, station_ID, data))
        except Exception as e:
            exception_handler(DelayedException(e))


    def _on_new_data(self, full_path):
        """A new file has been received via FTP (i.e. modified)"""
        # extract all required information from the name of the transferred data file
        file, file_extension = os.path.splitext(full_path)
        file_path, message_ID = os.path.split(file)

        if file_extension.upper() == self._data_file_extension:
            splitted_message_ID = message_ID.split( '_' )
            if len(splitted_message_ID) > 0:
                station_ID = splitted_message_ID[-1]
                data = self._read_data_files(message_ID, station_ID)
                return message_ID, station_ID, data

        return None, None, None


    def _read_data_files(self, message_ID, station_ID):
        wait_time = 0.05    # wait time between data file reading trials (in seconds)
        max_trials = 10     # reading trials for the data file before a fatal failure is assumed

        new_data_file_list = []
        try:
            # extract data from the ZIP-file(s)
            file_still_blocked = True
            counter = 0
            while file_still_blocked:
                try:
                    zip_file = ZipFile(self._data_directory + '/' + message_ID + self._data_file_extension, 'r')
                except PermissionError:
                    # workaround because the watchdog library cannot monitor the file close event and the file may still be open when the "modified" event is signalled
                    counter += 1
                    if counter > max_trials:
                        reraise
                    time.sleep(wait_time)
                else:
                    file_still_blocked = False
                    with zip_file:
                        new_data_file_list = zip_file.namelist()
                        zip_file.extractall(self._temp_data_directory)

            # read a list of WeatherData objects from the unzipped data files
            data = list()
            for curr_file_name in new_data_file_list:
                file = PCWetterstationFormatFile(self._temp_data_directory, curr_file_name)
                curr_data = file.read()
                data += curr_data[0]
        finally:
            # delete the temporary data files
            PCWetterstationFormatFile.deletedatafiles(self._temp_data_directory, new_data_file_list)

        return data


class FTPBroker(object):
    """Broker for weather data transmission based on FTP"""

    def __init__(self, request_queue, data_directory, data_file_extension, temp_data_directory, exception_handler):
{'fraction_non_alphanumeric': 0.06237226553629551, 'fraction_numerical': 0.001816667928241617, 'mean_word_length': 2.3990223822999743, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, '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': '6882365', 'n_tokens_mistral': 3580, 'n_tokens_neox': 3346, 'n_words': 907}