add lots of type annotations

Miguel Barão
1 parent 2a25e542
Showing 5 changed files with 89 additions and 61 deletions Show diff stats
BUGS.md
aprendizations/learnapp.py
aprendizations/main.py
aprendizations/serve.py
aprendizations/student.py
 # BUGS
+- marking all options right in a radio question breaks!
+- impedir que quando students.db não é encontrado, crie um ficheiro vazio.
+- opcao --prefix devia afectar a base de dados?
 - duplo clicks no botao "responder" dessincroniza as questões, ver debounce em https://stackoverflow.com/questions/20281546/how-to-prevent-calling-of-en-event-handler-twice-on-fast-clicks
 - quando termina topico devia apagar as perguntas todas (se falhar a gerar novo topico, aparecem perguntas do antigo)
 - apos clicar no botao responder, inactivar o input (importante quando o tempo de correcção é grande)
@@ -6,7 +6,7 @@ from contextlib import contextmanager  # `with` statement in db sessions
 import asyncio
 from datetime import datetime
 from random import random
-from typing import Dict
+from typing import Any, Dict, Iterable, List, Optional, Tuple
 # third party libraries
 import bcrypt
@@ -16,7 +16,7 @@ import networkx as nx
 # this project
 from .models import Student, Answer, Topic, StudentTopic
 from .student import StudentState
-from .questions import QFactory
+from .questions import Question, QFactory, QDict
 from .tools import load_yaml
 # setup logger for this module
@@ -56,16 +56,21 @@ class LearnApp(object):
     # ------------------------------------------------------------------------
     # init
     # ------------------------------------------------------------------------
-    def __init__(self, config_files, prefix, db, check=False):
+    def __init__(self,
+                 config_files: List[str],
+                 prefix: str,   # path to topics
+                 db: str,       # database filename
+                 check: bool = False) -> None:
+
         self.db_setup(db)       # setup database and check students
-        self.online = dict()    # online students
+        self.online: Dict[str, Dict] = dict()    # online students
         self.deps = nx.DiGraph(prefix=prefix)
         for c in config_files:
             self.populate_graph(c)
         # factory is a dict with question generators for all topics
-        self.factory = self.make_factory()    # {'qref': QFactory()}
+        self.factory: Dict[str, QFactory] = self.make_factory()
         # if graph has topics that are not in the database, add them
         self.db_add_missing_topics(self.deps.nodes())
@@ -74,10 +79,10 @@ class LearnApp(object):
             self.sanity_check_questions()
     # ------------------------------------------------------------------------
-    def sanity_check_questions(self):
+    def sanity_check_questions(self) -> None:
         logger.info('Starting sanity checks (may take a while...)')
-        errors = 0
+        errors: int = 0
         for qref in self.factory:
             logger.debug(f'checking {qref}...')
             try:
@@ -114,7 +119,8 @@ class LearnApp(object):
     # ------------------------------------------------------------------------
     # login
     # ------------------------------------------------------------------------
-    async def login(self, uid, pw):
+    async def login(self, uid: str, pw: str) -> bool:
+
         with self.db_session() as s:
             found = s.query(Student.name, Student.password) \
                      .filter_by(id=uid) \
@@ -168,14 +174,14 @@ class LearnApp(object):
     # ------------------------------------------------------------------------
     # logout
     # ------------------------------------------------------------------------
-    def logout(self, uid):
+    def logout(self, uid: str) -> None:
         del self.online[uid]
         logger.info(f'User "{uid}" logged out')
     # ------------------------------------------------------------------------
     # change_password. returns True if password is successfully changed.
     # ------------------------------------------------------------------------
-    async def change_password(self, uid, pw):
+    async def change_password(self, uid: str, pw: str) -> bool:
         if not pw:
             return False
@@ -191,12 +197,14 @@ class LearnApp(object):
         return True
     # ------------------------------------------------------------------------
-    # checks answer (updating student state) and returns grade.
+    # checks answer (updating student state) and returns grade. FIXME type of answer
     # ------------------------------------------------------------------------
-    async def check_answer(self, uid, answer):
+    async def check_answer(self, uid: str, answer) -> Tuple[Question, str]:
         knowledge = self.online[uid]['state']
         topic = knowledge.get_current_topic()
+
         q, action = await knowledge.check_answer(answer)  # may move questions
+
         logger.info(f'User "{uid}" got {q["grade"]:.2} in "{q["ref"]}"')
         # always save grade of answered question
@@ -213,8 +221,8 @@ class LearnApp(object):
         if knowledge.topic_has_finished():
             # finished topic, save into database
             logger.info(f'User "{uid}" finished "{topic}"')
