32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176 | async def process_l0_last_steps(
mission: str,
session: str,
flow_params: Level0FlowParams,
input_products: list[FlowInputProduct],
verbose: bool,
):
"""
Final processing steps that are common to all missions.
Raises:
ValueError: _description_
"""
logger = get_run_logger()
logger.info(
"Starting L0 last steps: mission=%r, session=%r, verbose=%r, input_products_count=%d",
mission,
session,
verbose,
len(input_products),
)
logger.info("Raw flow_params: %r (type=%s)", flow_params, type(flow_params).__name__)
for index, input_product in enumerate(input_products):
logger.info(
"Input product [%d]: %r (type=%s)",
index,
input_product,
type(input_product).__name__,
)
# Resolve parameters
flow_params = flow_params or Level0FlowParams()
try:
p = await flow_params.resolve(mission)
except Exception:
logger.exception("Failed to resolve flow parameters for mission=%r", mission)
raise
logger.info("Resolved flow parameters: %r (type=%s)", p, type(p).__name__)
flow_env = FlowEnv(FlowEnvArgs(owner_id=p.owner_identifier))
logger.info("Created FlowEnv for owner_id=%r", p.owner_identifier)
with flow_env.start_span(__name__, f"sentinel{mission}-level0-processing"):
logger.info(
"Looking up catalog session: session=%r, collection=%r",
session,
p.session_collection,
)
try:
item_session: Item | None = await get_single_catalog_item(flow_env, session, [p.session_collection])
except Exception:
logger.exception(
"Catalog lookup failed: session=%r, collection=%r",
session,
p.session_collection,
)
raise
if not item_session:
logger.error("❌ Session %r was not found; DPR processing cannot be launched.", session)
return
logger.info(f"✅ The session {session} has been found in the catalog.")
logger.info("Catalog item id=%r, properties=%r", item_session.id, item_session.properties)
# Satellite identifier
satellite_value = f"sentinel-{mission}{session[2].lower()}"
# Session datetime date
published = item_session.properties.get("datetime")
if not isinstance(published, str):
raise ValueError("Missing or invalid 'datetime' property in item_session")
end_datetime = datetime.fromisoformat(published)
start_datetime = end_datetime
# Call DPR flow
dpr_env = FlowEnvArgs(owner_id=p.owner_identifier)
dpr_parameters = {
"input_products": input_products,
"external_variables": {
"start_datetime": start_datetime,
"end_datetime": end_datetime,
"satellite": satellite_value,
},
"dask_cluster_label": p.dask_cluster_label,
"processor_name": p.processor_name,
"processor_version": p.processor_version,
"pipeline": p.pipeline,
"unit": p.unit,
"priority": p.priority,
"processing_mode": p.processing_mode,
"workflow": p.workflow,
"generated_product_to_collection_identifier": p.generated_product_to_collection_identifier or [],
"auxiliary_product_to_collection_identifier": p.auxiliary_product_to_collection_identifier or [],
"logging_level": p.logging_level,
}
logger.info("About to call call_dpr_flow with env=%r (type=%s)", dpr_env, type(dpr_env).__name__)
for parameter_name, parameter_value in dpr_parameters.items():
logger.info(
"call_dpr_flow parameter %s=%r (type=%s)",
parameter_name,
parameter_value,
type(parameter_value).__name__,
)
try:
l0_result = await call_dpr_flow(
dpr_env,
input_products=input_products,
external_variables={
"start_datetime": start_datetime,
"end_datetime": end_datetime,
"satellite": satellite_value,
},
dask_cluster_label=p.dask_cluster_label,
processor_name=p.processor_name,
processor_version=p.processor_version,
pipeline=p.pipeline,
unit=p.unit,
priority=p.priority,
processing_mode=p.processing_mode,
workflow=p.workflow,
generated_product_to_collection_identifier=(p.generated_product_to_collection_identifier or []),
auxiliary_product_to_collection_identifier=(p.auxiliary_product_to_collection_identifier or []),
logging_level=p.logging_level,
)
except Exception:
logger.exception(
"call_dpr_flow failed for mission=%r, session=%r, processor=%r:%r",
mission,
session,
p.processor_name,
p.processor_version,
)
raise
logger.info("call_dpr_flow completed successfully for mission=%r, session=%r", mission, session)
logger.info(
"L0 products prepared for persisted flow result: count=%d, products=%r",
len(l0_result),
l0_result,
)
return l0_result
|