<jupyter_start><jupyter_text>Macro-Pipeline Workflow  Set Run-Specific InputFill in the username/password for the SURF dCache. LAZ files updated since the last workflow run will be re-run through the full pipeline.<jupyter_code>webdav_login = input('WebDAV username: ')
webdav_password = getpass.getpass('WebDAV password: ')
last_run = datetime.datetime.strptime(input('Date last run (YYYY-MM-DD): '), '%Y-%m-%d')<jupyter_output><empty_output><jupyter_text>Check Connection to Remote Storage<jupyter_code>remote_path_root = pathlib.Path('/pnfs/grid.sara.nl/data/projects.nl/eecolidar/01_Escience/')
wd_opts = {
    'webdav_hostname': 'https://webdav.grid.surfsara.nl:2880',
    'webdav_login': webdav_login,
    'webdav_password': <PASSWORD>
}
assert get_wdclient(wd_opts).check(remote_path_root.as_posix())<jupyter_output><empty_output><jupyter_text>Setup ClusterSetup Dask cluster used for all the macro-pipeline calculations.<jupyter_code>local_tmp = pathlib.Path('/tmp')
cluster = LocalCluster(processes=True, 
                       n_workers=2, 
                       threads_per_worker=1, 
                       local_directory=local_tmp/'dask-worker-space')
# nprocs_per_node = 2
# cluster = SSHCluster(hosts=['172.17.0.2', 
#                             '172.17.0.2', 
#                             '172.17.0.3'], 
#                      connect_options={'known_hosts': None, 
#                                       'username': 'ubuntu', 
#                                       'client_keys': '/home/ubuntu/.ssh/id_rsa'},
#                      worker_options={'nthreads': 1, 
#                                      'nprocs': nprocs_per_node,
#                                      'local_directory': local_tmp/'dask-worker-space'}, 
#                      scheduler_options={'dashboard_address': '8787'})
cluster<jupyter_output><empty_output><jupyter_text>RetilingThe raw point-cloud files are downloaded and retiled to a regular grid.<jupyter_code># dCache path to raw LAZ files 
remote_path_ahn = remote_path_root / 'test_pipeline/test_full/raw'

# dCache path where to copy retiled PLY files
remote_path_retiled = remote_path_ahn.parent / 'retiled'

# details of the retiling schema
grid = {
    'min_x': -113107.81,
    'max_x': 398892.19,
    'min_y': 214783.87,
    'max_y': 726783.87,
    'n_tiles_side': 512
}

# determine which LAZ files have been updated since the last run 
laz_files = [f for f in list_remote(get_wdclient(wd_opts), remote_path_ahn.as_posix())
             if f.lower().endswith('.laz') and last_modified(wd_opts, remote_path_ahn/f) > last_run]
print('Retrieve and retile: {} LAZ files'.format(len(laz_files)))
# setup input dictionary to configure the retiling pipeline
retiling_input = {
    'setup_local_fs': {'tmp_folder': local_tmp.as_posix()},
    'pullremote': remote_path_ahn.as_posix(),
    'set_grid': grid,
    'split_and_redistribute': {},
    'validate': {},
    'pushremote': remote_path_retiled.as_posix(),
    'cleanlocalfs': {}
}

# write input dictionary to JSON file
with open('retiling.json', 'w') as f:
    json.dump(retiling_input, f)
macro = MacroPipeline()

# add pipeline list to macro-pipeline object and set the corresponding labels
macro.tasks = [Retiler(file).config(retiling_input).setup_webdav_client(wd_opts) for file in laz_files]
macro.set_labels([os.path.splitext(file)[0] for file in laz_files])

macro.setup_cluster(cluster=cluster)

# run! 
macro.run()

# save outcome results and check that no error occurred before continuing
macro.print_outcome(to_file='retiling.out')
assert not macro.get_failed_pipelines()<jupyter_output><empty_output><jupyter_text>Feature ExtractionFeatures computed for the retiled point-cloud data are assigned to a regular 'target' grid.<jupyter_code># target mesh size & list of features
tile_mesh_size = 10.
features = ['perc_95_normalized_height', 'pulse_penetration_ratio', 'entropy_normalized_height', 'point_density']

# dCache path where to copy the feature-enriched target data
remote_path_targets = remote_path_ahn.parent / 'targets'

