lms.cli.courses.assignments.scores.upload

  1"""
  2Upload scores (and optional comments) for an assignment.
  3"""
  4
  5import argparse
  6import ast
  7import sys
  8import typing
  9
 10import edq.util.dirent
 11
 12import lms.backend.instance
 13import lms.cli.common
 14import lms.cli.parser
 15import lms.model.backend
 16import lms.model.scores
 17import lms.model.users
 18
 19def run_cli(args: argparse.Namespace) -> int:
 20    """ Run the CLI. """
 21
 22    config = args._config_info.application_config
 23    backend = lms.backend.instance.get_backend(config)
 24
 25    course_query = lms.cli.common.check_required_course(backend, config)
 26    if (course_query is None):
 27        return 1
 28
 29    assignment_query = lms.cli.common.check_required_assignment(backend, config)
 30    if (assignment_query is None):
 31        return 2
 32
 33    scores = _load_scores(backend, args.path, args.skip_rows)
 34
 35    count = backend.courses_assignments_scores_resolve_and_upload(course_query, assignment_query, scores)
 36
 37    print(f"Uploaded {count} Scores")
 38
 39    return lms.cli.common.strict_check(config, (count != len(scores)),
 40        f"Expected to upload {len(scores)} scores, but uploaded {count}.", 3)
 41
 42def _load_scores(
 43        backend: lms.model.backend.APIBackend,
 44        path: str,
 45        skip_rows: bool,
 46        ) -> typing.Dict[lms.model.users.UserQuery, lms.model.scores.ScoreFragment]:
 47    scores = {}
 48
 49    with open(path, 'r', encoding = edq.util.dirent.DEFAULT_ENCODING) as file:
 50        lineno = 0
 51        real_rows = 0
 52        for line in file:
 53            lineno += 1
 54
 55            if (line.strip() == ''):
 56                continue
 57
 58            real_rows += 1
 59
 60            if (real_rows <= skip_rows):
 61                continue
 62
 63            parts = [part.strip() for part in line.split("\t")]
 64            if (len(parts) not in [2, 3]):
 65                raise ValueError(f"File '{path}' line {lineno} has the incorrect number of values. Expecting 2-3, found {len(parts)}.")
 66
 67            user_query = backend.parse_user_query(parts[0])
 68            if (user_query is None):
 69                raise ValueError(f"File '{path}' line {lineno} has a user query that could not be parsed: '{parts[0]}'.")
 70
 71            score = None
 72            if (parts[1] != ''):
 73                try:
 74                    score = float(ast.literal_eval(parts[1]))
 75                except Exception:
 76                    raise ValueError(f"File '{path}' line {lineno} has a score that cannot be converted to a number: '{parts[1]}'.")  # pylint: disable=raise-missing-from
 77
 78            comment = None
 79            if (len(parts) == 3):
 80                comment = parts[2]
 81
 82            scores[user_query] = lms.model.scores.ScoreFragment(score = score, comment = comment)
 83
 84    return scores
 85
 86def main() -> int:
 87    """ Get a parser, parse the args, and call run. """
 88    return run_cli(_get_parser().parse_args())
 89
 90def _get_parser() -> argparse.ArgumentParser:
 91    """ Get the parser. """
 92
 93    parser = lms.cli.parser.get_parser(__doc__.strip(),
 94            include_course = True,
 95            include_assignment = True,
 96            include_skip_rows = True,
 97            include_strict = True,
 98    )
 99
100    parser.add_argument('path', metavar = 'PATH',
101        action = 'store', type = str,
102        help = 'Path to a TSV file where each row has 2-3 columns: user query, score, and comment (optional).')
103
104    return parser
105
106if (__name__ == '__main__'):
107    sys.exit(main())