sporada/nopaque
1
0
Fork 0
mirror of https://gitlab.ub.uni-bielefeld.de/sfb1288inf/nopaque.git synced 2025-06-16 11:00:41 +00:00
Code Issues Projects Releases Wiki Activity

modernize type hinting

Browse Source
This commit is contained in:
Patrick Jentsch
2024-09-25 17:46:53 +02:00
parent 02e6c7c16c
commit 492fdc9d28
13 changed files with 116 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
app/corpora/cqi_over_sio/__init__.py
View File

@ -7,7 +7,7 @@ from flask_login import current_user
from flask_socketio import Namespace
from inspect import signature
from threading import Lock
from typing import Callable, Dict, List, Optional
from typing import Callable
from app import db, docker_client, hashids, socketio
from app.decorators import socketio_login_required
from app.models import Corpus, CorpusStatus
@ -37,7 +37,8 @@ Basic concept:
4.2 The CQiClient and (Mutex) Lock belonging to it are teared down.
'''
CQI_API_FUNCTION_NAMES: List[str] = [
CQI_API_FUNCTION_NAMES: list[str] = [
'ask_feature_cl_2_3',
'ask_feature_cqi_1_0',
'ask_feature_cqp_2_3',
@ -93,7 +94,7 @@ class CQiOverSocketIO(Namespace):
@socketio_login_required
def on_init(self, db_corpus_hashid: str):
db_corpus_id: int = hashids.decode(db_corpus_hashid)
db_corpus: Optional[Corpus] = Corpus.query.get(db_corpus_id)
db_corpus: Corpus | None = Corpus.query.get(db_corpus_id)
if db_corpus is None:
return {'code': 404, 'msg': 'Not Found'}
if not (db_corpus.user == current_user
@ -134,7 +135,7 @@ class CQiOverSocketIO(Namespace):
return {'code': 200, 'msg': 'OK'}
@socketio_login_required
def on_exec(self, fn_name: str, fn_args: Dict = {}):
def on_exec(self, fn_name: str, fn_args: dict = {}):
try:
cqi_client: CQiClient = session['cqi_over_sio']['cqi_client']
cqi_client_lock: Lock = session['cqi_over_sio']['cqi_client_lock']
@ -198,7 +199,7 @@ class CQiOverSocketIO(Namespace):
except (BrokenPipeError, CQiException):
pass
cqi_client_lock.release()
db_corpus: Optional[Corpus] = Corpus.query.get(db_corpus_id)
db_corpus: Corpus | None = Corpus.query.get(db_corpus_id)
if db_corpus is None:
return
db_corpus.num_analysis_sessions = Corpus.num_analysis_sessions - 1

33
app/corpora/cqi_over_sio/extensions.py
View File

@ -8,7 +8,6 @@ from cqi.models.attributes import (
)
from cqi.status import StatusOk as CQiStatusOk
from flask import session
from typing import Dict, List
import gzip
import json
import math
@ -17,7 +16,7 @@ from app.models import Corpus
from .utils import lookups_by_cpos, partial_export_subcorpus, export_subcorpus
CQI_EXTENSION_FUNCTION_NAMES: List[str] = [
CQI_EXTENSION_FUNCTION_NAMES: list[str] = [
'ext_corpus_update_db',
'ext_corpus_static_data',
'ext_corpus_paginate_corpus',
@ -37,7 +36,7 @@ def ext_corpus_update_db(corpus: str) -> CQiStatusOk:
return CQiStatusOk()
def ext_corpus_static_data(corpus: str) -> Dict:
def ext_corpus_static_data(corpus: str) -> dict:
db_corpus_id: int = session['cqi_over_sio']['db_corpus_id']
db_corpus: Corpus = Corpus.query.get(db_corpus_id)
@ -48,8 +47,8 @@ def ext_corpus_static_data(corpus: str) -> Dict:
cqi_client: CQiClient = session['cqi_over_sio']['cqi_client']
cqi_corpus: CQiCorpus = cqi_client.corpora.get(corpus)
cqi_p_attrs: List[CQiPositionalAttribute] = cqi_corpus.positional_attributes.list()
cqi_s_attrs: List[CQiStructuralAttribute] = cqi_corpus.structural_attributes.list()
cqi_p_attrs: list[CQiPositionalAttribute] = cqi_corpus.positional_attributes.list()
cqi_s_attrs: list[CQiStructuralAttribute] = cqi_corpus.structural_attributes.list()
static_data = {
'corpus': {
@ -64,19 +63,19 @@ def ext_corpus_static_data(corpus: str) -> Dict:
for p_attr in cqi_p_attrs:
print(f'corpus.freqs.{p_attr.name}')
static_data['corpus']['freqs'][p_attr.name] = []
p_attr_id_list: List[int] = list(range(p_attr.lexicon_size))
p_attr_id_list: list[int] = list(range(p_attr.lexicon_size))
static_data['corpus']['freqs'][p_attr.name].extend(p_attr.freqs_by_ids(p_attr_id_list))
del p_attr_id_list
print(f'p_attrs.{p_attr.name}')
static_data['p_attrs'][p_attr.name] = []
cpos_list: List[int] = list(range(cqi_corpus.size))
cpos_list: list[int] = list(range(cqi_corpus.size))
static_data['p_attrs'][p_attr.name].extend(p_attr.ids_by_cpos(cpos_list))
del cpos_list
print(f'values.p_attrs.{p_attr.name}')
static_data['values']['p_attrs'][p_attr.name] = []
p_attr_id_list: List[int] = list(range(p_attr.lexicon_size))
p_attr_id_list: list[int] = list(range(p_attr.lexicon_size))
static_data['values']['p_attrs'][p_attr.name].extend(p_attr.values_by_ids(p_attr_id_list))
del p_attr_id_list
@ -127,23 +126,23 @@ def ext_corpus_static_data(corpus: str) -> Dict:
print(f's_attrs.{s_attr.name}.lexicon.{id}.bounds')
static_data['s_attrs'][s_attr.name]['lexicon'][id]['bounds'] = [lbound, rbound]
static_data['s_attrs'][s_attr.name]['lexicon'][id]['freqs'] = {}
cpos_list: List[int] = list(range(lbound, rbound + 1))
cpos_list: list[int] = list(range(lbound, rbound + 1))
for p_attr in cqi_p_attrs:
p_attr_ids: List[int] = []
p_attr_ids: list[int] = []
p_attr_ids.extend(p_attr.ids_by_cpos(cpos_list))
print(f's_attrs.{s_attr.name}.lexicon.{id}.freqs.{p_attr.name}')
static_data['s_attrs'][s_attr.name]['lexicon'][id]['freqs'][p_attr.name] = dict(Counter(p_attr_ids))
del p_attr_ids
del cpos_list
sub_s_attrs: List[CQiStructuralAttribute] = cqi_corpus.structural_attributes.list(filters={'part_of': s_attr})
sub_s_attrs: list[CQiStructuralAttribute] = cqi_corpus.structural_attributes.list(filters={'part_of': s_attr})
print(f's_attrs.{s_attr.name}.values')
static_data['s_attrs'][s_attr.name]['values'] = [
sub_s_attr.name[(len(s_attr.name) + 1):]
for sub_s_attr in sub_s_attrs
]
s_attr_id_list: List[int] = list(range(s_attr.size))
sub_s_attr_values: List[str] = []
s_attr_id_list: list[int] = list(range(s_attr.size))
sub_s_attr_values: list[str] = []
for sub_s_attr in sub_s_attrs:
tmp = []
tmp.extend(sub_s_attr.values_by_ids(s_attr_id_list))
@ -173,7 +172,7 @@ def ext_corpus_paginate_corpus(
corpus: str,
page: int = 1,
per_page: int = 20
) -> Dict:
) -> dict:
cqi_client: CQiClient = session['cqi_over_sio']['cqi_client']
cqi_corpus = cqi_client.corpora.get(corpus)
# Sanity checks
@ -219,7 +218,7 @@ def ext_cqp_paginate_subcorpus(
context: int = 50,
page: int = 1,
per_page: int = 20
) -> Dict:
) -> dict:
corpus_name, subcorpus_name = subcorpus.split(':', 1)
cqi_client: CQiClient = session['cqi_over_sio']['cqi_client']
cqi_corpus = cqi_client.corpora.get(corpus_name)
@ -266,7 +265,7 @@ def ext_cqp_partial_export_subcorpus(
subcorpus: str,
match_id_list: list,
context: int = 50
) -> Dict:
) -> dict:
corpus_name, subcorpus_name = subcorpus.split(':', 1)
cqi_client: CQiClient = session['cqi_over_sio']['cqi_client']
cqi_corpus = cqi_client.corpora.get(corpus_name)
@ -278,7 +277,7 @@ def ext_cqp_partial_export_subcorpus(
def ext_cqp_export_subcorpus(
subcorpus: str,
context: int = 50
) -> Dict:
) -> dict:
corpus_name, subcorpus_name = subcorpus.split(':', 1)
cqi_client: CQiClient = session['cqi_over_sio']['cqi_client']
cqi_corpus = cqi_client.corpora.get(corpus_name)

13
app/corpora/cqi_over_sio/utils.py
View File

@ -1,13 +1,12 @@
from cqi.models.corpora import Corpus as CQiCorpus
from cqi.models.subcorpora import Subcorpus as CQiSubcorpus
from typing import Dict, List
def lookups_by_cpos(corpus: CQiCorpus, cpos_list: List[int]) -> Dict:
def lookups_by_cpos(corpus: CQiCorpus, cpos_list: list[int]) -> dict:
lookups = {}
lookups['cpos_lookup'] = {cpos: {} for cpos in cpos_list}
for attr in corpus.positional_attributes.list():
cpos_attr_values: List[str] = attr.values_by_cpos(cpos_list)
cpos_attr_values: list[str] = attr.values_by_cpos(cpos_list)
for i, cpos in enumerate(cpos_list):
lookups['cpos_lookup'][cpos][attr.name] = cpos_attr_values[i]
for attr in corpus.structural_attributes.list():
@ -15,7 +14,7 @@ def lookups_by_cpos(corpus: CQiCorpus, cpos_list: List[int]) -> Dict:
# attr.has_values == False
if attr.has_values:
continue
cpos_attr_ids: List[int] = attr.ids_by_cpos(cpos_list)
cpos_attr_ids: list[int] = attr.ids_by_cpos(cpos_list)
for i, cpos in enumerate(cpos_list):
if cpos_attr_ids[i] == -1:
continue
@ -39,9 +38,9 @@ def lookups_by_cpos(corpus: CQiCorpus, cpos_list: List[int]) -> Dict:
def partial_export_subcorpus(
subcorpus: CQiSubcorpus,
match_id_list: List[int],
match_id_list: list[int],
context: int = 25
) -> Dict:
) -> dict:
if subcorpus.size == 0:
return {"matches": []}
match_boundaries = []
@ -91,7 +90,7 @@ def export_subcorpus(
context: int = 25,
cutoff: float = float('inf'),
offset: int = 0
) -> Dict:
) -> dict:
if subcorpus.size == 0:
return {"matches": []}
first_match = max(0, offset)