We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
mlos_bench: read results file format tests
@poojanilangekar, could you please add some tests for the changes in #460 so that we make sure that the code accepts both input file formats we expect to be able to support?  Thanks!

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `mlos_bench/mlos_bench/environments/local/local_env.py`
Content:
```
1 #
2 # Copyright (c) Microsoft Corporation.
3 # Licensed under the MIT License.
4 #
5 """
6 Scheduler-side benchmark environment to run scripts locally.
7 """
8 
9 import json
10 import logging
11 import sys
12 
13 from datetime import datetime
14 from tempfile import TemporaryDirectory
15 from contextlib import nullcontext
16 
17 from types import TracebackType
18 from typing import Any, Dict, Iterable, List, Mapping, Optional, Tuple, Type, Union
19 from typing_extensions import Literal
20 
21 import pandas
22 
23 from mlos_bench.environments.status import Status
24 from mlos_bench.environments.base_environment import Environment
25 from mlos_bench.environments.script_env import ScriptEnv
26 from mlos_bench.services.base_service import Service
27 from mlos_bench.services.types.local_exec_type import SupportsLocalExec
28 from mlos_bench.tunables.tunable_groups import TunableGroups
29 from mlos_bench.util import path_join
30 
31 _LOG = logging.getLogger(__name__)
32 
33 
34 class LocalEnv(ScriptEnv):
35     # pylint: disable=too-many-instance-attributes
36     """
37     Scheduler-side Environment that runs scripts locally.
38     """
39 
40     def __init__(self,
41                  *,
42                  name: str,
43                  config: dict,
44                  global_config: Optional[dict] = None,
45                  tunables: Optional[TunableGroups] = None,
46                  service: Optional[Service] = None):
47         """
48         Create a new environment for local execution.
49 
50         Parameters
51         ----------
52         name: str
53             Human-readable name of the environment.
54         config : dict
55             Free-format dictionary that contains the benchmark environment
56             configuration. Each config must have at least the "tunable_params"
57             and the "const_args" sections.
58             `LocalEnv` must also have at least some of the following parameters:
59             {setup, run, teardown, dump_params_file, read_results_file}
60         global_config : dict
61             Free-format dictionary of global parameters (e.g., security credentials)
62             to be mixed in into the "const_args" section of the local config.
63         tunables : TunableGroups
64             A collection of tunable parameters for *all* environments.
65         service: Service
66             An optional service object (e.g., providing methods to
67             deploy or reboot a VM, etc.).
68         """
69         super().__init__(name=name, config=config, global_config=global_config,
70                          tunables=tunables, service=service)
71 
72         assert self._service is not None and isinstance(self._service, SupportsLocalExec), \
73             "LocalEnv requires a service that supports local execution"
74         self._local_exec_service: SupportsLocalExec = self._service
75 
76         self._temp_dir: Optional[str] = None
77         self._temp_dir_context: Union[TemporaryDirectory, nullcontext, None] = None
78 
79         self._dump_params_file: Optional[str] = self.config.get("dump_params_file")
80         self._dump_meta_file: Optional[str] = self.config.get("dump_meta_file")
81 
82         self._read_results_file: Optional[str] = self.config.get("read_results_file")
83         self._read_telemetry_file: Optional[str] = self.config.get("read_telemetry_file")
84 
85     def __enter__(self) -> Environment:
86         assert self._temp_dir is None and self._temp_dir_context is None
87         self._temp_dir_context = self._local_exec_service.temp_dir_context(self.config.get("temp_dir"))
88         self._temp_dir = self._temp_dir_context.__enter__()
89         return super().__enter__()
90 
91     def __exit__(self, ex_type: Optional[Type[BaseException]],
92                  ex_val: Optional[BaseException],
93                  ex_tb: Optional[TracebackType]) -> Literal[False]:
94         """
95         Exit the context of the benchmarking environment.
96         """
97         assert not (self._temp_dir is None or self._temp_dir_context is None)
98         self._temp_dir_context.__exit__(ex_type, ex_val, ex_tb)
99         self._temp_dir = None
100         self._temp_dir_context = None
101         return super().__exit__(ex_type, ex_val, ex_tb)
102 
103     def setup(self, tunables: TunableGroups, global_config: Optional[dict] = None) -> bool:
104         """
105         Check if the environment is ready and set up the application
106         and benchmarks, if necessary.
107 
108         Parameters
109         ----------
110         tunables : TunableGroups
111             A collection of tunable OS and application parameters along with their
112             values. In a local environment these could be used to prepare a config
113             file on the scheduler prior to transferring it to the remote environment,
114             for instance.
115         global_config : dict
116             Free-format dictionary of global parameters of the environment
117             that are not used in the optimization process.
118 
119         Returns
120         -------
121         is_success : bool
122             True if operation is successful, false otherwise.
123         """
124         if not super().setup(tunables, global_config):
125             return False
126 
127         _LOG.info("Set up the environment locally: '%s' at %s", self, self._temp_dir)
128         assert self._temp_dir is not None
129 
130         if self._dump_params_file:
131             fname = path_join(self._temp_dir, self._dump_params_file)
132             _LOG.debug("Dump tunables to file: %s", fname)
133             with open(fname, "wt", encoding="utf-8") as fh_tunables:
134                 # json.dump(self._params, fh_tunables)  # Tunables *and* const_args
135                 json.dump(self._tunable_params.get_param_values(), fh_tunables)
136 
137         if self._dump_meta_file:
138             fname = path_join(self._temp_dir, self._dump_meta_file)
139             _LOG.debug("Dump tunables metadata to file: %s", fname)
140             with open(fname, "wt", encoding="utf-8") as fh_meta:
141                 json.dump({
142                     tunable.name: tunable.meta
143                     for (tunable, _group) in self._tunable_params if tunable.meta
144                 }, fh_meta)
145 
146         if self._script_setup:
147             return_code = self._local_exec(self._script_setup, self._temp_dir)
148             self._is_ready = bool(return_code == 0)
149         else:
150             self._is_ready = True
151 
152         return self._is_ready
153 
154     def run(self) -> Tuple[Status, Optional[Dict[str, float]]]:
155         """
156         Run a script in the local scheduler environment.
157 
158         Returns
159         -------
160         (status, output) : (Status, dict)
161             A pair of (Status, output) values, where `output` is a dict
162             with the results or None if the status is not COMPLETED.
163             If run script is a benchmark, then the score is usually expected to
164             be in the `score` field.
165         """
166         (status, _) = result = super().run()
167         if not status.is_ready():
168             return result
169 
170         assert self._temp_dir is not None
171 
172         if self._script_run:
173             return_code = self._local_exec(self._script_run, self._temp_dir)
174             if return_code != 0:
175                 return (Status.FAILED, None)
176 
177         # FIXME: We should not be assuming that the only output file type is a CSV.
178         if not self._read_results_file:
179             _LOG.debug("Not reading the data at: %s", self)
180             return (Status.SUCCEEDED, {})
181 
182         data: pandas.DataFrame = pandas.read_csv(
183             self._config_loader_service.resolve_path(
184                 self._read_results_file, extra_paths=[self._temp_dir]))
185 
186         if sys.platform == 'win32':
187             data.rename(str.rstrip, axis='columns', inplace=True)
188 
189         _LOG.debug("Read data:\n%s", data)
190         if list(data.columns) == ["metric", "value"]:
191             _LOG.warning(
192                 "Local run has %d rows: assume long format of (metric, value)", len(data))
193             data = pandas.DataFrame([data.value.to_list()], columns=data.metric.to_list())
194 
195         data_dict = data.iloc[-1].to_dict()
196         _LOG.info("Local run complete: %s ::\n%s", self, data_dict)
197         return (Status.SUCCEEDED, data_dict) if data_dict else (Status.FAILED, None)
198 
199     def status(self) -> Tuple[Status, List[Tuple[datetime, str, Any]]]:
200 
201         (status, _) = super().status()
202         if not (self._is_ready and self._read_telemetry_file):
203             return (status, [])
204 
205         assert self._temp_dir is not None
206         try:
207             # FIXME: We should not be assuming that the only output file type is a CSV.
208             data: pandas.DataFrame = pandas.read_csv(
209                 self._config_loader_service.resolve_path(
210                     self._read_telemetry_file, extra_paths=[self._temp_dir]),
211                 parse_dates=[0],
212             )
213         except FileNotFoundError as ex:
214             _LOG.warning("Telemetry CSV file not found: %s :: %s", self._read_telemetry_file, ex)
215             return (status, [])
216 
217         if sys.platform == 'win32':
218             data.rename(str.rstrip, axis='columns', inplace=True)
219 
220         _LOG.debug("Read telemetry data:\n%s", data)
221         if list(data.columns) != ["timestamp", "metric", "value"]:
222             _LOG.warning(
223                 'Telemetry CSV file should have columns ["timestamp", "metric", "value"] :: %s',
224                 self._read_telemetry_file)
225 
226         col_dtypes: Mapping[int, Type] = {0: datetime}
227         return (status, [
228             (pandas.Timestamp(ts).to_pydatetime(), metric, value)
229             for (ts, metric, value) in data.to_records(index=False, column_dtypes=col_dtypes)
230         ])
231 
232     def teardown(self) -> None:
233         """
234         Clean up the local environment.
235         """
236         if self._script_teardown:
237             _LOG.info("Local teardown: %s", self)
238             return_code = self._local_exec(self._script_teardown)
239             _LOG.info("Local teardown complete: %s :: %s", self, return_code)
240         super().teardown()
241 
242     def _local_exec(self, script: Iterable[str], cwd: Optional[str] = None) -> int:
243         """
244         Execute a script locally in the scheduler environment.
245 
246         Parameters
247         ----------
248         script : Iterable[str]
249             Lines of the script to run locally.
250             Treat every line as a separate command to run.
251         cwd : Optional[str]
252             Work directory to run the script at.
253 
254         Returns
255         -------
256         return_code : int
257             Return code of the script. 0 if successful.
258         """
259         env_params = self._get_env_params()
260         _LOG.info("Run script locally on: %s at %s with env %s", self, cwd, env_params)
261         (return_code, _stdout, stderr) = self._local_exec_service.local_exec(
262             script, env=env_params, cwd=cwd)
263         if return_code != 0:
264             _LOG.warning("ERROR: Local script returns code %d stderr:\n%s", return_code, stderr)
265         return return_code
266
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/mlos_bench/mlos_bench/environments/local/local_env.py b/mlos_bench/mlos_bench/environments/local/local_env.py
--- a/mlos_bench/mlos_bench/environments/local/local_env.py
+++ b/mlos_bench/mlos_bench/environments/local/local_env.py
@@ -179,22 +179,36 @@
             _LOG.debug("Not reading the data at: %s", self)
             return (Status.SUCCEEDED, {})
 
-        data: pandas.DataFrame = pandas.read_csv(
+        data = self._normalize_columns(pandas.read_csv(
             self._config_loader_service.resolve_path(
-                self._read_results_file, extra_paths=[self._temp_dir]))
-
-        if sys.platform == 'win32':
-            data.rename(str.rstrip, axis='columns', inplace=True)
+                self._read_results_file, extra_paths=[self._temp_dir]),
+            index_col=False,
+        ))
 
         _LOG.debug("Read data:\n%s", data)
         if list(data.columns) == ["metric", "value"]:
-            _LOG.warning(
-                "Local run has %d rows: assume long format of (metric, value)", len(data))
+            _LOG.info("Local results have (metric,value) header and %d rows: assume long format", len(data))
             data = pandas.DataFrame([data.value.to_list()], columns=data.metric.to_list())
+        elif len(data) == 1:
+            _LOG.info("Local results have 1 row: assume wide format")
+        else:
+            raise ValueError(f"Invalid data format: {data}")
 
         data_dict = data.iloc[-1].to_dict()
         _LOG.info("Local run complete: %s ::\n%s", self, data_dict)
-        return (Status.SUCCEEDED, data_dict) if data_dict else (Status.FAILED, None)
+        return (Status.SUCCEEDED, data_dict)
+
+    @staticmethod
+    def _normalize_columns(data: pandas.DataFrame) -> pandas.DataFrame:
+        """
+        Strip trailing spaces from column names (Windows only).
+        """
+        # Windows cmd interpretation of > redirect symbols can leave trailing spaces in
+        # the final column, which leads to misnamed columns.
+        # For now, we simply strip trailing spaces from column names to account for that.
+        if sys.platform == 'win32':
+            data.rename(str.rstrip, axis='columns', inplace=True)
+        return data
 
     def status(self) -> Tuple[Status, List[Tuple[datetime, str, Any]]]:
 
@@ -204,25 +218,25 @@
 
         assert self._temp_dir is not None
         try:
+            fname = self._config_loader_service.resolve_path(
+                self._read_telemetry_file, extra_paths=[self._temp_dir])
+
             # FIXME: We should not be assuming that the only output file type is a CSV.
-            data: pandas.DataFrame = pandas.read_csv(
-                self._config_loader_service.resolve_path(
-                    self._read_telemetry_file, extra_paths=[self._temp_dir]),
-                parse_dates=[0],
-            )
+            data = self._normalize_columns(
+                pandas.read_csv(fname, index_col=False, parse_dates=[0]))
+
+            expected_col_names = ["timestamp", "metric", "value"]
+            if len(data.columns) != len(expected_col_names):
+                raise ValueError(f'Telemetry data must have columns {expected_col_names}')
+            elif list(data.columns) != expected_col_names:
+                # Assume no header - this is ok for telemetry data.
+                data = pandas.read_csv(
+                    fname, index_col=False, parse_dates=[0], names=expected_col_names)
         except FileNotFoundError as ex:
             _LOG.warning("Telemetry CSV file not found: %s :: %s", self._read_telemetry_file, ex)
             return (status, [])
 
-        if sys.platform == 'win32':
-            data.rename(str.rstrip, axis='columns', inplace=True)
-
         _LOG.debug("Read telemetry data:\n%s", data)
-        if list(data.columns) != ["timestamp", "metric", "value"]:
-            _LOG.warning(
-                'Telemetry CSV file should have columns ["timestamp", "metric", "value"] :: %s',
-                self._read_telemetry_file)
-
         col_dtypes: Mapping[int, Type] = {0: datetime}
         return (status, [
             (pandas.Timestamp(ts).to_pydatetime(), metric, value)

{"golden_diff": "diff --git a/mlos_bench/mlos_bench/environments/local/local_env.py b/mlos_bench/mlos_bench/environments/local/local_env.py\n--- a/mlos_bench/mlos_bench/environments/local/local_env.py\n+++ b/mlos_bench/mlos_bench/environments/local/local_env.py\n@@ -179,22 +179,36 @@\n             _LOG.debug(\"Not reading the data at: %s\", self)\n             return (Status.SUCCEEDED, {})\n \n-        data: pandas.DataFrame = pandas.read_csv(\n+        data = self._normalize_columns(pandas.read_csv(\n             self._config_loader_service.resolve_path(\n-                self._read_results_file, extra_paths=[self._temp_dir]))\n-\n-        if sys.platform == 'win32':\n-            data.rename(str.rstrip, axis='columns', inplace=True)\n+                self._read_results_file, extra_paths=[self._temp_dir]),\n+            index_col=False,\n+        ))\n \n         _LOG.debug(\"Read data:\\n%s\", data)\n         if list(data.columns) == [\"metric\", \"value\"]:\n-            _LOG.warning(\n-                \"Local run has %d rows: assume long format of (metric, value)\", len(data))\n+            _LOG.info(\"Local results have (metric,value) header and %d rows: assume long format\", len(data))\n             data = pandas.DataFrame([data.value.to_list()], columns=data.metric.to_list())\n+        elif len(data) == 1:\n+            _LOG.info(\"Local results have 1 row: assume wide format\")\n+        else:\n+            raise ValueError(f\"Invalid data format: {data}\")\n \n         data_dict = data.iloc[-1].to_dict()\n         _LOG.info(\"Local run complete: %s ::\\n%s\", self, data_dict)\n-        return (Status.SUCCEEDED, data_dict) if data_dict else (Status.FAILED, None)\n+        return (Status.SUCCEEDED, data_dict)\n+\n+    @staticmethod\n+    def _normalize_columns(data: pandas.DataFrame) -> pandas.DataFrame:\n+        \"\"\"\n+        Strip trailing spaces from column names (Windows only).\n+        \"\"\"\n+        # Windows cmd interpretation of > redirect symbols can leave trailing spaces in\n+        # the final column, which leads to misnamed columns.\n+        # For now, we simply strip trailing spaces from column names to account for that.\n+        if sys.platform == 'win32':\n+            data.rename(str.rstrip, axis='columns', inplace=True)\n+        return data\n \n     def status(self) -> Tuple[Status, List[Tuple[datetime, str, Any]]]:\n \n@@ -204,25 +218,25 @@\n \n         assert self._temp_dir is not None\n         try:\n+            fname = self._config_loader_service.resolve_path(\n+                self._read_telemetry_file, extra_paths=[self._temp_dir])\n+\n             # FIXME: We should not be assuming that the only output file type is a CSV.\n-            data: pandas.DataFrame = pandas.read_csv(\n-                self._config_loader_service.resolve_path(\n-                    self._read_telemetry_file, extra_paths=[self._temp_dir]),\n-                parse_dates=[0],\n-            )\n+            data = self._normalize_columns(\n+                pandas.read_csv(fname, index_col=False, parse_dates=[0]))\n+\n+            expected_col_names = [\"timestamp\", \"metric\", \"value\"]\n+            if len(data.columns) != len(expected_col_names):\n+                raise ValueError(f'Telemetry data must have columns {expected_col_names}')\n+            elif list(data.columns) != expected_col_names:\n+                # Assume no header - this is ok for telemetry data.\n+                data = pandas.read_csv(\n+                    fname, index_col=False, parse_dates=[0], names=expected_col_names)\n         except FileNotFoundError as ex:\n             _LOG.warning(\"Telemetry CSV file not found: %s :: %s\", self._read_telemetry_file, ex)\n             return (status, [])\n \n-        if sys.platform == 'win32':\n-            data.rename(str.rstrip, axis='columns', inplace=True)\n-\n         _LOG.debug(\"Read telemetry data:\\n%s\", data)\n-        if list(data.columns) != [\"timestamp\", \"metric\", \"value\"]:\n-            _LOG.warning(\n-                'Telemetry CSV file should have columns [\"timestamp\", \"metric\", \"value\"] :: %s',\n-                self._read_telemetry_file)\n-\n         col_dtypes: Mapping[int, Type] = {0: datetime}\n         return (status, [\n             (pandas.Timestamp(ts).to_pydatetime(), metric, value)\n", "issue": "mlos_bench: read results file format tests\n@poojanilangekar, could you please add some tests for the changes in #460 so that we make sure that the code accepts both input file formats we expect to be able to support?  Thanks!\n", "before_files": [{"content": "#\n# Copyright (c) Microsoft Corporation.\n# Licensed under the MIT License.\n#\n\"\"\"\nScheduler-side benchmark environment to run scripts locally.\n\"\"\"\n\nimport json\nimport logging\nimport sys\n\nfrom datetime import datetime\nfrom tempfile import TemporaryDirectory\nfrom contextlib import nullcontext\n\nfrom types import TracebackType\nfrom typing import Any, Dict, Iterable, List, Mapping, Optional, Tuple, Type, Union\nfrom typing_extensions import Literal\n\nimport pandas\n\nfrom mlos_bench.environments.status import Status\nfrom mlos_bench.environments.base_environment import Environment\nfrom mlos_bench.environments.script_env import ScriptEnv\nfrom mlos_bench.services.base_service import Service\nfrom mlos_bench.services.types.local_exec_type import SupportsLocalExec\nfrom mlos_bench.tunables.tunable_groups import TunableGroups\nfrom mlos_bench.util import path_join\n\n_LOG = logging.getLogger(__name__)\n\n\nclass LocalEnv(ScriptEnv):\n    # pylint: disable=too-many-instance-attributes\n    \"\"\"\n    Scheduler-side Environment that runs scripts locally.\n    \"\"\"\n\n    def __init__(self,\n                 *,\n                 name: str,\n                 config: dict,\n                 global_config: Optional[dict] = None,\n                 tunables: Optional[TunableGroups] = None,\n                 service: Optional[Service] = None):\n        \"\"\"\n        Create a new environment for local execution.\n\n        Parameters\n        ----------\n        name: str\n            Human-readable name of the environment.\n        config : dict\n            Free-format dictionary that contains the benchmark environment\n            configuration. Each config must have at least the \"tunable_params\"\n            and the \"const_args\" sections.\n            `LocalEnv` must also have at least some of the following parameters:\n            {setup, run, teardown, dump_params_file, read_results_file}\n        global_config : dict\n            Free-format dictionary of global parameters (e.g., security credentials)\n            to be mixed in into the \"const_args\" section of the local config.\n        tunables : TunableGroups\n            A collection of tunable parameters for *all* environments.\n        service: Service\n            An optional service object (e.g., providing methods to\n            deploy or reboot a VM, etc.).\n        \"\"\"\n        super().__init__(name=name, config=config, global_config=global_config,\n                         tunables=tunables, service=service)\n\n        assert self._service is not None and isinstance(self._service, SupportsLocalExec), \\\n            \"LocalEnv requires a service that supports local execution\"\n        self._local_exec_service: SupportsLocalExec = self._service\n\n        self._temp_dir: Optional[str] = None\n        self._temp_dir_context: Union[TemporaryDirectory, nullcontext, None] = None\n\n        self._dump_params_file: Optional[str] = self.config.get(\"dump_params_file\")\n        self._dump_meta_file: Optional[str] = self.config.get(\"dump_meta_file\")\n\n        self._read_results_file: Optional[str] = self.config.get(\"read_results_file\")\n        self._read_telemetry_file: Optional[str] = self.config.get(\"read_telemetry_file\")\n\n    def __enter__(self) -> Environment:\n        assert self._temp_dir is None and self._temp_dir_context is None\n        self._temp_dir_context = self._local_exec_service.temp_dir_context(self.config.get(\"temp_dir\"))\n        self._temp_dir = self._temp_dir_context.__enter__()\n        return super().__enter__()\n\n    def __exit__(self, ex_type: Optional[Type[BaseException]],\n                 ex_val: Optional[BaseException],\n                 ex_tb: Optional[TracebackType]) -> Literal[False]:\n        \"\"\"\n        Exit the context of the benchmarking environment.\n        \"\"\"\n        assert not (self._temp_dir is None or self._temp_dir_context is None)\n        self._temp_dir_context.__exit__(ex_type, ex_val, ex_tb)\n        self._temp_dir = None\n        self._temp_dir_context = None\n        return super().__exit__(ex_type, ex_val, ex_tb)\n\n    def setup(self, tunables: TunableGroups, global_config: Optional[dict] = None) -> bool:\n        \"\"\"\n        Check if the environment is ready and set up the application\n        and benchmarks, if necessary.\n\n        Parameters\n        ----------\n        tunables : TunableGroups\n            A collection of tunable OS and application parameters along with their\n            values. In a local environment these could be used to prepare a config\n            file on the scheduler prior to transferring it to the remote environment,\n            for instance.\n        global_config : dict\n            Free-format dictionary of global parameters of the environment\n            that are not used in the optimization process.\n\n        Returns\n        -------\n        is_success : bool\n            True if operation is successful, false otherwise.\n        \"\"\"\n        if not super().setup(tunables, global_config):\n            return False\n\n        _LOG.info(\"Set up the environment locally: '%s' at %s\", self, self._temp_dir)\n        assert self._temp_dir is not None\n\n        if self._dump_params_file:\n            fname = path_join(self._temp_dir, self._dump_params_file)\n            _LOG.debug(\"Dump tunables to file: %s\", fname)\n            with open(fname, \"wt\", encoding=\"utf-8\") as fh_tunables:\n                # json.dump(self._params, fh_tunables)  # Tunables *and* const_args\n                json.dump(self._tunable_params.get_param_values(), fh_tunables)\n\n        if self._dump_meta_file:\n            fname = path_join(self._temp_dir, self._dump_meta_file)\n            _LOG.debug(\"Dump tunables metadata to file: %s\", fname)\n            with open(fname, \"wt\", encoding=\"utf-8\") as fh_meta:\n                json.dump({\n                    tunable.name: tunable.meta\n                    for (tunable, _group) in self._tunable_params if tunable.meta\n                }, fh_meta)\n\n        if self._script_setup:\n            return_code = self._local_exec(self._script_setup, self._temp_dir)\n            self._is_ready = bool(return_code == 0)\n        else:\n            self._is_ready = True\n\n        return self._is_ready\n\n    def run(self) -> Tuple[Status, Optional[Dict[str, float]]]:\n        \"\"\"\n        Run a script in the local scheduler environment.\n\n        Returns\n        -------\n        (status, output) : (Status, dict)\n            A pair of (Status, output) values, where `output` is a dict\n            with the results or None if the status is not COMPLETED.\n            If run script is a benchmark, then the score is usually expected to\n            be in the `score` field.\n        \"\"\"\n        (status, _) = result = super().run()\n        if not status.is_ready():\n            return result\n\n        assert self._temp_dir is not None\n\n        if self._script_run:\n            return_code = self._local_exec(self._script_run, self._temp_dir)\n            if return_code != 0:\n                return (Status.FAILED, None)\n\n        # FIXME: We should not be assuming that the only output file type is a CSV.\n        if not self._read_results_file:\n            _LOG.debug(\"Not reading the data at: %s\", self)\n            return (Status.SUCCEEDED, {})\n\n        data: pandas.DataFrame = pandas.read_csv(\n            self._config_loader_service.resolve_path(\n                self._read_results_file, extra_paths=[self._temp_dir]))\n\n        if sys.platform == 'win32':\n            data.rename(str.rstrip, axis='columns', inplace=True)\n\n        _LOG.debug(\"Read data:\\n%s\", data)\n        if list(data.columns) == [\"metric\", \"value\"]:\n            _LOG.warning(\n                \"Local run has %d rows: assume long format of (metric, value)\", len(data))\n            data = pandas.DataFrame([data.value.to_list()], columns=data.metric.to_list())\n\n        data_dict = data.iloc[-1].to_dict()\n        _LOG.info(\"Local run complete: %s ::\\n%s\", self, data_dict)\n        return (Status.SUCCEEDED, data_dict) if data_dict else (Status.FAILED, None)\n\n    def status(self) -> Tuple[Status, List[Tuple[datetime, str, Any]]]:\n\n        (status, _) = super().status()\n        if not (self._is_ready and self._read_telemetry_file):\n            return (status, [])\n\n        assert self._temp_dir is not None\n        try:\n            # FIXME: We should not be assuming that the only output file type is a CSV.\n            data: pandas.DataFrame = pandas.read_csv(\n                self._config_loader_service.resolve_path(\n                    self._read_telemetry_file, extra_paths=[self._temp_dir]),\n                parse_dates=[0],\n            )\n        except FileNotFoundError as ex:\n            _LOG.warning(\"Telemetry CSV file not found: %s :: %s\", self._read_telemetry_file, ex)\n            return (status, [])\n\n        if sys.platform == 'win32':\n            data.rename(str.rstrip, axis='columns', inplace=True)\n\n        _LOG.debug(\"Read telemetry data:\\n%s\", data)\n        if list(data.columns) != [\"timestamp\", \"metric\", \"value\"]:\n            _LOG.warning(\n                'Telemetry CSV file should have columns [\"timestamp\", \"metric\", \"value\"] :: %s',\n                self._read_telemetry_file)\n\n        col_dtypes: Mapping[int, Type] = {0: datetime}\n        return (status, [\n            (pandas.Timestamp(ts).to_pydatetime(), metric, value)\n            for (ts, metric, value) in data.to_records(index=False, column_dtypes=col_dtypes)\n        ])\n\n    def teardown(self) -> None:\n        \"\"\"\n        Clean up the local environment.\n        \"\"\"\n        if self._script_teardown:\n            _LOG.info(\"Local teardown: %s\", self)\n            return_code = self._local_exec(self._script_teardown)\n            _LOG.info(\"Local teardown complete: %s :: %s\", self, return_code)\n        super().teardown()\n\n    def _local_exec(self, script: Iterable[str], cwd: Optional[str] = None) -> int:\n        \"\"\"\n        Execute a script locally in the scheduler environment.\n\n        Parameters\n        ----------\n        script : Iterable[str]\n            Lines of the script to run locally.\n            Treat every line as a separate command to run.\n        cwd : Optional[str]\n            Work directory to run the script at.\n\n        Returns\n        -------\n        return_code : int\n            Return code of the script. 0 if successful.\n        \"\"\"\n        env_params = self._get_env_params()\n        _LOG.info(\"Run script locally on: %s at %s with env %s\", self, cwd, env_params)\n        (return_code, _stdout, stderr) = self._local_exec_service.local_exec(\n            script, env=env_params, cwd=cwd)\n        if return_code != 0:\n            _LOG.warning(\"ERROR: Local script returns code %d stderr:\\n%s\", return_code, stderr)\n        return return_code\n", "path": "mlos_bench/mlos_bench/environments/local/local_env.py"}], "after_files": [{"content": "#\n# Copyright (c) Microsoft Corporation.\n# Licensed under the MIT License.\n#\n\"\"\"\nScheduler-side benchmark environment to run scripts locally.\n\"\"\"\n\nimport json\nimport logging\nimport sys\n\nfrom datetime import datetime\nfrom tempfile import TemporaryDirectory\nfrom contextlib import nullcontext\n\nfrom types import TracebackType\nfrom typing import Any, Dict, Iterable, List, Mapping, Optional, Tuple, Type, Union\nfrom typing_extensions import Literal\n\nimport pandas\n\nfrom mlos_bench.environments.status import Status\nfrom mlos_bench.environments.base_environment import Environment\nfrom mlos_bench.environments.script_env import ScriptEnv\nfrom mlos_bench.services.base_service import Service\nfrom mlos_bench.services.types.local_exec_type import SupportsLocalExec\nfrom mlos_bench.tunables.tunable_groups import TunableGroups\nfrom mlos_bench.util import path_join\n\n_LOG = logging.getLogger(__name__)\n\n\nclass LocalEnv(ScriptEnv):\n    # pylint: disable=too-many-instance-attributes\n    \"\"\"\n    Scheduler-side Environment that runs scripts locally.\n    \"\"\"\n\n    def __init__(self,\n                 *,\n                 name: str,\n                 config: dict,\n                 global_config: Optional[dict] = None,\n                 tunables: Optional[TunableGroups] = None,\n                 service: Optional[Service] = None):\n        \"\"\"\n        Create a new environment for local execution.\n\n        Parameters\n        ----------\n        name: str\n            Human-readable name of the environment.\n        config : dict\n            Free-format dictionary that contains the benchmark environment\n            configuration. Each config must have at least the \"tunable_params\"\n            and the \"const_args\" sections.\n            `LocalEnv` must also have at least some of the following parameters:\n            {setup, run, teardown, dump_params_file, read_results_file}\n        global_config : dict\n            Free-format dictionary of global parameters (e.g., security credentials)\n            to be mixed in into the \"const_args\" section of the local config.\n        tunables : TunableGroups\n            A collection of tunable parameters for *all* environments.\n        service: Service\n            An optional service object (e.g., providing methods to\n            deploy or reboot a VM, etc.).\n        \"\"\"\n        super().__init__(name=name, config=config, global_config=global_config,\n                         tunables=tunables, service=service)\n\n        assert self._service is not None and isinstance(self._service, SupportsLocalExec), \\\n            \"LocalEnv requires a service that supports local execution\"\n        self._local_exec_service: SupportsLocalExec = self._service\n\n        self._temp_dir: Optional[str] = None\n        self._temp_dir_context: Union[TemporaryDirectory, nullcontext, None] = None\n\n        self._dump_params_file: Optional[str] = self.config.get(\"dump_params_file\")\n        self._dump_meta_file: Optional[str] = self.config.get(\"dump_meta_file\")\n\n        self._read_results_file: Optional[str] = self.config.get(\"read_results_file\")\n        self._read_telemetry_file: Optional[str] = self.config.get(\"read_telemetry_file\")\n\n    def __enter__(self) -> Environment:\n        assert self._temp_dir is None and self._temp_dir_context is None\n        self._temp_dir_context = self._local_exec_service.temp_dir_context(self.config.get(\"temp_dir\"))\n        self._temp_dir = self._temp_dir_context.__enter__()\n        return super().__enter__()\n\n    def __exit__(self, ex_type: Optional[Type[BaseException]],\n                 ex_val: Optional[BaseException],\n                 ex_tb: Optional[TracebackType]) -> Literal[False]:\n        \"\"\"\n        Exit the context of the benchmarking environment.\n        \"\"\"\n        assert not (self._temp_dir is None or self._temp_dir_context is None)\n        self._temp_dir_context.__exit__(ex_type, ex_val, ex_tb)\n        self._temp_dir = None\n        self._temp_dir_context = None\n        return super().__exit__(ex_type, ex_val, ex_tb)\n\n    def setup(self, tunables: TunableGroups, global_config: Optional[dict] = None) -> bool:\n        \"\"\"\n        Check if the environment is ready and set up the application\n        and benchmarks, if necessary.\n\n        Parameters\n        ----------\n        tunables : TunableGroups\n            A collection of tunable OS and application parameters along with their\n            values. In a local environment these could be used to prepare a config\n            file on the scheduler prior to transferring it to the remote environment,\n            for instance.\n        global_config : dict\n            Free-format dictionary of global parameters of the environment\n            that are not used in the optimization process.\n\n        Returns\n        -------\n        is_success : bool\n            True if operation is successful, false otherwise.\n        \"\"\"\n        if not super().setup(tunables, global_config):\n            return False\n\n        _LOG.info(\"Set up the environment locally: '%s' at %s\", self, self._temp_dir)\n        assert self._temp_dir is not None\n\n        if self._dump_params_file:\n            fname = path_join(self._temp_dir, self._dump_params_file)\n            _LOG.debug(\"Dump tunables to file: %s\", fname)\n            with open(fname, \"wt\", encoding=\"utf-8\") as fh_tunables:\n                # json.dump(self._params, fh_tunables)  # Tunables *and* const_args\n                json.dump(self._tunable_params.get_param_values(), fh_tunables)\n\n        if self._dump_meta_file:\n            fname = path_join(self._temp_dir, self._dump_meta_file)\n            _LOG.debug(\"Dump tunables metadata to file: %s\", fname)\n            with open(fname, \"wt\", encoding=\"utf-8\") as fh_meta:\n                json.dump({\n                    tunable.name: tunable.meta\n                    for (tunable, _group) in self._tunable_params if tunable.meta\n                }, fh_meta)\n\n        if self._script_setup:\n            return_code = self._local_exec(self._script_setup, self._temp_dir)\n            self._is_ready = bool(return_code == 0)\n        else:\n            self._is_ready = True\n\n        return self._is_ready\n\n    def run(self) -> Tuple[Status, Optional[Dict[str, float]]]:\n        \"\"\"\n        Run a script in the local scheduler environment.\n\n        Returns\n        -------\n        (status, output) : (Status, dict)\n            A pair of (Status, output) values, where `output` is a dict\n            with the results or None if the status is not COMPLETED.\n            If run script is a benchmark, then the score is usually expected to\n            be in the `score` field.\n        \"\"\"\n        (status, _) = result = super().run()\n        if not status.is_ready():\n            return result\n\n        assert self._temp_dir is not None\n\n        if self._script_run:\n            return_code = self._local_exec(self._script_run, self._temp_dir)\n            if return_code != 0:\n                return (Status.FAILED, None)\n\n        # FIXME: We should not be assuming that the only output file type is a CSV.\n        if not self._read_results_file:\n            _LOG.debug(\"Not reading the data at: %s\", self)\n            return (Status.SUCCEEDED, {})\n\n        data = self._normalize_columns(pandas.read_csv(\n            self._config_loader_service.resolve_path(\n                self._read_results_file, extra_paths=[self._temp_dir]),\n            index_col=False,\n        ))\n\n        _LOG.debug(\"Read data:\\n%s\", data)\n        if list(data.columns) == [\"metric\", \"value\"]:\n            _LOG.info(\"Local results have (metric,value) header and %d rows: assume long format\", len(data))\n            data = pandas.DataFrame([data.value.to_list()], columns=data.metric.to_list())\n        elif len(data) == 1:\n            _LOG.info(\"Local results have 1 row: assume wide format\")\n        else:\n            raise ValueError(f\"Invalid data format: {data}\")\n\n        data_dict = data.iloc[-1].to_dict()\n        _LOG.info(\"Local run complete: %s ::\\n%s\", self, data_dict)\n        return (Status.SUCCEEDED, data_dict)\n\n    @staticmethod\n    def _normalize_columns(data: pandas.DataFrame) -> pandas.DataFrame:\n        \"\"\"\n        Strip trailing spaces from column names (Windows only).\n        \"\"\"\n        # Windows cmd interpretation of > redirect symbols can leave trailing spaces in\n        # the final column, which leads to misnamed columns.\n        # For now, we simply strip trailing spaces from column names to account for that.\n        if sys.platform == 'win32':\n            data.rename(str.rstrip, axis='columns', inplace=True)\n        return data\n\n    def status(self) -> Tuple[Status, List[Tuple[datetime, str, Any]]]:\n\n        (status, _) = super().status()\n        if not (self._is_ready and self._read_telemetry_file):\n            return (status, [])\n\n        assert self._temp_dir is not None\n        try:\n            fname = self._config_loader_service.resolve_path(\n                self._read_telemetry_file, extra_paths=[self._temp_dir])\n\n            # FIXME: We should not be assuming that the only output file type is a CSV.\n            data = self._normalize_columns(\n                pandas.read_csv(fname, index_col=False, parse_dates=[0]))\n\n            expected_col_names = [\"timestamp\", \"metric\", \"value\"]\n            if len(data.columns) != len(expected_col_names):\n                raise ValueError(f'Telemetry data must have columns {expected_col_names}')\n            elif list(data.columns) != expected_col_names:\n                # Assume no header - this is ok for telemetry data.\n                data = pandas.read_csv(\n                    fname, index_col=False, parse_dates=[0], names=expected_col_names)\n        except FileNotFoundError as ex:\n            _LOG.warning(\"Telemetry CSV file not found: %s :: %s\", self._read_telemetry_file, ex)\n            return (status, [])\n\n        _LOG.debug(\"Read telemetry data:\\n%s\", data)\n        col_dtypes: Mapping[int, Type] = {0: datetime}\n        return (status, [\n            (pandas.Timestamp(ts).to_pydatetime(), metric, value)\n            for (ts, metric, value) in data.to_records(index=False, column_dtypes=col_dtypes)\n        ])\n\n    def teardown(self) -> None:\n        \"\"\"\n        Clean up the local environment.\n        \"\"\"\n        if self._script_teardown:\n            _LOG.info(\"Local teardown: %s\", self)\n            return_code = self._local_exec(self._script_teardown)\n            _LOG.info(\"Local teardown complete: %s :: %s\", self, return_code)\n        super().teardown()\n\n    def _local_exec(self, script: Iterable[str], cwd: Optional[str] = None) -> int:\n        \"\"\"\n        Execute a script locally in the scheduler environment.\n\n        Parameters\n        ----------\n        script : Iterable[str]\n            Lines of the script to run locally.\n            Treat every line as a separate command to run.\n        cwd : Optional[str]\n            Work directory to run the script at.\n\n        Returns\n        -------\n        return_code : int\n            Return code of the script. 0 if successful.\n        \"\"\"\n        env_params = self._get_env_params()\n        _LOG.info(\"Run script locally on: %s at %s with env %s\", self, cwd, env_params)\n        (return_code, _stdout, stderr) = self._local_exec_service.local_exec(\n            script, env=env_params, cwd=cwd)\n        if return_code != 0:\n            _LOG.warning(\"ERROR: Local script returns code %d stderr:\\n%s\", return_code, stderr)\n        return return_code\n", "path": "mlos_bench/mlos_bench/environments/local/local_env.py"}]}