check_recsys.py

def check_line_by_line(fn_result, fn_truth):
    logger.info(f"Checking line by line in {fn_result} and {fn_truth}")
    with open(fn_result, 'r') as results:
        res = results.readlines()
    with open(fn_truth, 'r') as truth:
        tru = truth.readlines()

    for i, (line_t, line_r) in enumerate(zip(tru, res)):
        features = line_t.split('\x01')
        twid, usid, p1, p2, p3, p4 = line_r.split(',')
        assert features[2] == twid
        assert features[9] == usid
        assert p1 is not None
        assert p2 is not None
        assert p3 is not None
        assert p4 is not None
        assert 0.0 <= float(p1) <= 1.0
        assert 0.0 <= float(p2) <= 1.0
        assert 0.0 <= float(p3) <= 1.0
        assert 0.0 <= float(p4) <= 1.0
    logger.info(f"Checked {i} lines")


def check_file(fn_result, fn_truth):
    logger.info(f"Checking file {fn_truth} to ensure every "
                f"tweet ID is found in {fn_result}")
    results = {}
    with open(fn_result, 'r') as fh:
        for line_r in fh:
            twid, usid, p1, p2, p3, p4 = line_r.split(',')
            results[(twid, usid)] = (p1, p2, p3, p4)
    with open(fn_truth, 'r') as fh:
        for j, line_t in enumerate(fh.readlines()):
            features = line_t.split('\x01')
            twid = features[2]
            usid = features[9]
            assert (twid, usid) in results
    logger.info(f"Checked {j} lines")