# determine which tiles have been updated since last run, and extract tile index numbers
tiles = [t.strip('/') for t in list_remote(get_wdclient(wd_opts), remote_path_retiled.as_posix())
         if fnmatch.fnmatch(t, 'tile_*_*/') and last_modified(wd_opts, remote_path_retiled/t) > last_run]
tile_indices = [[int(el) for el in tile.split('_')[1:]] for tile in tiles]
print('Retrieve and process: {} tiles'.format(len(tile_indices)))
# setup input dictionary to configure the feature extraction pipeline
feature_extraction_input = {
    'setup_local_fs': {'tmp_folder': local_tmp.as_posix()},
    'pullremote': remote_path_retiled.as_posix(),
    'load': {'attributes': ['raw_classification']},
    'normalize': 1,
    'apply_filter': {
        'filter_type': 'select_equal', 
        'attribute': 'raw_classification',
        'value': [1, 6]#ground surface (2), water (9), buildings (6), artificial objects (26), vegetation (?), and unclassified (1)
    },
    'generate_targets': {
        'tile_mesh_size' : tile_mesh_size,
        'validate' : True,
        **grid
    },
    'extract_features': {
        'feature_names': features,
        'volume_type': 'cell',
        'volume_size': tile_mesh_size
    },
    'export_targets': {
        'attributes': features,
        'multi_band_files': False
    },
    'pushremote': remote_path_targets.as_posix(),
#     'cleanlocalfs': {}
}

# write input dictionary to JSON file
with open('feature_extraction.json', 'w') as f:
    json.dump(feature_extraction_input, f)
macro = MacroPipeline()

# add pipeline list to macro-pipeline object and set the corresponding labels
macro.tasks = [DataProcessing(t, tile_index=idx).config(feature_extraction_input).setup_webdav_client(wd_opts) 
               for t, idx in zip(tiles, tile_indices)]
macro.set_labels(tiles)

macro.setup_cluster(cluster=cluster)

# run!
macro.run()

# save outcome results and check that no error occurred before continuing
macro.print_outcome(to_file='feature_extraction.out')
assert not macro.get_failed_pipelines()<jupyter_output><empty_output><jupyter_text>GeoTIFF ExportExport the rasterized features from the target grid to GeoTIFF files.<jupyter_code># dCache path where to copy the GeoTIFF files
remote_path_geotiffs = remote_path_ahn.parent / 'geotiffs'
# setup input dictionary to configure the GeoTIFF export pipeline
geotiff_export_input = {
    'setup_local_fs': {'tmp_folder': local_tmp.as_posix()},
    'pullremote': remote_path_targets.as_posix(),
    'parse_point_cloud': {},
    'data_split': {'xSub': 1, 'ySub': 1},
    'create_subregion_geotiffs': {'output_handle': 'geotiff'},
    'pushremote': remote_path_geotiffs.as_posix(),
    'cleanlocalfs': {}   
}

# write input dictionary to JSON file
with open('geotiff_export.json', 'w') as f:
    json.dump(geotiff_export_input, f)
macro = MacroPipeline()

# add pipeline list to macro-pipeline object and set the corresponding labels
macro.tasks = [GeotiffWriter(input_dir=feature, bands=feature).config(geotiff_export_input).setup_webdav_client(wd_opts) 
               for feature in features]
macro.set_labels(features)

macro.setup_cluster(cluster=cluster)

# run!
macro.run()

# save outcome results and check that no error occurred before continuing
macro.print_outcome(to_file='geotiff_export.out')
assert not macro.get_failed_pipelines()<jupyter_output><empty_output><jupyter_text>Terminate cluster<jupyter_code>cluster.close()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11712191872085276, 'fraction_numerical': 0.011592271818787475, 'mean_word_length': 4.176551724137931, 'pattern_counts': {'":': 0, '<': 30, '<?xml version=': 0, '>': 32, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 3, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 5, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '7', 'dir': 'jupyter-structured-clean-dedup', 'id': '644140', 'license': 'Apache-2.0', 'n_tokens_mistral': 2519, 'n_tokens_neox': 2406, 'n_words': 539, 'path': 'workflow.ipynb', 'repo_name': 'eEcoLiDAR/lcMacroPipeline', 'size': '7510'}