weather-impact

1
import logging
2
import os
3
import subprocess
4
import sys
5
from typing import List
6

7
# Constants
8
LOG_FORMAT = "%(asctime)s - %(levelname)s - %(message)s"
9

10
# Set up logging to stdout
11
logging.basicConfig(
12
    level=logging.INFO, format=LOG_FORMAT, handlers=[logging.StreamHandler(sys.stdout)]
13
)
14

15

16
def get_command_path(command_name: str) -> str:
17
    """Return the full path to the command executable."""
18
    return os.popen(f"which {command_name}").read().strip()
19

20

21
def run_command(command: List[str]) -> None:
22
    """Execute a command using subprocess and log its output."""
23
    try:
24
        result = subprocess.run(command, capture_output=True, text=True)
25
        logging.info(f"'{command}' output: {result.stdout}")
26
        if result.stderr:
27
            logging.error(f"'{command}' error: {result.stderr}")
28
    except Exception as e:
29
        logging.error(f"Failed to run '{command}': {e}")
30

31

32
def run_polaris_fetch(
33
    sat_name: str, start_date: str, end_date: str, data_dir_path: str
34
) -> None:
35
    """Fetch satellite data using Polaris."""
36
    json_name = f"{sat_name.lower()}_normalized_frames.json"
37
    cache_dir = os.path.join(data_dir_path, sat_name.lower())
38
    cache_dir = os.path.abspath(cache_dir)
39

40
    command = [
41
        get_command_path("polaris"),
42
        "fetch",
43
        "--start_date",
44
        start_date,
45
        "--end_date",
46
        end_date,
47
        "--cache_dir",
48
        cache_dir,
49
        sat_name,
50
        os.path.join(cache_dir, json_name),
51
    ]
52
    run_command(command)
53

54

55
def run_polaris_learn(sat_name: str, data_dir_path: str) -> None:
56
    """Analyze satellite data using Polaris."""
57
    graph_name = f"{sat_name.lower()}-graph.json"
58
    cache_dir = os.path.join(data_dir_path, sat_name.lower())
59
    cache_dir = os.path.abspath(cache_dir)
60

61
    command = [
62
        get_command_path("polaris"),
63
        "learn",
64
        "--force_cpu",
65
        "--output_graph_file",
66
        os.path.join(cache_dir, graph_name),
67
        os.path.join(cache_dir, f"{sat_name.lower()}_normalized_frames.json"),
68
    ]
69
    run_command(command)
70

71

72
def run_polaris_behave(sat_name: str, data_dir_path: str) -> None:
73
    """Analyze satellite behavior using Polaris."""
74
    input_file = os.path.join(
75
        data_dir_path, sat_name.lower(), f"{sat_name.lower()}_normalized_frames.json"
76
    )
77
    output_file = os.path.join(
78
        data_dir_path, sat_name.lower(), f"{sat_name.lower()}-anomaly_analysis.json"
79
    )
80
    command = [
81
        get_command_path("polaris"),
82
        "behave",
83
        input_file,
84
        "--output_file",
85
        output_file,
86
    ]
87
    run_command(command)
88

89

90
def main() -> None:
91
    sat_name: str = "LightSail-2"
92
    start_date: str = "2019-07-07"
93
    end_date: str = "2020-11-15"
94
    data_dir_path: str = "../data"
95

96
    run_polaris_fetch(sat_name, start_date, end_date, data_dir_path)
97
    run_polaris_learn(sat_name, data_dir_path)
98
    run_polaris_behave(sat_name, data_dir_path)
99

100

101
if __name__ == "__main__":
102
    main()
103