v0.1 update citi bike doc and data version

docs/source/scenarios/citi_bike.rst

@@ -515,8 +515,17 @@ Quick Start
 Data Preparation
 ^^^^^^^^^^^^^^^^
 
-To start the simulation of Citi Bike scenario, users need to first generate the
-related data. Below is the introduction to the related commands:
+To start the simulation of Citi Bike scenario, users have two options for the data preparation:
+
+* If the topology data is not generated in advance, the system will automatically download and
+  process the relevant data when the environment is created. The data will be stored in a
+  temporary folder and be automatically deleted after the experiment.
+
+* Before creating the environment, users can also manually download and generate relevant data.
+  This approach will save you a lot of time if you need to conduct several experiments on the
+  same topology. Therefore, we encourage you to generate the relevant data manually first.
+
+The following is the introduction to related commands:
 
 Environment List Command
 ~~~~~~~~~~~~~~~~~~~~~~~~
@@ -563,19 +572,19 @@ For the example above, the directory structure should be like:
    |-- ~/.maro
        |-- data
        |   |-- citi_bike
-       |       |-- [topology]
-       |           |-- .build          # bin data file
-       |           |-- .source
-       |               |-- .download   # original data file
-       |               |-- .clean      # cleaned data file
-       |-- temp                        # download temp file
-
-Convert Command
-~~~~~~~~~~~~~~~
+       |       |-- .build          # bin data file
+       |           |-- [topology]  # topology
+       |       |-- .source
+       |           |-- .download   # original data file
+       |           |-- .clean      # cleaned data file
+       |-- temp                    # download temp file
+
+Build Command
+~~~~~~~~~~~~~
 
-The data ``convert`` command is used to convert the CSV data files to binary data
+The data ``build`` command is used to build the CSV data files to binary data
 files that the simulator needs. Currently, there are three arguments for the data
-``convert`` command:
+``build`` command:
 
 * ``--meta``\ : required, used to specify the path of the meta file. The source
   columns that to be converted and the data type of each columns should be
@@ -587,7 +596,7 @@ files that the simulator needs. Currently, there are three arguments for the dat
 
 .. code-block:: sh
 
-   maro data convert --meta ~/.maro/data/citibike/meta/trips.yml --file ~/.maro/data/citibike/source/_clean/ny201801/trip.csv --output ~/.maro/data/citibike/_build/ny201801/trip.bin
+   maro data build --meta ~/.maro/data/citibike/meta/trips.yml --file ~/.maro/data/citibike/source/_clean/ny201801/trip.csv --output ~/.maro/data/citibike/_build/ny201801/trip.bin
 
 Environment Interface
 ^^^^^^^^^^^^^^^^^^^^^

maro/__misc__.py

@@ -4,4 +4,4 @@
 
 __version__ = "0.1.1a11"
 
-__data_version__ = "0.1"
+__data_version__ = "0.2"

maro/cli/grass/executors/grass_azure_executor.py

@@ -19,8 +19,10 @@
 from maro.cli.utils.details import (load_cluster_details, save_cluster_details, load_job_details, save_job_details,
                                     load_schedule_details, save_schedule_details)
 from maro.cli.utils.executors.azure_executor import AzureExecutor
-from maro.cli.utils.naming import (generate_cluster_id, generate_job_id, generate_component_id, generate_node_name,
-                                   get_valid_file_name)
+from maro.cli.utils.naming import (
+    generate_cluster_id, generate_job_id, generate_component_id, generate_node_name,
+    get_valid_file_name
+)
 from maro.cli.utils.params import GlobalParams, GlobalPaths
 from maro.cli.utils.subprocess import SubProcess
 from maro.cli.utils.validation import validate_and_fill_dict
@@ -194,18 +196,30 @@ def _init_master(self):
         self.grass_executor.retry_until_connected(node_ip_address=master_public_ip_address)
 
         # Create folders