-            level = knowledge.get_topic_level(topic)
-            date = str(knowledge.get_topic_date(topic))
+            level: float = knowledge.get_topic_level(topic)
+            date: str = str(knowledge.get_topic_date(topic))
             with self.db_session() as s:
                 a = s.query(StudentTopic) \
@@ -242,19 +250,19 @@ class LearnApp(object):
     # ------------------------------------------------------------------------
     # Start new topic
     # ------------------------------------------------------------------------
-    async def start_topic(self, uid, topic):
+    async def start_topic(self, uid: str, topic: str) -> None:
         student = self.online[uid]['state']
         try:
             await student.start_topic(topic)
         except Exception as e:
-            logger.warning(f'User "{uid}" couldn\'t start "{topic}": {e}')
+            logger.warning(f'User "{uid}" could not start "{topic}": {e}')
         else:
             logger.info(f'User "{uid}" started topic "{topic}"')
     # ------------------------------------------------------------------------
     # Fill db table 'Topic' with topics from the graph if not already there.
     # ------------------------------------------------------------------------
-    def db_add_missing_topics(self, topics):
+    def db_add_missing_topics(self, topics: List[str]) -> None:
         with self.db_session() as s:
             new_topics = [Topic(id=t) for t in topics
                           if (t,) not in s.query(Topic.id)]
@@ -267,15 +275,16 @@ class LearnApp(object):
     # ------------------------------------------------------------------------
     # setup and check database contents
     # ------------------------------------------------------------------------
-    def db_setup(self, db):
+    def db_setup(self, db: str) -> None:
+
         logger.info(f'Checking database "{db}":')
         engine = sa.create_engine(f'sqlite:///{db}', echo=False)
         self.Session = sa.orm.sessionmaker(bind=engine)
         try:
             with self.db_session() as s:
-                n = s.query(Student).count()
-                m = s.query(Topic).count()
-                q = s.query(Answer).count()
+                n: int = s.query(Student).count()
+                m: int = s.query(Topic).count()
+                q: int = s.query(Answer).count()
         except Exception:
             logger.error(f'Database "{db}" not usable!')
             raise DatabaseUnusableError()
@@ -293,21 +302,22 @@ class LearnApp(object):
     #
     # Edges are obtained from the deps defined in the YAML file for each topic.
     # ------------------------------------------------------------------------
-    def populate_graph(self, conffile: str):
+    def populate_graph(self, conffile: str) -> None:
+
         logger.info(f'Populating graph from:  {conffile}...')
-        config = load_yaml(conffile)    # course configuration
+        config: Dict[str, Any] = load_yaml(conffile)    # course configuration
         # default attributes that apply to the topics
-        default_file = config.get('file', 'questions.yaml')
-        default_shuffle_questions = config.get('shuffle_questions', True)
-        default_choose = config.get('choose', 9999)
-        default_forgetting_factor = config.get('forgetting_factor', 1.0)
-        default_maxtries = config.get('max_tries', 3)
-        default_append_wrong = config.get('append_wrong', True)
-        default_min_level = config.get('min_level', 0.01)  # to unlock topic
+        default_file: str = config.get('file', 'questions.yaml')
+        default_shuffle_questions: bool = config.get('shuffle_questions', True)
+        default_choose: int = config.get('choose', 9999)
+        default_forgetting_factor: float = config.get('forgetting_factor', 1.0)
+        default_maxtries: int = config.get('max_tries', 3)
+        default_append_wrong: bool = config.get('append_wrong', True)
+        default_min_level: float = config.get('min_level', 0.01)  # to unlock
         # iterate over topics and populate graph
-        topics = config.get('topics', {})
+        topics: Dict[str, Dict] = config.get('topics', {})
         g = self.deps       # dependency graph
         g.add_nodes_from(topics.keys())
@@ -345,15 +355,18 @@ class LearnApp(object):
     # Buils dictionary of question factories
     # ------------------------------------------------------------------------
     def make_factory(self) -> Dict[str, QFactory]:
+
         logger.info('Building questions factory:')
-        factory = {}    # {'qref': QFactory()}
+        factory: Dict[str, QFactory] = {}
         g = self.deps
         for tref in g.nodes():
             t = g.node[tref]
             # load questions as list of dicts
-            topicpath = path.join(g.graph['prefix'], tref)
-            questions = load_yaml(path.join(topicpath, t['file']), default=[])
+            topicpath: str = path.join(g.graph['prefix'], tref)
+            fullpath: str = path.join(topicpath, t['file'])
+
+            questions: List[QDict] = load_yaml(fullpath, default=[])
             # update refs to include topic as prefix.
             # refs are required to be unique only within the file.
@@ -389,16 +402,16 @@ class LearnApp(object):
         return self.online[uid].get('name', '')
     # ------------------------------------------------------------------------
