from app import db, logger, socketio from app.decorators import socketio_login_required from app.events import connected_sessions from app.models import Corpus, User from .cqi import CQiClient from flask import current_app, request from flask_login import current_user import math ''' ' A dictionary containing lists of, with corpus ids associated, Socket.IO ' session ids (sid). {: [, ...], ...} ''' corpus_analysis_sessions = {} ''' ' A dictionary containing Socket.IO session id - CQi client pairs. ' {: CQiClient, ...} ' A dictionary containing Socket.IO session id - and its status. ' {: 'running'|'idle'|'abort'} ''' corpus_analysis_clients = {} corpus_analysis_clients_status = {} @socketio.on('send_analysis_status_cli') @socketio_login_required def update_status(response): update_analysis_status(response['status']) @socketio.on('corpus_analysis_init') @socketio_login_required def init_corpus_analysis(corpus_id): logger.warning('Initiating corpus analysis.') socketio.start_background_task(corpus_analysis_session_handler, current_app._get_current_object(), corpus_id, current_user.id, request.sid) update_analysis_status('idle') @socketio.on('corpus_analysis_query') @socketio_login_required def corpus_analysis_query(query): update_analysis_status('running') logger.warning('Recieved a query.') logger.warning('Clients current session ID is: {}'.format(request.sid)) current_client_status = get_analysis_status() client = corpus_analysis_clients.get(request.sid) if client is None: socketio.emit('query', '[424]: Failed Dependency', room=request.sid) return corpus = client.corpora.get('CORPUS') try: results = corpus.query(query) except Exception as e: logger.warning(e) response = str(e) if response == 'CQI_CQP_ERROR_GENERAL': response = {'code': 1} socketio.emit('corpus_analysis_query', response, room=request.sid) else: chunk_size = 100 chunk_start = 0 context = 100 progress = 0 while chunk_start <= results.size: current_client_status = get_analysis_status() logger.warning('Status at while start: {}'.format(current_client_status)) if current_client_status == 'abort': update_analysis_status('idle') logger.warning('Aborting due to status being "abort".') break else: chunk = results.dump_values(context=context, first_result=chunk_start, num_results=chunk_size) if (results.size == 0): progress = 100 else: progress = ((chunk_start + chunk_size) / results.size) * 100 progress = min(100, int(math.ceil(progress))) socketio.emit('corpus_analysis_query', {'chunk': chunk, 'progress': progress, 'num_matches_total': results.size, 'code': 0}, room=request.sid) chunk_start += chunk_size current_client_status = get_analysis_status() if (progress == 100): update_analysis_status('idle') elif (current_client_status != 'abort'): update_analysis_status('running') logger.warning('Status at while end: {}'.format(get_analysis_status())) def update_analysis_status(status_str): # first: sets new status on server side session_id = request.sid corpus_analysis_clients_status[session_id] = status_str logger.warning('Set new status to "{}" on server side for {}.'.format(status_str, session_id)) # second: send new status to client side if status_str != 'abort': message = {'status': status_str} socketio.emit('send_analysis_status_server', message, room=session_id) logger.warning('Sending status "{}" to client.'.format(status_str)) def get_analysis_status(): status = corpus_analysis_clients_status[request.sid] logger.warning('Current status on server side is: {}'.format(status)) return status @socketio.on('inspect_match') @socketio_login_required def inspect_match(message): client = corpus_analysis_clients.get(request.sid) if client is None: socketio.emit('query', '[424]: Failed Dependency', room=request.sid) return # Get more context for given match CPOS match_context = client.get_sentences(message['cpos'], get_surrounding_s=True, l_r_s_context_additional_len=3) socketio.emit('match_context', match_context, room=request.sid) def corpus_analysis_session_handler(app, corpus_id, user_id, session_id): with app.app_context(): ''' Setup analysis session ''' corpus = Corpus.query.get(corpus_id) user = User.query.get(user_id) if corpus is None: logger.warning("ERROR 1") response = {'code': 404, 'msg': 'Not Found'} socketio.emit('corpus_analysis_init', response, room=session_id) return elif not (corpus.creator == user or user.is_administrator()): logger.warning("ERROR 2") response = {'code': 403, 'msg': 'Forbidden'} socketio.emit('corpus_analysis_init', response, room=session_id) return while corpus.status != 'analysing': db.session.refresh(corpus) socketio.sleep(3) client = CQiClient('corpus_{}_analysis'.format(corpus_id)) try: client.connect() except Exception: logger.warning("ERROR 3") response = {'code': 500, 'msg': 'Internal Server Error'} socketio.emit('corpus_analysis_init', response, room=session_id) return corpus_analysis_clients[session_id] = client if corpus_id not in corpus_analysis_sessions: corpus_analysis_sessions[corpus_id] = [session_id] else: corpus_analysis_sessions[corpus_id].append(session_id) response = {'code': 201, 'msg': 'Created'} logger.warning("ERROR 4") socketio.emit('corpus_analysis_init', response, room=session_id) ''' Observe analysis session ''' while session_id in connected_sessions: socketio.sleep(3) ''' Teardown analysis session ''' try: client.disconnect() except Exception: pass corpus_analysis_clients.pop(session_id, None) corpus_analysis_sessions[corpus_id].remove(session_id) if not corpus_analysis_sessions[corpus_id]: corpus_analysis_sessions.pop(corpus_id, None) corpus.status = 'stop analysis' db.session.commit()