Add option to pass the PFDL program as a string

pfdl_scheduler/scheduler.py

@@ -22,8 +22,7 @@
 from pfdl_scheduler.api.task_api import TaskAPI
 from pfdl_scheduler.api.service_api import ServiceAPI
 
-from pfdl_scheduler.utils.parsing_utils import load_file
-from pfdl_scheduler.utils.parsing_utils import parse_file
+from pfdl_scheduler.utils.parsing_utils import parse_program
 
 from pfdl_scheduler.petri_net.generator import Node, PetriNetGenerator
 from pfdl_scheduler.petri_net.logic import PetriNetLogic
@@ -73,28 +72,29 @@ class Scheduler(Subject):
 
     def __init__(
         self,
-        pfdl_file_path: str,
+        pfdl_program: str,
         generate_test_ids: bool = False,
         draw_petri_net: bool = True,
         scheduler_id: str = "",
         dashboard_host_address: str = "",
     ) -> None:
         """Initialize the object.
 
-        If the given path leads to a valid PFDL file the parsing will be started. If no errors
-        occur the model of the PFDL File will be transformed into a petri net and be drawn if
-        the `draw_petri_net` flag is set. If `generate_test_ids` is set the ids of the called
+        If the given program contains a path that leads to a valid PFDL file or consists of a
+        valid PFDL string directly, the parsing will be started. If no errors occur the model
+        of the PFDL file will be transformed into a petri net and be drawn if the
+        `draw_petri_net` flag is set. If `generate_test_ids` is set the ids of the called
         tasks and services will be an enumeration starting at 0.
 
         Args:
-            pfdl_file_path: The path to the PFDL file.
+            pfdl_program: A path to the PFDL file or directly the PFDL program as a string.
             generate_test_ids: A boolean indicating whether test ids should be generated.
             draw_petri_net: A boolean indicating whether the petri net should be drawn.
             scheduler_id: A unique ID to identify the Scheduer / Production Order
             dashboard_host_address: The address of the Dashboard (if existing)
         """
         self.init_scheduler(scheduler_id, generate_test_ids)
