-
Notifications
You must be signed in to change notification settings - Fork 0
/
driver.py
executable file
·81 lines (67 loc) · 2.53 KB
/
driver.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
#!/usr/bin/env python
"""Semester course planning.
See test code below. See ``doc`` directory in this project for further details.
Example: ./driver.py
[Results will be printed to the console.]
The contents of the input file ``CSE505/results/cse_courses.lp`` are as follows:
.. code-block:: clingo
% Course atoms
course(cse101, 3, "Undergraduate", 1, 1).
course(cse102, 3, "Undergraduate", 0, 0).
course(cse110, 3, "Undergraduate", 1, 1).
.
.
.
% Prerequisites for CSE113
:- course(cse113, 4, "Undergraduate", 0, 1),
not course(ams151, _, "Undergraduate", _, _),
not course(mat125, _, "Undergraduate", _, _),
not course(mat131, _, "Undergraduate", _, _).
% Prerequisites for CSE114
:- course(cse114, 4, "Undergraduate", 1, 0),
not course(cse101, _, "Undergraduate", _, _),
not course(ise108, _, "Undergraduate", _, _).
"""
import os
import sys
# Add project root to the Python path
_PKG_PATH = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
sys.path.append(_PKG_PATH)
# Import project modules
from src.kg.knowledge_graph import KnowledgeGraph, scrape_sbu_solar
from src.clapi.clapi import process_course_data_clingo, query_clingo
if not (os.path.exists("results")):
os.mkdir("results")
results_dir: str = os.path.abspath("results")
# 1. Create knowledge graph if it does not exist
if (not (os.path.exists(f"{results_dir}/cse_courses.json"))) and (
not (os.path.exists(f"{results_dir}/cse_courses.lp"))
):
kg: KnowledgeGraph = scrape_sbu_solar(
url="https://prod.ps.stonybrook.edu/psc/csprodg/EMPLOYEE/CAMP/c/COMMUNITY_ACCESS.SSS_BROWSE_CATLG.GBL?",
major_three_letter_code="cse",
wait_time=10,
headless=True,
verbose=True,
output_filename=f"{results_dir}/cse_courses",
)
kg.df.to_json(f"{results_dir}/cse_courses.json", orient="index", indent=4)
# 2. Convert JSON to Clingo if it does not exist
if not (os.path.exists(f"{results_dir}/cse_courses.lp")):
clingo_file: str = process_course_data_clingo(
json_file=f"{results_dir}/cse_courses.json",
output_file=f"{results_dir}/cse_courses.lp",
)
# 3. Query the knowledge graph
# NOTE: Not all of the files used in the query are
# created (automatically) in this script.
query_result: str = query_clingo(
knowledge=f"{results_dir}/cse_courses.lp",
num_models=None,
configuration="handy",
parallel_mode=None,
query=(
f"{os.path.join(results_dir,'cse_prereqs.lp')}",
f"{os.path.join(results_dir,'sem.lp')}",
),
)