Added required keys to fake_results, fixed formatting

Author: fjwillemsen

kernel_tuner/file_utils.py

@@ -25,12 +25,8 @@ def store_output_file(output_filename, results, tune_params, objective="time"):
     if output_filename[-5:] != ".json":
         output_filename += ".json"
 
-    timing_keys = [
-        "compile_time", "benchmark_time", "framework_time", "strategy_time",
-        "verification_time"
-    ]
-    not_measurement_keys = list(
-        tune_params.keys()) + timing_keys + ["timestamp"]
+    timing_keys = ["compile_time", "benchmark_time", "framework_time", "strategy_time", "verification_time"]
+    not_measurement_keys = list(tune_params.keys()) + timing_keys + ["timestamp"]
 
     output_data = []
 
@@ -39,10 +35,8 @@ def store_output_file(output_filename, results, tune_params, objective="time"):
         out = {}
 
         out["timestamp"] = result["timestamp"]
-        out["configuration"] = {
-            k: v
-            for k, v in result.items() if k in tune_params
-        }
+        out["configuration"] = { k: v
+                                 for k, v in result.items() if k in tune_params }
 
         # collect configuration specific timings
         timings = dict()
@@ -110,14 +104,12 @@ def get_dependencies(package='kernel_tuner'):
 
 def get_device_query(target):
     if target == "nvidia":
-        nvidia_smi_out = subprocess.run(["nvidia-smi", "--query", "-x"],
-                                        capture_output=True)
+        nvidia_smi_out = subprocess.run(["nvidia-smi", "--query", "-x"], capture_output=True)
         nvidia_smi = xmltodict.parse(nvidia_smi_out.stdout)
         del nvidia_smi["nvidia_smi_log"]["gpu"]["processes"]
         return nvidia_smi
     elif target == "amd":
-        rocm_smi_out = subprocess.run(["rocm-smi", "--showallinfo", "--json"],
-                                      capture_output=True)
+        rocm_smi_out = subprocess.run(["rocm-smi", "--showallinfo", "--json"], capture_output=True)
         return json.loads(rocm_smi_out.stdout)
     else:
         raise ValueError("get_device_query target not supported")
@@ -135,8 +127,7 @@ def store_metadata_file(metadata_filename, target="nvidia"):
     # only works if nvidia-smi (for NVIDIA) or rocm-smi (for AMD) is present, raises FileNotFoundError when not present
     device_query = get_device_query(target)
 
-    metadata["environment"] = dict(device_query=device_query,
-                                   requirements=get_dependencies())
+    metadata["environment"] = dict(device_query=device_query, requirements=get_dependencies())
 
     # write metadata to JSON file
     version, _ = output_file_schema("metadata")

test/test_integration.py

@@ -6,17 +6,31 @@
 
 from kernel_tuner import integration
 from kernel_tuner import util
+from datetime import datetime, timezone
+
 
 @pytest.fixture()
 def fake_results():
     #create fake results for testing
-    tune_params = {"a": [1, 2, 4], "b": [4, 5, 6]}
+    tune_params = {
+        "a": [1, 2, 4],
+        "b": [4, 5, 6]
+    }
     problem_size = 100
     parameter_space = itertools.product(*tune_params.values())
     results = [dict(zip(tune_params.keys(), element)) for element in parameter_space]
-    for i,r in enumerate(results):
-        r["time"] = 100.0+i
-    env = {"device_name": "My GPU"}
+
+    for i, r in enumerate(results):
+        r["time"] = 100.0 + i
+        r["compile_time"] = 20.0 + (i / 5)
+        r["benchmark_time"] = 20.0 + (i / 5)
+        r["framework_time"] = 20.0 + (i / 5)
+        r["strategy_time"] = 20.0 + (i / 5)
+        r["verification_time"] = 20.0 + (i / 5)
+        r["timestamp"] = str(datetime.now(timezone.utc))
+    env = {
+        "device_name": "My GPU"
+    }
 
     return "fake_kernel", "fake_string", tune_params, problem_size, parameter_space, results, env
 
@@ -41,14 +55,14 @@ def test_store_results(fake_results):
         assert len([d for d in stored_data if d["device_name"] == "My_GPU" and d["problem_size"] == "1000"]) == 3
 
         #test if results for a different GPU can be added
-        integration.store_results(filename, kernel_name, kernel_string, tune_params, problem_size, results, {"device_name": "Another GPU"}, top=3)
+        integration.store_results(filename, kernel_name, kernel_string, tune_params, problem_size, results, { "device_name": "Another GPU"}, top=3)
         meta, stored_data = integration._read_results_file(filename)
 
         assert len(set([d["device_name"] for d in stored_data])) == 2
 
         #test if overwriting results works
-        for i,r in enumerate(results):
-            r["time"] = 50.0+i
+        for i, r in enumerate(results):
+            r["time"] = 50.0 + i
         integration.store_results(filename, kernel_name, kernel_string, tune_params, problem_size, results, env, top=0.1)
         meta, stored_data = integration._read_results_file(filename)
 
@@ -60,7 +74,6 @@ def test_store_results(fake_results):
         util.delete_temp_file(filename)
 
 
-
 def test_setup_device_targets(fake_results):
 
     results_filename = "temp_test_results_file.json"
@@ -82,7 +95,7 @@ def test_setup_device_targets(fake_results):
         assert "#define b 4" in output_str
 
         #test output when more then one problem size is used, and best configuration is different
-        for i,e in enumerate(results):
+        for i, e in enumerate(results):
             if e['a'] == 1 and e['b'] == 4:
                 e['time'] += 100
         integration.store_results(results_filename, kernel_name, kernel_string, tune_params, 1000, results, env, top=3)
@@ -94,7 +107,7 @@ def test_setup_device_targets(fake_results):
         assert expected in output_str
 
         #test output when more then one problem size is used, and best configuration depends on total time
-        for i,e in enumerate(results):
+        for i, e in enumerate(results):
             if e['a'] == 1 and e['b'] == 6:
                 e['time'] -= 3
         integration.store_results(results_filename, kernel_name, kernel_string, tune_params, 1000, results, env, top=3)
@@ -106,7 +119,7 @@ def test_setup_device_targets(fake_results):
         assert expected in output_str
 
         #test output when more then one GPU is used
-        for i,e in enumerate(results):
+        for i, e in enumerate(results):
             if e['a'] == 1 and e['b'] == 6:
                 e['time'] += 3.1
         env['device_name'] = "My_GPU2"
@@ -134,7 +147,7 @@ def test_setup_device_targets_max(fake_results):
     kernel_name, kernel_string, tune_params, problem_size, parameter_space, results, env = fake_results
 
     #add GFLOP/s as metric
-    for i,e in enumerate(results):
+    for i, e in enumerate(results):
         e['GFLOP/s'] = 1e5 / e['time']
 
     try:
@@ -148,7 +161,7 @@ def test_setup_device_targets_max(fake_results):
         assert "#define b 4" in output_str
 
         #test output when more then one problem size is used, and best configuration is different
-        for i,e in enumerate(results):
+        for i, e in enumerate(results):
             if e['a'] == 1 and e['b'] == 4:
                 e['time'] += 100
                 e['GFLOP/s'] = 1e5 / e['time']
@@ -160,8 +173,6 @@ def test_setup_device_targets_max(fake_results):
         expected = "\n".join(["TARGET_My_GPU", "#define a 1", "#define b 5"])
         assert expected in output_str
 
-
     finally:
         util.delete_temp_file(results_filename)
         util.delete_temp_file(header_filename)
-