-    def get_student_state(self, uid: str):
+    def get_student_state(self, uid: str) -> List[Dict[str, Any]]:
         return self.online[uid]['state'].get_knowledge_state()
     # ------------------------------------------------------------------------
-    def get_student_progress(self, uid: str):
+    def get_student_progress(self, uid: str) -> float:
         return self.online[uid]['state'].get_topic_progress()
     # ------------------------------------------------------------------------
-    def get_current_question(self, uid: str):
-        return self.online[uid]['state'].get_current_question()     # dict
+    def get_current_question(self, uid: str) -> Optional[Question]:
+        return self.online[uid]['state'].get_current_question()
     # ------------------------------------------------------------------------
     def get_current_question_id(self, uid: str) -> str:
@@ -410,27 +423,29 @@ class LearnApp(object):
     # ------------------------------------------------------------------------
     def get_student_topic(self, uid: str) -> str:
-        return self.online[uid]['state'].get_current_topic()        # str
+        return self.online[uid]['state'].get_current_topic()
     # ------------------------------------------------------------------------
     def get_title(self) -> str:
-        return self.deps.graph.get('title', '')  # FIXME
+        return self.deps.graph.get('title', '')
     # ------------------------------------------------------------------------
     def get_topic_name(self, ref: str) -> str:
         return self.deps.node[ref]['name']
     # ------------------------------------------------------------------------
-    def get_current_public_dir(self, uid):
-        topic = self.online[uid]['state'].get_current_topic()
-        prefix = self.deps.graph['prefix']
+    def get_current_public_dir(self, uid: str) -> str:
+        topic: str = self.online[uid]['state'].get_current_topic()
+        prefix: str = self.deps.graph['prefix']
         return path.join(prefix, topic, 'public')
     # ------------------------------------------------------------------------
-    def get_rankings(self, uid):
+    def get_rankings(self, uid: str) -> Iterable[Tuple[str, str, float, float]]:
+
         logger.info(f'User "{uid}" get rankings')
         with self.db_session() as s:
             students = s.query(Student.id, Student.name).all()
+
             # topic progress
             student_topics = s.query(StudentTopic.student_id,
                                      StudentTopic.topic_id,
@@ -448,10 +463,11 @@ class LearnApp(object):
                          all())
         # compute percentage of right answers
-        perf = {uid: right.get(uid, 0.0)/total[uid] for uid in total}
+        perf: Dict[str, float] = {uid: right.get(uid, 0.0)/total[uid]
+                                  for uid in total}
         # compute topic progress
-        prog = {s[0]: 0.0 for s in students}
+        prog: Dict[str, float] = {s[0]: 0.0 for s in students}
         now = datetime.now()
         for uid, topic, level, date in student_topics:
             date = datetime.strptime(date, "%Y-%m-%d %H:%M:%S.%f")
@@ -6,6 +6,7 @@ import logging
 from os import environ, path
 import ssl
 import sys
+from typing import Any, Dict
 # this project
 from .learnapp import LearnApp, DatabaseUnusableError
@@ -61,7 +62,7 @@ def parse_cmdline_arguments():
 # ----------------------------------------------------------------------------
-def get_logger_config(debug=False):
+def get_logger_config(debug: bool = False) -> Dict[str, Any]:
     if debug:
         filename, level = 'logger-debug.yaml', 'DEBUG'
     else:
@@ -70,7 +71,7 @@ def get_logger_config(debug=False):
     config_dir = environ.get('XDG_CONFIG_HOME', '~/.config/')
     config_file = path.join(path.expanduser(config_dir), APP_NAME, filename)