-        sync_mkdir(remote_path=GlobalPaths.MARO_GRASS_LIB,
-                   admin_username=admin_username, node_ip_address=master_public_ip_address)
-        sync_mkdir(remote_path=f"{GlobalPaths.MARO_CLUSTERS}/{self.cluster_name}",
-                   admin_username=admin_username, node_ip_address=master_public_ip_address)
-        sync_mkdir(remote_path=f"{GlobalPaths.MARO_CLUSTERS}/{self.cluster_name}/data",
-                   admin_username=admin_username, node_ip_address=master_public_ip_address)
-        sync_mkdir(remote_path=f"{GlobalPaths.MARO_CLUSTERS}/{self.cluster_name}/images",
-                   admin_username=admin_username, node_ip_address=master_public_ip_address)
-        sync_mkdir(remote_path=f"{GlobalPaths.MARO_CLUSTERS}/{self.cluster_name}/jobs",
-                   admin_username=admin_username, node_ip_address=master_public_ip_address)
-        sync_mkdir(remote_path=f"{GlobalPaths.MARO_CLUSTERS}/{self.cluster_name}/schedules",
-                   admin_username=admin_username, node_ip_address=master_public_ip_address)
+        sync_mkdir(
+            remote_path=GlobalPaths.MARO_GRASS_LIB,
+            admin_username=admin_username, node_ip_address=master_public_ip_address
+        )
+        sync_mkdir(
+            remote_path=f"{GlobalPaths.MARO_CLUSTERS}/{self.cluster_name}",
+            admin_username=admin_username, node_ip_address=master_public_ip_address
+        )
+        sync_mkdir(
+            remote_path=f"{GlobalPaths.MARO_CLUSTERS}/{self.cluster_name}/data",
+            admin_username=admin_username, node_ip_address=master_public_ip_address
+        )
+        sync_mkdir(
+            remote_path=f"{GlobalPaths.MARO_CLUSTERS}/{self.cluster_name}/images",
+            admin_username=admin_username, node_ip_address=master_public_ip_address
+        )
+        sync_mkdir(
+            remote_path=f"{GlobalPaths.MARO_CLUSTERS}/{self.cluster_name}/jobs",
+            admin_username=admin_username, node_ip_address=master_public_ip_address
+        )
+        sync_mkdir(
+            remote_path=f"{GlobalPaths.MARO_CLUSTERS}/{self.cluster_name}/schedules",
+            admin_username=admin_username, node_ip_address=master_public_ip_address
+        )
 
         # Copy required files
         copy_files_to_node(
@@ -698,8 +712,10 @@ def _count_running_containers(node_details: dict):
 
     # maro grass image
 
-    def push_image(self, image_name: str, image_path: str, remote_context_path: str,
-                   remote_image_name: str):
+    def push_image(
+        self, image_name: str, image_path: str, remote_context_path: str,
+        remote_image_name: str
+    ):
         # Load details
         cluster_details = self.cluster_details
         admin_username = cluster_details['user']['admin_username']

maro/cli/k8s/executors/k8s_azure_executor.py

@@ -239,8 +239,8 @@ def _create_k8s_secret(self):
 
         # Create k8s secret
         command = f'kubectl create secret generic {cluster_id}-k8s-secret ' \
-                  f'--from-literal=azurestorageaccountname={cluster_id}st ' \
-                  f'--from-literal=azurestorageaccountkey={storage_key}'
+            f'--from-literal=azurestorageaccountname={cluster_id}st ' \
+            f'--from-literal=azurestorageaccountkey={storage_key}'
         _ = SubProcess.run(command)
         logger.debug(command)
 
@@ -442,9 +442,9 @@ def push_data(self, local_path: str, remote_dir: str):
         if not target_dir.startswith("/"):
             raise CliException("Invalid remote path")
         copy_command = f'azcopy copy ' \
-                       f'"{source_path}" ' \
-                       f'"https://{cluster_id}st.file.core.windows.net/{cluster_id}-fs{target_dir}?{sas}" ' \
-                       f'--recursive=True'
+            f'"{source_path}" ' \
+            f'"https://{cluster_id}st.file.core.windows.net/{cluster_id}-fs{target_dir}?{sas}" ' \
+            f'--recursive=True'
         _ = SubProcess.run(copy_command)
 
     def pull_data(self, local_dir: str, remote_path: str):
@@ -464,9 +464,9 @@ def pull_data(self, local_dir: str, remote_path: str):
         if not source_path.startswith("/"):
             raise CliException("Invalid remote path")
         copy_command = f'azcopy copy ' \
-                       f'"https://{cluster_id}st.file.core.windows.net/{cluster_id}-fs{source_path}?{sas}" ' \
-                       f'"{os.path.expanduser(target_dir)}" ' \
-                       f'--recursive=True'
+            f'"https://{cluster_id}st.file.core.windows.net/{cluster_id}-fs{source_path}?{sas}" ' \
+            f'"{os.path.expanduser(target_dir)}" ' \
+            f'--recursive=True'
         _ = SubProcess.run(copy_command)
 
     def remove_data(self, remote_path: str):
@@ -481,8 +481,8 @@ def remove_data(self, remote_path: str):
 
         # Remove data
         copy_command = f'azcopy remove ' \
-                       f'"https://{cluster_id}st.file.core.windows.net/{cluster_id}-fs{remote_path}?{sas}" ' \
-                       f'--recursive=True'
+            f'"https://{cluster_id}st.file.core.windows.net/{cluster_id}-fs{remote_path}?{sas}" ' \
+            f'--recursive=True'
         _ = SubProcess.run(copy_command)
 
     def _check_and_get_account_sas(self):
@@ -545,13 +545,13 @@ def _start_job(self, job_details: dict):
 
         # Apply k8s config
         command = f"kubectl apply -f " \
-                  f"{GlobalPaths.MARO_CLUSTERS}/{self.cluster_name}/jobs/{job_name}/k8s_configs/jobs.yml"
+            f"{GlobalPaths.MARO_CLUSTERS}/{self.cluster_name}/jobs/{job_name}/k8s_configs/jobs.yml"
         _ = SubProcess.run(command)
 
     def stop_job(self, job_name: str):
         # Stop job
         command = f"kubectl delete -f " \
-                  f"{GlobalPaths.MARO_CLUSTERS}/{self.cluster_name}/jobs/{job_name}/k8s_configs/jobs.yml"
+            f"{GlobalPaths.MARO_CLUSTERS}/{self.cluster_name}/jobs/{job_name}/k8s_configs/jobs.yml"
         _ = SubProcess.run(command)
 
     @staticmethod
@@ -613,8 +613,10 @@ def _create_k8s_job_config(self, job_name: str) -> dict:
 
         return k8s_job_config
 
-    def _create_k8s_container_config(self, job_details: dict, k8s_container_config_template: dict,
-                                     component_type: str, component_index: int):
+    def _create_k8s_container_config(
+        self, job_details: dict, k8s_container_config_template: dict,
+        component_type: str, component_index: int
+    ):
         # Copy config
         k8s_container_config = deepcopy(k8s_container_config_template)
 
@@ -717,7 +719,7 @@ def _build_image_address(self, image_name: str) -> str:
 
     def list_job(self):
         # Get jobs details
-        command = f"kubectl get jobs -o=json"
+        command = "kubectl get jobs -o=json"
         return_str = SubProcess.run(command)
         job_details_list = json.loads(return_str)["items"]
         jobs_details = {}

maro/cli/maro.py

@@ -15,7 +15,7 @@
 from maro.utils.logger import CliLogger
 
 MARO_BANNER = """
- __  __    _    ____   ___
+ __  __    _    ____   ___
 |  \/  |  / \  |  _ \ / _ \
 | |\/| | / _ \ | |_) | | | |
 | |  | |/ ___ \|  _ <| |_| |
@@ -43,10 +43,12 @@ def main():
     # maro env
     parser_env = subparsers.add_parser(
         'env',
-        help=('Get all environment-related information, '
-              'such as the supported scenarios, topologies. '
-              'And it is also responsible to generate data to the specific environment, '
-              'which has external data dependency.'),
+        help=(
+            'Get all environment-related information, '
+            'such as the supported scenarios, topologies. '
+            'And it is also responsible to generate data to the specific environment, '
+            'which has external data dependency.'
+        ),
         parents=[global_parser]
     )
     parser_env.set_defaults(func=_help_func(parser=parser_env))
@@ -803,9 +805,11 @@ def load_parser_data(prev_parser: ArgumentParser, global_parser: ArgumentParser)
         type=int,
         default=None,
         required=False,
-        help=("Specified start timestamp (in UTC) for binary file, "
-              "then this timestamp will be considered as tick=0 for binary reader, "
-              "this can be used to adjust the reader pipeline."))
+        help="""
+        Specified start timestamp (in UTC) for binary file,
+        then this timestamp will be considered as tick=0 for binary reader,
+        this can be used to adjust the reader pipeline.
+        """)
 
     build_cmd_parser.set_defaults(func=convert)
 

maro/utils/logger.py

@@ -112,8 +112,10 @@ class Logger(object):
             ``DEBUG``.
     """
 
-    def __init__(self, tag: str, format_: LogFormat = LogFormat.full, dump_folder: str = cwd, dump_mode: str = 'w',
-                 extension_name: str = 'log', auto_timestamp: bool = True, stdout_level="INFO"):
+    def __init__(
+        self, tag: str, format_: LogFormat = LogFormat.full, dump_folder: str = cwd, dump_mode: str = 'w',
+        extension_name: str = 'log', auto_timestamp: bool = True, stdout_level="INFO"
+    ):
         self._file_format = FORMAT_NAME_TO_FILE_FORMAT[format_]
         self._stdout_format = FORMAT_NAME_TO_STDOUT_FORMAT[format_] \
             if format_ in FORMAT_NAME_TO_STDOUT_FORMAT else \
@@ -207,9 +209,11 @@ def critical(self, msg, *args):
 class InternalLogger(Logger):
     """An internal logger uses for recording the internal system's log."""
 
-    def __init__(self, component_name: str, tag: str = "maro_internal", format_: LogFormat = LogFormat.internal,
-                 dump_folder: str = None, dump_mode: str = 'a', extension_name: str = 'log',
-                 auto_timestamp: bool = False):
+    def __init__(
+        self, component_name: str, tag: str = "maro_internal", format_: LogFormat = LogFormat.internal,
+        dump_folder: str = None, dump_mode: str = 'a', extension_name: str = 'log',
+        auto_timestamp: bool = False
+    ):
         current_time = f"{datetime.now().strftime('%Y%m%d%H%M')}"
         self._dump_folder = dump_folder if dump_folder else \
             os.path.join(os.path.expanduser("~"), ".maro/log", current_time, str(os.getpid()))