Coverage for sparkle/CLI/core/compute_features.py: 0%

1#!/usr/bin/env python3 

2# -*- coding: UTF-8 -*- 

3"""Compute features for an instance, only for internal calls from Sparkle.""" 

4import argparse 

5from pathlib import Path 

6from filelock import FileLock 

7 

8from sparkle.CLI.help import global_variables as gv 

9from sparkle.structures import FeatureDataFrame 

10from sparkle.instance import Instance_Set 

11from sparkle.solver import Extractor 

12 

13 

14if __name__ == "__main__": 

15 # Define command line arguments 

16 parser = argparse.ArgumentParser() 

17 parser.add_argument("--instance", required=True, type=str, 

18 help="path to instance file(s) to run on") 

19 parser.add_argument("--extractor", required=True, type=str, 

20 help="path to feature extractor") 

21 parser.add_argument("--feature-csv", required=True, type=str, 

22 help="path to feature data CSV file") 

23 parser.add_argument("--cutoff", required=True, type=str, 

24 help="the maximum CPU time for the extractor.") 

25 parser.add_argument("--feature-group", required=False, type=str, 

26 help="the group of features to compute, if available for the " 

27 "extractor. If not available or provided, all groups will" 

28 " be computed.") 

29 parser.add_argument("--log-dir", type=Path, required=False, 

30 help="path to the log directory") 

31 args = parser.parse_args() 

32 

33 # Process command line arguments 

34 log_dir =\ 

35 args.log_dir if args.log_dir is not None else gv.settings().DEFAULT_tmp_output 

36 instance_path = Path(args.instance) 

37 instance_name = instance_path 

38 if not instance_path.exists(): 

39 # If its an instance name (Multi-file instance), retrieve path list 

40 data_set = Instance_Set(instance_path.parent) 

41 instance_path = data_set.get_path_by_name(Path(instance_name).name) 

42 

43 extractor_path = Path(args.extractor) 

44 feature_data_csv_path = Path(args.feature_csv) 

45 cutoff_extractor = args.cutoff 

46 

47 # Ensure stringifcation of path objects 

48 if isinstance(instance_path, list): 

49 instance_list = [str(filepath) for filepath in instance_path] 

50 else: 

51 instance_list = [instance_path] 

52 

53 extractor = Extractor(extractor_path, 

54 gv.settings().DEFAULT_runsolver_exec, 

55 log_dir) 

56 features = extractor.run(instance_list, 

57 feature_group=args.feature_group, 

58 cutoff_time=cutoff_extractor, 

59 log_dir=log_dir) 

60 

61 # Now that we have our result, we write it to the FeatureDataCSV with a FileLock 

62 lock = FileLock(f"{feature_data_csv_path}.lock") 

63 if features is not None: 

64 print(f"Writing features to CSV: {instance_name}, {extractor_path.name}") 

65 with lock.acquire(timeout=60): 

66 feature_data = FeatureDataFrame(feature_data_csv_path) 

67 for feature_group, feature_name, value in features: 

68 feature_data.set_value(str(instance_name), extractor_path.name, 

69 feature_group, feature_name, float(value)) 

70 feature_data.save_csv() 

71 lock.release() 

72 else: 

73 print("EXCEPTION during retrieving extractor results.\n" 

74 f"****** WARNING: Feature vector computation on instance {instance_path}" 

75 " failed! ******")