mirror of
https://gitea.psi.ch/APOG/acsmnode.git
synced 2025-06-28 20:30:48 +02:00
123 lines
4.4 KiB
Python
123 lines
4.4 KiB
Python
import sys, os
|
|
|
|
try:
|
|
thisFilePath = os.path.abspath(__file__)
|
|
print(thisFilePath)
|
|
except NameError:
|
|
print("[Notice] The __file__ attribute is unavailable in this environment (e.g., Jupyter or IDLE).")
|
|
print("When using a terminal, make sure the working directory is set to the script's location to prevent path issues (for the DIMA submodule)")
|
|
#print("Otherwise, path to submodule DIMA may not be resolved properly.")
|
|
thisFilePath = os.getcwd() # Use current directory or specify a default
|
|
|
|
projectPath = os.path.normpath(os.path.join(thisFilePath, "..", "..",'..')) # Move up to project root
|
|
|
|
if projectPath not in sys.path:
|
|
sys.path.insert(0,projectPath)
|
|
|
|
import yaml
|
|
import re
|
|
import argparse
|
|
|
|
def flatten_yaml(data, sep=', '):
|
|
"""Flatten only nested values in the YAML dict by joining dict values into strings."""
|
|
flat = {}
|
|
for k, v in data.items():
|
|
if isinstance(v, dict):
|
|
# Join all values in the dict as a single string with separator
|
|
flat[k] = sep.join(str(val) for val in v.values())
|
|
else:
|
|
flat[k] = v
|
|
return flat
|
|
|
|
def load_template(path):
|
|
with open(path, 'r') as f:
|
|
return f.read()
|
|
|
|
def load_yaml(path):
|
|
with open(path, 'r') as f:
|
|
metadata = yaml.safe_load(f)
|
|
actris_metadata = {
|
|
'originator': metadata['originator'],
|
|
'submitter': metadata['submitter'],
|
|
'station_abbr': metadata['station_abbr'],
|
|
}
|
|
actris_metadata['originator_name'] = metadata['originator'].get('name', '')
|
|
actris_metadata['submitter_name'] = metadata['submitter'].get('name', '')
|
|
return flatten_yaml(actris_metadata)
|
|
|
|
def fill_placeholders(template, data):
|
|
def replacer(match):
|
|
key = match.group(1)
|
|
return str(data.get(key, f"${{{key}}}")) # Leave placeholder if not found
|
|
return re.sub(r'\$\{(\w+)\}', replacer, template)
|
|
|
|
def main(data_descriptor_path, dry_run = None):
|
|
|
|
|
|
if not os.path.exists(data_descriptor_path):
|
|
raise FileNotFoundError(f'Data descriptor not found at: {data_descriptor_path}')
|
|
|
|
metadata = load_yaml(data_descriptor_path)
|
|
print(metadata)
|
|
|
|
STATION_ABBR = metadata.get('station_abbr', None)
|
|
if not STATION_ABBR:
|
|
raise RuntimeError(
|
|
f'"station_abbr" is not defined in {data_descriptor_path}. '
|
|
'Make sure you specify it as station_abbr: "JFJ" or station_abbr: "PAY"'
|
|
)
|
|
|
|
# Define header paths
|
|
header_template_map = {
|
|
"JFJ": "pipelines/actris_header/JFJ_ACSM_017.actris_header",
|
|
"PAY": "pipelines/actris_header/PAY_ACSM_092.actris_header",
|
|
}
|
|
|
|
header_uptate_map = {
|
|
"JFJ": "third_party/acsmProcessingSoftware/src/cfg/actris_header/JFJ_ACSM_017.actris_header",
|
|
"PAY": "third_party/acsmProcessingSoftware/src/cfg/actris_header/PAY_ACSM_092.actris_header",
|
|
}
|
|
|
|
if STATION_ABBR not in header_template_map:
|
|
raise RuntimeError(f'Attribute station_abbr "{STATION_ABBR}" is not supported. Choose from: {list(header_template_map)}')
|
|
|
|
#projectPath = resolve_project_path()
|
|
header_path = os.path.join(projectPath, header_template_map[STATION_ABBR])
|
|
|
|
if not os.path.exists(header_path):
|
|
raise FileNotFoundError(f"Header template not found at: {header_path}")
|
|
|
|
print(f"Using template: {header_path}")
|
|
template_str = load_template(header_path)
|
|
filled = fill_placeholders(template_str, metadata)
|
|
|
|
if dry_run:
|
|
out_path = "output.test.ext"
|
|
print("[DRY RUN] Target header was not overwritten.")
|
|
else:
|
|
out_path = os.path.join(projectPath, header_uptate_map[STATION_ABBR])
|
|
print("[LIVE RUN] Target header will be updated.")
|
|
print(f"Writing to: {out_path}")
|
|
|
|
|
|
|
|
|
|
with open(out_path, 'w') as f:
|
|
f.write(filled)
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser(description="Fill ACTRIS header from metadata descriptor.")
|
|
parser.add_argument(
|
|
"data_descriptor_path",
|
|
help="Path to metadata YAML (data descriptor)"
|
|
)
|
|
parser.add_argument(
|
|
"--dry-run", "-d",
|
|
action="store_true",
|
|
help="Do not overwrite original template. Write output to 'output.test.ext'."
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
data_descriptor_path = args.data_descriptor_path
|
|
dry_run = args.dry_run
|
|
main(data_descriptor_path,dry_run ) |