Commit fe4ca88a authored by Konstantin Schulz's avatar Konstantin Schulz

moved additional classes to the OpenAPI specification

parent 39e4f5d5
Pipeline #11662 passed with stages
in 2 minutes and 52 seconds
......@@ -7,6 +7,7 @@ omit =
*/site-packages/*
*/migrations/*
# cannot run tests for files that are generated and updated automatically
*/openapi/*
*/models_auto.py
parallel = True
......
......@@ -43,6 +43,10 @@ To autogenerate a new migration script:
----------------------------------------------------------------
# Models
To generate class structures for this project automatically:
1. Install OpenAPI Generator (using, e.g., `brew install openapi-generator`).
2. Run: `openapi-generator generate -i ./mcserver/mcserver_api.yaml -g python-flask -o ./openapi/ && python openapi_generator.py`.
# Testing
To check the coverage of the current tests, run
`coverage run --rcfile=.coveragerc tests.py && coverage combine && coverage report -m`.
......@@ -36,9 +36,9 @@ class CorpusStorageManagerAPI(Resource):
args: Dict = flask.request.args
cts_urn: str = args["urn"]
ar: AnnisResponse = CorpusService.get_corpus(cts_urn=cts_urn, is_csm=True)
if not ar.nodes:
if not ar.graph_data.nodes:
abort(404)
return NetworkService.make_json_response(ar.__dict__)
return NetworkService.make_json_response(ar.to_dict())
def post(self):
"""Given the relevant corpus data, gives back search results as graph data."""
......
import json
from typing import Dict, List
import flask
from flask_restful import Resource
from flask_restful.reqparse import RequestParser
from mcserver.app.models import ExerciseData, GraphData, Solution, SolutionElement, AnnisResponse
from mcserver.app.models import ExerciseData, GraphData, Solution, AnnisResponse, make_solution_element_from_salt_id
from mcserver.app.services import CorpusService, AnnotationService, NetworkService
......@@ -30,7 +27,7 @@ class SubgraphAPI(Resource):
ctx_left: int = int(args["ctx_left"])
ctx_right: int = int(args["ctx_right"])
ar: AnnisResponse = CorpusService.get_subgraph(urn, aql, ctx_left, ctx_right, is_csm=True)
return NetworkService.make_json_response(ar.__dict__)
return NetworkService.make_json_response(ar.to_dict())
def post(self):
""" Returns subgraph data for a given CTS URN and AQL. """
......@@ -45,9 +42,9 @@ class SubgraphAPI(Resource):
for aql in aqls:
node_ids: List[str] = CorpusService.find_matches(cts_urn, aql, is_csm=True)
for node_id in node_ids:
gd: GraphData = AnnotationService.get_single_subgraph(disk_urn, [node_id], ctx_left, ctx_right,
is_csm=True)
exercise_data_list.append(ExerciseData(graph=gd, uri="",
solutions=[Solution(target=SolutionElement(salt_id=node_id))]))
gd: GraphData = AnnotationService.get_single_subgraph(
disk_urn, [node_id], ctx_left, ctx_right, is_csm=True)
exercise_data_list.append(ExerciseData(
graph=gd, uri="", solutions=[Solution(target=make_solution_element_from_salt_id(node_id))]))
ret_val: List[dict] = [x.serialize() for x in exercise_data_list]
return NetworkService.make_json_response(ret_val)
......@@ -24,7 +24,6 @@ class TextComplexityAPI(Resource):
urn: str = args["urn"]
measure: str = args["measure"]
ar_dict: dict = args.get("annis_response", None)
ar: AnnisResponse = AnnisResponse(json_dict=ar_dict) if ar_dict else CorpusService.get_corpus(urn, is_csm=True)
gd: GraphData = GraphData(json_dict=ar.__dict__)
tc: TextComplexity = TextComplexityService.text_complexity(measure, urn, True, gd)
return NetworkService.make_json_response(tc.__dict__)
ar: AnnisResponse = AnnisResponse.from_dict(ar_dict) if ar_dict else CorpusService.get_corpus(urn, is_csm=True)
tc: TextComplexity = TextComplexityService.text_complexity(measure, urn, True, ar.graph_data)
return NetworkService.make_json_response(tc.to_dict())
......@@ -5,23 +5,24 @@ import sys
from logging.handlers import RotatingFileHandler
from threading import Thread
from time import strftime
from typing import Type, List
from typing import Type
import connexion
import flask
import open_alchemy
from connexion import FlaskApp
from flask import Flask, got_request_exception, request, Response, send_from_directory
from flask_cors import CORS
from flask_migrate import Migrate
from flask_sqlalchemy import SQLAlchemy
from open_alchemy import init_yaml
from mcserver.config import Config
db: SQLAlchemy = SQLAlchemy() # session_options={"autocommit": True}
migrate: Migrate = Migrate(directory=Config.MIGRATIONS_DIRECTORY)
# do this _BEFORE_ you add any APIs to your application
init_yaml(Config.API_SPEC_FILE_PATH, base=db.Model,
models_filename=os.path.join(Config.MC_SERVER_DIRECTORY, "models_auto.py"))
if not hasattr(open_alchemy.models, Config.DATABASE_TABLE_CORPUS):
# do this _BEFORE_ you add any APIs to your application
init_yaml(Config.API_SPEC_YAML_FILE_PATH, base=db.Model,
models_filename=os.path.join(Config.MC_SERVER_DIRECTORY, "models_auto.py"))
def apply_event_handlers(app: FlaskApp):
......@@ -78,7 +79,7 @@ def init_app_common(cfg: Type[Config] = Config, is_csm: bool = False) -> Flask:
connexion_app: FlaskApp = connexion.FlaskApp(
__name__, port=(cfg.CORPUS_STORAGE_MANAGER_PORT if is_csm else cfg.HOST_PORT),
specification_dir=Config.MC_SERVER_DIRECTORY)
connexion_app.add_api(Config.API_SPEC_FILE_PATH, arguments={'title': 'Machina Callida Backend REST API'})
connexion_app.add_api(Config.API_SPEC_YAML_FILE_PATH, arguments={'title': 'Machina Callida Backend REST API'})
apply_event_handlers(connexion_app)
app: Flask = connexion_app.app
# allow CORS requests for all API routes
......@@ -87,11 +88,11 @@ def init_app_common(cfg: Type[Config] = Config, is_csm: bool = False) -> Flask:
app.app_context().push()
db.init_app(app)
migrate.init_app(app, db)
if is_csm or cfg.TESTING:
db.create_all()
if is_csm:
from mcserver.app.services.databaseService import DatabaseService
DatabaseService.init_db_alembic()
if is_csm or cfg.TESTING:
db.create_all()
from mcserver.app.services.textService import TextService
TextService.init_proper_nouns_list()
TextService.init_stop_words_latin()
......@@ -119,7 +120,6 @@ def log_exception(sender_app: Flask, exception, **extra):
exception -- the exception to be logged
**extra -- any additional arguments
"""
# TODO: RETURN ERROR IN JSON FORMAT
sender_app.logger.exception(f"ERROR for {flask.request.url}")
......
......@@ -8,7 +8,7 @@ from connexion.lifecycle import ConnexionResponse
from flask import Response
from mcserver.app import db
from mcserver.app.models import ExerciseType, Solution, ExerciseData, AnnisResponse, Phenomenon, TextComplexity, \
TextComplexityMeasure, ResourceType, ExerciseMC
TextComplexityMeasure, ResourceType, ExerciseMC, GraphData
from mcserver.app.services import AnnotationService, CorpusService, NetworkService, TextComplexityService
from mcserver.config import Config
from mcserver.models_auto import Exercise, TExercise, UpdateInfo
......@@ -31,7 +31,7 @@ def get(eid: str) -> Union[Response, ConnexionResponse]:
if exercise is None:
return connexion.problem(404, Config.ERROR_TITLE_NOT_FOUND, Config.ERROR_MESSAGE_EXERCISE_NOT_FOUND)
ar: AnnisResponse = CorpusService.get_corpus(cts_urn=exercise.urn, is_csm=False)
if not ar.nodes:
if not ar.graph_data.nodes:
return connexion.problem(404, Config.ERROR_TITLE_NOT_FOUND, Config.ERROR_MESSAGE_CORPUS_NOT_FOUND)
exercise.last_access_time = datetime.utcnow().timestamp()
db.session.commit()
......@@ -40,7 +40,7 @@ def get(eid: str) -> Union[Response, ConnexionResponse]:
ar.uri = NetworkService.get_exercise_uri(exercise)
ar.exercise_id = exercise.eid
ar.exercise_type = exercise_type.value
return NetworkService.make_json_response(ar.__dict__)
return NetworkService.make_json_response(ar.to_dict())
def get_graph_data(title: str, conll_string_or_urn: str, aqls: List[str], exercise_type: ExerciseType,
......@@ -79,7 +79,7 @@ def make_new_exercise(conll: str, correct_feedback: str, exercise_type: str, gen
# create a response
return AnnisResponse(
solutions=json.loads(new_exercise.solutions), uri=f"{Config.SERVER_URI_FILE}/{new_exercise.eid}",
exercise_id=xml_guid)
exercise_id=xml_guid, graph_data=GraphData(links=[], nodes=[]))
def map_exercise_data_to_database(exercise_data: ExerciseData, exercise_type: str, instructions: str, xml_guid: str,
......@@ -93,7 +93,7 @@ def map_exercise_data_to_database(exercise_data: ExerciseData, exercise_type: st
# add content to solutions
solutions: List[Solution] = adjust_solutions(exercise_data=exercise_data, solutions=solutions,
exercise_type=exercise_type)
quiz_solutions: str = json.dumps([x.serialize() for x in solutions])
quiz_solutions: str = json.dumps([x.to_dict() for x in solutions])
tc: TextComplexity = TextComplexityService.text_complexity(TextComplexityMeasure.all.name, urn, False,
exercise_data.graph)
new_exercise: Exercise = ExerciseMC.from_dict(
......@@ -130,7 +130,7 @@ def post(exercise_data: dict) -> Union[Response, ConnexionResponse]:
return connexion.problem(500, Config.ERROR_TITLE_INTERNAL_SERVER_ERROR,
Config.ERROR_MESSAGE_INTERNAL_SERVER_ERROR)
solutions_dict_list: List[Dict] = response["solutions"]
solutions: List[Solution] = [Solution(json_dict=x) for x in solutions_dict_list]
solutions: List[Solution] = [Solution.from_dict(x) for x in solutions_dict_list]
ar: AnnisResponse = make_new_exercise(
conll=response["conll"], correct_feedback=exercise_data.get("correct_feedback", ""),
exercise_type=exercise_data["type"], general_feedback=exercise_data.get("general_feedback", ""),
......@@ -140,4 +140,4 @@ def post(exercise_data: dict) -> Union[Response, ConnexionResponse]:
search_values=exercise_data["search_values"], solutions=solutions,
type_translation=exercise_data.get("type_translation", ""), urn=urn,
work_author=exercise_data.get("work_author", ""), work_title=exercise_data.get("work_title", ""))
return NetworkService.make_json_response(ar.__dict__)
return NetworkService.make_json_response(ar.to_dict())
"""The corpus list API. Add it to your REST API to provide users with a list of metadata for available texts."""
from datetime import datetime
from typing import List, Set
import conllu
......
......@@ -18,8 +18,8 @@ class RawTextAPI(Resource):
args = self.reqparse.parse_args()
urn: str = args["urn"]
ar: AnnisResponse = CorpusService.get_corpus(cts_urn=urn, is_csm=False)
if not ar.nodes:
if not ar.graph_data.nodes:
abort(404)
gd: GraphData = GraphData(json_dict=ar.__dict__)
ar.text_complexity = TextComplexityService.text_complexity(TextComplexityMeasure.all.name, urn, False, gd)
return NetworkService.make_json_response(ar.__dict__)
ar.text_complexity = TextComplexityService.text_complexity(TextComplexityMeasure.all.name, urn, False,
ar.graph_data).to_dict()
return NetworkService.make_json_response(ar.to_dict())
......@@ -19,6 +19,5 @@ class TextComplexityAPI(Resource):
urn: str = args["urn"]
measure: str = args["measure"]
ar: AnnisResponse = CorpusService.get_corpus(urn, is_csm=False)
gd: GraphData = GraphData(json_dict=ar.__dict__)
tc: TextComplexity = TextComplexityService.text_complexity(measure, urn, False, gd)
return NetworkService.make_json_response(tc.__dict__)
tc: TextComplexity = TextComplexityService.text_complexity(measure, urn, False, ar.graph_data)
return NetworkService.make_json_response(tc.to_dict())
......@@ -33,17 +33,17 @@ class VocabularyAPI(Resource):
for char in string.punctuation:
vocabulary_set.add(char)
ar: AnnisResponse = CorpusService.get_corpus(cts_urn=urn, is_csm=False)
graph_data: GraphData = GraphData(json_dict=ar.__dict__)
if show_oov:
# this is not a request for sentence ranges, so we can take a shortcut
for node in graph_data.nodes:
for node in ar.graph_data.nodes:
if not is_match(target_lemma=node.udep_lemma, vocabulary_set=vocabulary_set):
node.is_oov = True
ar: AnnisResponse = AnnisResponse(solutions=[], uri="", exercise_id="", graph_data=graph_data)
gd: GraphData = GraphData(json_dict=ar.__dict__)
ar.text_complexity = TextComplexityService.text_complexity(TextComplexityMeasure.all.name, urn, False, gd)
return NetworkService.make_json_response(ar.__dict__)
sentences: List[Sentence] = check_vocabulary(graph_data, vocabulary_set)
ar: AnnisResponse = AnnisResponse(
solutions=[], uri="", exercise_id="", graph_data=ar.graph_data)
ar.text_complexity = TextComplexityService.text_complexity(TextComplexityMeasure.all.name, urn, False,
ar.graph_data).to_dict()
return NetworkService.make_json_response(ar.to_dict())
sentences: List[Sentence] = check_vocabulary(ar.graph_data, vocabulary_set)
return NetworkService.make_json_response([x.__dict__ for x in sentences])
......
......@@ -2,9 +2,25 @@
from typing import Dict, List, Union, Any
from enum import Enum
import typing
from sqlalchemy.orm.state import InstanceState
from mcserver.config import Config
from mcserver.models_auto import TExercise, Corpus, TCorpus, Exercise, TLearningResult, LearningResult
from openapi.openapi_server.models import SolutionElement, Solution, Link, Node, TextComplexity, AnnisResponse, \
GraphData
AnnisResponse = AnnisResponse
GraphData = GraphData
LinkMC = Link
NodeMC = Node
SolutionElement = SolutionElement
TextComplexity = TextComplexity
def make_solution_element_from_salt_id(salt_id: str) -> SolutionElement:
"""Extracts necessary information from a SALT ID string to create a solution element."""
salt_parts: List[str] = salt_id.split("#")[-1].split("tok")
sentence_id = int(salt_parts[0].replace("sent", ""))
token_id = int(salt_parts[1].replace("tok", ""))
return SolutionElement(content="", salt_id=salt_id, sentence_id=sentence_id, token_id=token_id)
class Case(Enum):
......@@ -369,142 +385,6 @@ class XapiStatement:
context=self.context.serialize(), result=self.result.serialize())
class LinkMC:
annis_component_name: str
annis_component_type: str
source: str
target: str
udep_deprel: str
def __init__(self, annis_component_name: str = "", annis_component_type: str = "", source: str = "",
target: str = "", udep_deprel: str = None, json_dict: dict = None):
if json_dict:
self.__dict__ = json_dict
else:
self.annis_component_name = annis_component_name
self.annis_component_type = annis_component_type
self.source = source
self.target = target
if udep_deprel is not None:
self.udep_deprel = udep_deprel
def __eq__(self, other):
if isinstance(other, LinkMC):
for key in other.__dict__:
if not isinstance(other.__dict__[key], InstanceState) and other.__dict__[key] != self.__dict__[key]:
return False
return True
else:
return False
class NodeMC:
annis_node_name: str
annis_node_type: str
annis_tok: str
annis_type: str
id: str
is_oov: bool
udep_lemma: str
udep_upostag: str
udep_xpostag: str
udep_feats: str
solution: str
def __init__(self, annis_node_name: str = "", annis_node_type: str = "", annis_tok: str = "", annis_type: str = "",
node_id: str = "", udep_upostag: str = "", udep_xpostag: str = "", udep_feats: str = "",
solution: str = "", udep_lemma: str = None, is_oov: bool = None, json_dict: dict = None):
if json_dict:
self.__dict__ = json_dict
else:
self.annis_node_name = annis_node_name
self.annis_node_type = annis_node_type
self.annis_tok = annis_tok
self.annis_type = annis_type
self.id = node_id
if udep_lemma is not None:
self.udep_lemma = udep_lemma
self.udep_upostag = udep_upostag
self.udep_xpostag = udep_xpostag
self.udep_feats = udep_feats
self.solution = solution
self.is_oov = is_oov
def __eq__(self, other):
if isinstance(other, NodeMC):
return self.annis_node_name == other.annis_node_name and self.annis_node_type == other.annis_node_type and self.annis_tok == other.annis_tok and self.annis_type == other.annis_type and self.id == other.id and self.udep_lemma == other.udep_lemma and self.udep_upostag == other.udep_upostag and self.udep_xpostag == other.udep_xpostag and self.solution == other.solution
else:
return False
class GraphData:
directed: bool
graph: Dict
links: List[LinkMC]
multigraph: bool
nodes: List[NodeMC]
def __init__(self, directed: bool = None, graph: Dict = None, links: List[LinkMC] = None, multigraph: bool = None,
nodes: List[NodeMC] = None, json_dict: dict = None):
if json_dict is None:
self.directed = directed
self.graph = graph
self.links = links
self.multigraph = multigraph
self.nodes: List[NodeMC] = nodes
else:
self.directed = json_dict["directed"]
self.graph = json_dict["graph"]
self.multigraph = json_dict["multigraph"]
self.links = [LinkMC(json_dict=x) for x in json_dict["links"]]
self.nodes = [NodeMC(json_dict=x) for x in json_dict["nodes"]]
def serialize(self) -> dict:
ret_val: dict = self.__dict__.copy()
ret_val["links"] = [x.__dict__ for x in self.links]
ret_val["nodes"] = [x.__dict__ for x in self.nodes]
return ret_val
class SolutionElement:
sentence_id: int
token_id: int
content: str
salt_id: str
def __init__(self, sentence_id: int = 0, token_id: int = 0, content: str = None, json_dict: Dict = None,
salt_id: str = None):
if json_dict:
self.__dict__ = json_dict
elif salt_id:
salt_parts: List[str] = salt_id.split("#")[-1].split("tok")
self.sentence_id = int(salt_parts[0].replace("sent", ""))
self.token_id = int(salt_parts[1].replace("tok", ""))
self.salt_id = salt_id
self.content = content
else:
self.sentence_id = sentence_id
self.token_id = token_id
self.content = content
class Solution:
target: SolutionElement
value: SolutionElement
def __init__(self, target: SolutionElement = SolutionElement(), value: SolutionElement = SolutionElement(),
json_dict: dict = None):
if json_dict:
self.target = SolutionElement(json_dict=json_dict["target"])
self.value = SolutionElement(json_dict=json_dict["value"])
else:
self.target = target
self.value = value
def serialize(self) -> dict:
return dict(target=self.target.__dict__, value=self.value.__dict__)
class ExerciseData:
"""Model for exercise data. Holds textual annotations as a graph"""
graph: GraphData
......@@ -514,19 +394,19 @@ class ExerciseData:
def __init__(self, graph: GraphData = None, uri: str = None, solutions: List[Solution] = None,
json_dict: dict = None):
if json_dict is not None:
self.graph = GraphData(json_dict=json_dict["graph"])
self.graph = GraphData.from_dict(json_dict["graph"])
self.uri = json_dict["uri"]
self.solutions = [Solution(json_dict=solution_dict) for solution_dict in json_dict["solutions"]]
self.solutions = [Solution.from_dict(solution_dict) for solution_dict in json_dict["solutions"]]
else:
self.graph = graph
self.solutions = [] if solutions is None else solutions
self.uri = uri
def serialize(self) -> dict:
ret_val: dict = {"solutions": [x.serialize() for x in self.solutions],
ret_val: dict = {"solutions": [x.to_dict() for x in self.solutions],
"graph": dict(multigraph=self.graph.multigraph, directed=self.graph.directed,
graph=self.graph.graph, nodes=[x.__dict__ for x in self.graph.nodes],
links=[x.__dict__ for x in self.graph.links]), "uri": self.uri}
graph=self.graph.graph, nodes=[x.to_dict() for x in self.graph.nodes],
links=[x.to_dict() for x in self.graph.links]), "uri": self.uri}
return ret_val
......@@ -618,53 +498,3 @@ class FrequencyAnalysis(List[FrequencyItem]):
def serialize(self) -> List[dict]:
return [x.serialize() for x in self]
class AnnisResponse:
def __init__(self, solutions: List[Solution] = None, uri: str = "", exercise_id: str = "",
graph_data: GraphData = None, frequency_analysis: FrequencyAnalysis = None,
text_complexity: dict = None, exercise_type: ExerciseType = None,
json_dict: dict = None):
if json_dict is None:
self.directed: bool = graph_data.directed if graph_data else False
self.exercise_id: str = exercise_id
self.exercise_type: str = exercise_type.value if exercise_type else ""
self.frequency_analysis: List[dict] = [] if frequency_analysis is None else frequency_analysis.serialize()
self.graph: dict = graph_data.graph if graph_data else {}
self.links: List[dict] = [x.__dict__ for x in graph_data.links] if graph_data else []
self.multigraph: bool = graph_data.multigraph if graph_data else False
self.nodes: List[dict] = [x.__dict__ for x in graph_data.nodes] if graph_data else []
self.solutions: List[Solution] = solutions
self.text_complexity: dict = text_complexity if text_complexity else {}
self.uri: str = uri
else:
self.__dict__ = json_dict
class TextComplexity(dict):
def __init__(self, n_w: int = 0, pos: int = 0, n_sent: int = 0, avg_w_per_sent: float = 0, avg_w_len: float = 0,
n_punct: int = 0, n_types: int = 0, lex_den: float = 0, n_clause: int = 0, n_subclause: int = 0,
n_abl_abs: int = 0, n_gerund: int = 0, n_inf: int = 0, n_part: int = 0, all: float = 0,
json_dict: dict = None):
super(TextComplexity).__init__()
if json_dict is None:
self.n_w: int = n_w
self.pos: int = pos
self.n_sent: int = n_sent
self.avg_w_per_sent: float = avg_w_per_sent
self.avg_w_len: float = avg_w_len
self.n_punct: int = n_punct
self.n_types: int = n_types
self.lex_den: float = lex_den
self.n_clause: int = n_clause
self.n_subclause: int = n_subclause
self.n_abl_abs: int = n_abl_abs
self.n_gerund: int = n_gerund
self.n_inf: int = n_inf
self.n_part: int = n_part
self.all: float = all
else:
self.update(json_dict)
for key in json_dict:
self.__setattr__(key, json_dict[key])
......@@ -252,7 +252,7 @@ class AnnotationService:
""" Maps a node dictionary to the native NodeMC class. """
return NodeMC(annis_node_name=node["annis::node_name"], annis_node_type=node["annis::node_type"],
annis_tok=node.get("annis::tok", None), annis_type=node.get("annis::type", None),
node_id=str(node.get("id", "")), udep_lemma=node.get("udep::lemma", None),
id=str(node.get("id", "")), udep_lemma=node.get("udep::lemma", None),
udep_upostag=node.get("udep::upostag", None), udep_xpostag=node.get("udep::xpostag", None),
udep_feats=node.get("udep::feats", None))
......
......@@ -13,10 +13,9 @@ from lxml import etree
from networkx import graph, MultiDiGraph
from networkx.readwrite import json_graph
from requests import HTTPError
from mcserver.app import db
from mcserver.app.models import CitationLevel, GraphData, Solution, ExerciseType, Phenomenon, FrequencyAnalysis, \
AnnisResponse, SolutionElement, CorpusMC
AnnisResponse, CorpusMC, make_solution_element_from_salt_id
from mcserver.app.services import AnnotationService, XMLservice, TextService, FileService, FrequencyService, \
CustomCorpusService
from mcserver.config import Config
......@@ -97,7 +96,7 @@ class CorpusService:
# get graph data for further processing
graph_data_raw: dict = CorpusService.get_graph_data(cts_urn)
if not graph_data_raw:
return AnnisResponse()
return AnnisResponse(graph_data=GraphData(links=[], nodes=[]))
graph_data: GraphData = AnnotationService.map_graph_data(graph_data_raw)
ar: AnnisResponse = AnnisResponse(solutions=[], uri="", exercise_id="", graph_data=graph_data)
return ar
......@@ -105,14 +104,13 @@ class CorpusService:
# there is actually no text, only a URN, so we need to get it ourselves
url: str = f"{Config.INTERNET_PROTOCOL}{Config.HOST_IP_CSM}:{Config.CORPUS_STORAGE_MANAGER_PORT}/"
response: requests.Response = requests.get(url, params=dict(urn=cts_urn))
return AnnisResponse(json_dict=json.loads(response.text))
return AnnisResponse(graph_data=GraphData.from_dict(json.loads(response.text)))
@staticmethod
def get_frequency_analysis(urn: str, is_csm: bool) -> FrequencyAnalysis:
""" Collects frequency statistics for various combinations of linguistic annotations in a corpus. """
if is_csm:
ar: AnnisResponse = CorpusService.get_corpus(urn, is_csm)
gd: GraphData = GraphData(json_dict=ar.__dict__)
search_phenomena: List[List[Phenomenon]] = []
for head_phenomenon in Phenomenon:
for base_phenomenon in Phenomenon:
......@@ -126,7 +124,7 @@ class CorpusService:
fa += FrequencyService.add_case_frequencies(disk_urn, search_phenomenon)
elif search_phenomenon[0] in [Phenomenon.lemma, Phenomenon.partOfSpeech]:
fa += FrequencyService.add_generic_frequencies(disk_urn, search_phenomenon)
FrequencyService.add_dependency_frequencies(gd, fa)
FrequencyService.add_dependency_frequencies(ar.graph_data, fa)
return FrequencyService.extract_case_values(fa)
else:
url: str = Config.INTERNET_PROTOCOL + f"{Config.HOST_IP_CSM}:{Config.CORPUS_STORAGE_MANAGER_PORT}" + \
......@@ -195,21 +193,23 @@ class CorpusService:
# it's cloze or markWords; the solutions only have a target, no explicit value
if search_phenomena[0] == Phenomenon.dependency:
node_ids = [node_ids[i] for i in range(len(node_ids)) if i % 2 != 0]
matches += [Solution(target=SolutionElement(salt_id=x)) for x in node_ids]
matches += [Solution(target=make_solution_element_from_salt_id(x)) for x in node_ids]
else:
matches += [Solution(target=SolutionElement(salt_id=x)) for x in node_ids]