-        self.pfdl_file_valid, self.process, pfdl_string = parse_file(pfdl_file_path)
+        self.pfdl_file_valid, self.process, pfdl_string = parse_program(pfdl_program)
 
         if self.pfdl_file_valid:
             self.petri_net_generator = PetriNetGenerator(
@@ -326,7 +326,7 @@ def on_service_started(self, service_api: ServiceAPI) -> None:
 
             if service_api.input_parameters:
                 service_api.input_parameters = copy.deepcopy(service_api.service.input_parameters)
-    
+
             self.substitute_loop_indexes(service_api)
         elif self.generate_test_ids:
             new_uuid = str(self.test_id_counters[1])

pfdl_scheduler/utils/parsing_utils.py

@@ -8,6 +8,7 @@
 
 # standard libraries
 import re
+import os
 from typing import Tuple, Union
 from pathlib import Path
 
@@ -67,17 +68,17 @@ def parse_string(
     return (False, None)
 
 
-def parse_file(file_path: str) -> Tuple[bool, Union[None, Process], str]:
-    """Loads the content of the file from the given path and calls the parse_string function.
+def parse_program(program: str) -> Tuple[bool, Union[None, Process], str]:
+    """Loads the content of the program from either the given path or the PFDL program directly and calls the parse_string function.
 
     Args:
-        file_path: The path to the PFDL file.
+        program: Either a path to the PFDL file or directly the PFDL program as a string.
 
     Returns:
         A boolan indicating validity of the PFDL file, the content of the file, and the
         process object if so, otherwise None.
     """
-    pfdl_string = load_file(file_path)
+    pfdl_string, file_path = extract_content_and_file_path(program)
     return *parse_string(pfdl_string, file_path), pfdl_string
 
 
@@ -93,6 +94,29 @@ def write_tokens_to_file(token_stream: CommonTokenStream) -> None:
             file.write(token_text + "\n")
 
 
+def extract_content_and_file_path(program):
+    """Extracts the file path and loads the PFDL string for a given program.
+
+    Args:
+        program: Either a path to the PFDL file or directly the PFDL program as a string.
+
+    Returns:
+        The content of the PFDL file as a string and the file path if it is contained in given the program,
+        otherwise an empty string.
+
+    """
+    file_path = ""
+    if os.path.exists(program):
+        # PFDL program was passed as a file path
+        file_path = program
+        pfdl_string = load_file(program)
+    else:
+        # expect program to contain a PFDL string
+        pfdl_string = program
+
+    return (pfdl_string, file_path)
+
+
 def load_file(file_path: str) -> str:
     """Loads the content of the file from the given path.
 

pfdl_scheduler/validation/error_handler.py

@@ -11,9 +11,9 @@
 
 
 class ErrorHandler:
-    """Keeps track of the total error amount in an PFDL file.
+    """Keeps track of the total error amount in a PFDL file.
 
-    Provides a method for printing an erro which counts the errors.
+    Provides a method for printing an error which counts the errors.
 
     Attributes:
         total_error_count: Total number of errors.

tests/integration_tests/test_scheduling_marking.py

@@ -35,14 +35,16 @@ def setUp(self) -> None:
         self.task_started_triggered: List[str] = []
         self.task_finished_triggered: List[str] = []
 
-    def load_file(self, test_file_name: str) -> None:
+    def load_file(self, test_file_name: str, is_pfdl_string: bool) -> None:
         """Loads a file from the given path and parses it if it is a PFDL program."""
+        if is_pfdl_string:
+            pfdl_program = test_file_name
+        else:
+            pfdl_program = TEST_FILE_FOLDER_PATH + test_file_name + ".pfdl"
+        self.scheduler = Scheduler(pfdl_program, True, False)
 
-        file_path = TEST_FILE_FOLDER_PATH + test_file_name + ".pfdl"
-        self.scheduler = Scheduler(file_path, True, False)
-
-    def setup(self, test_case_name: str) -> None:
-        self.load_file(test_case_name)
+    def setup(self, test_case_name: str, is_pfdl_string: bool = False) -> None:
+        self.load_file(test_case_name, is_pfdl_string)
 
         self.assertFalse(self.scheduler.running)
         self.scheduler.start()
@@ -72,6 +74,20 @@ def test_simple_task(self) -> None:
 
         self.assertTrue(self.token_in_last_place())
 
+    def test_simple_task_with_pfdl_string(self) -> None:
+        file_path = TEST_FILE_FOLDER_PATH + "simple_task.pfdl"
+        file_content = ""
+        with open(file_path, "r", encoding="utf-8") as file:
+            file_content = file.read()
+
+        # directly pass the pfdl string to the scheduler
+        self.setup(file_content, True)
+
+        event = Event("service_finished", data={"service_id": "0"})
+        self.fire_event(event)
+
+        self.assertTrue(self.token_in_last_place())
+
     def test_multiple_services(self) -> None:
         self.setup("multiple_services")
 

validate_pfdl_file.py

@@ -10,7 +10,7 @@
 
 import argparse
 
-from pfdl_scheduler.utils.parsing_utils import parse_file
+from pfdl_scheduler.utils.parsing_utils import parse_program
 
 EXAMPLES_PATH = "examples/"
 
@@ -22,7 +22,7 @@ def main():
 
     args = parser.parse_args()
     if args.file_path != None:
-        parse_file(args.file_path)
+        parse_program(args.file_path)
     else:
         folder_path = EXAMPLES_PATH
         if args.folder_path != None:
@@ -32,7 +32,7 @@ def main():
 
         for example_filename in example_filenames:
             print("File " + example_filename + " parsed:")
-            parse_file(folder_path + example_filename)
+            parse_program(folder_path + example_filename)
             print("\n")