-    default_config = {
+    default_config: Dict[str, Any] = {
         'version': 1,
         'formatters': {
             'standard': {
@@ -8,6 +8,7 @@ import mimetypes
 from os import path
 import signal
 import sys
+from typing import List, Optional, Tuple, Union
 import uuid
 # third party libraries
@@ -39,6 +40,7 @@ def admin_only(func):
 # WebApplication - Tornado Web Server
 # ============================================================================
 class WebApplication(tornado.web.Application):
+
     def __init__(self, learnapp, debug=False):
         handlers = [
             (r'/login',         LoginHandler),
@@ -304,6 +306,7 @@ class QuestionHandler(BaseHandler):
         # --- answers are in a list. fix depending on question type
         qtype = self.learn.get_student_question_type(user)
+        ans: Optional[Union[List, str]]
         if qtype in ('success', 'information', 'info'):
             ans = None
         elif qtype == 'radio' and not answer:
@@ -365,7 +368,7 @@ class QuestionHandler(BaseHandler):
 # ----------------------------------------------------------------------------
 # Signal handler to catch Ctrl-C and abort server
 # ----------------------------------------------------------------------------
-def signal_handler(signal, frame):
+def signal_handler(signal, frame) -> None:
     r = input(' --> Stop webserver? (yes/no) ').lower()
     if r == 'yes':
         tornado.ioloop.IOLoop.current().stop()
@@ -376,7 +379,11 @@ def signal_handler(signal, frame):
 # ----------------------------------------------------------------------------
-def run_webserver(app, ssl, port=8443, debug=False):
+def run_webserver(app,
+                  ssl,
+                  port: int = 8443,
+                  debug: bool = False) -> None:
+
     # --- create web application
     try:
         webapp = WebApplication(app, debug=debug)
@@ -3,7 +3,7 @@
 import random
 from datetime import datetime
 import logging
-from typing import List, Optional
+from typing import List, Optional, Tuple
 # third party libraries
 import networkx as nx
@@ -28,20 +28,21 @@ class StudentState(object):
     # =======================================================================
     # methods that update state
     # =======================================================================
-    def __init__(self, deps, factory, state={}):
+    def __init__(self, deps, factory, state={}) -> None:
         self.deps = deps        # shared dependency graph
         self.factory = factory  # question factory
         self.state = state  # {'topic': {'level': 0.5, 'date': datetime}, ...}
         self.update_topic_levels()  # applies forgetting factor
         self.unlock_topics()        # whose dependencies have been completed
-        self.topic_sequence = self.recommend_topic_sequence()  # ['ref1', ...]
-        self.current_topic = None
+
+        self.topic_sequence: List[str] = self.recommend_topic_sequence()
+        self.current_topic: Optional[str] = None
     # ------------------------------------------------------------------------
     # Updates the proficiency levels of the topics, with forgetting factor
     # ------------------------------------------------------------------------
-    def update_topic_levels(self):
+    def update_topic_levels(self) -> None:
         now = datetime.now()
         for tref, s in self.state.items():
             dt = now - s['date']
@@ -51,7 +52,7 @@ class StudentState(object):
     # ------------------------------------------------------------------------
     # Unlock topics whose dependencies are satisfied (> min_level)
     # ------------------------------------------------------------------------
-    def unlock_topics(self):
+    def unlock_topics(self) -> None:
         for topic in self.deps.nodes():
             if topic not in self.state:  # if locked
                 pred = self.deps.predecessors(topic)
@@ -72,8 +73,8 @@ class StudentState(object):
     #    questions: list of generated questions to do in the topic
     #    current_question: the current question to be presented
     # ------------------------------------------------------------------------
-    async def start_topic(self, topic: str):
-        logger.debug(f'starting "{topic}"')
+    async def start_topic(self, topic: str) -> None:
+        logger.debug(f'start topic "{topic}"')
         # avoid regenerating questions in the middle of the current topic
         if self.current_topic == topic:
@@ -112,7 +113,7 @@ class StudentState(object):
     # The topic level is updated in state and unlocks are performed.
     # The current topic is unchanged.
     # ------------------------------------------------------------------------
-    def finish_topic(self):
+    def finish_topic(self) -> None:
         logger.debug(f'finished {self.current_topic}')
         self.state[self.current_topic] = {
@@ -129,8 +130,8 @@ class StudentState(object):
     #   - if answer ok, goes to next question
     #   - if wrong, counts number of tries. If exceeded, moves on.
     # ------------------------------------------------------------------------
-    async def check_answer(self, answer):
-        q = self.current_question
+    async def check_answer(self, answer) -> Tuple[Question, str]:
+        q: Question = self.current_question
         q['answer'] = answer
         q['finish_time'] = datetime.now()
         logger.debug(f'checking answer of {q["ref"]}...')
1		1
2	# BUGS	2	# BUGS
3		3
		4	+- marking all options right in a radio question breaks!
		5	+- impedir que quando students.db não é encontrado, crie um ficheiro vazio.
		6	+- opcao --prefix devia afectar a base de dados?
4	- duplo clicks no botao "responder" dessincroniza as questões, ver debounce em https://stackoverflow.com/questions/20281546/how-to-prevent-calling-of-en-event-handler-twice-on-fast-clicks	7	- duplo clicks no botao "responder" dessincroniza as questões, ver debounce em https://stackoverflow.com/questions/20281546/how-to-prevent-calling-of-en-event-handler-twice-on-fast-clicks
5	- quando termina topico devia apagar as perguntas todas (se falhar a gerar novo topico, aparecem perguntas do antigo)	8	- quando termina topico devia apagar as perguntas todas (se falhar a gerar novo topico, aparecem perguntas do antigo)
6	- apos clicar no botao responder, inactivar o input (importante quando o tempo de correcção é grande)	9	- apos clicar no botao responder, inactivar o input (importante quando o tempo de correcção é grande)