Ana içeriğe geç

Core API

selection_service.core.earthquake_api.EarthquakeAPI

Kullanıcıya yönelik üst düzey Facade.

Doğrudan kullanım

api = EarthquakeAPI( provider_names=[ProviderName.AFAD, ProviderName.PEER], strategies=[TBDYSelectionStrategy(config)], ) result = api.run_sync(criteria, "TBDY_2018_Gaussian") api.download_waveforms(result.value.selected_df)

Servislerle doğrudan çalışmak isteyenler için

api.query → EarthquakeQueryService api.downloader → WaveformDownloadService api.registry → ProviderRegistry

Source code in src/selection_service/core/earthquake_api.py 
 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
class EarthquakeAPI:
    """Kullanıcıya yönelik üst düzey Facade.

    Doğrudan kullanım:
        api = EarthquakeAPI(
            provider_names=[ProviderName.AFAD, ProviderName.PEER],
            strategies=[TBDYSelectionStrategy(config)],
        )
        result = api.run_sync(criteria, "TBDY_2018_Gaussian")
        api.download_waveforms(result.value.selected_df)

    Servislerle doğrudan çalışmak isteyenler için:
        api.query      → EarthquakeQueryService
        api.downloader → WaveformDownloadService
        api.registry   → ProviderRegistry
    """

    def __init__(
        self,
        provider_names: List[ProviderName],
        strategies: List[ISelectionStrategy],
        use_cache: bool = True,
        **kwargs: Any,
    ) -> None:
        # 1. Provider'ları oluştur ve kaydet
        self.registry = ProviderRegistry.build(
            provider_names, use_cache=use_cache, **kwargs
        )

        # 2. Sorgulama servisi
        self.query = EarthquakeQueryService(
            registry=self.registry,
            strategies=strategies,
        )

        # 3. Download servisi
        self.downloader = WaveformDownloadService(registry=self.registry)

    # ── Sorgulama (EarthquakeQueryService'e delege) ────────────────

    async def run_async(
        self,
        criteria: SearchCriteria,
        strategy_name: str,
    ) -> Result[PipelineResult, PipelineError]:
        """Pipeline'ı asenkron çalıştır."""
        return await self.query.run_async(criteria, strategy_name)

    def search_fdsn_stations(self, **params: Any) -> Result:
        """Search station/channel metadata on the configured FDSN service."""
        provider = self.registry.get(ProviderName.FDSN.value)
        if provider is None or not hasattr(provider, "fetch_stations_sync"):
            return Result.fail(
                ProviderError(
                    ProviderName.FDSN.value,
                    LookupError("FDSN provider is not enabled."),
                    "FDSN station query is unavailable",
                )
            )
        return provider.fetch_stations_sync(**params)

    async def search_fdsn_stations_async(self, **params: Any) -> Result:
        """Asynchronously search FDSN station/channel metadata."""
        provider = self.registry.get(ProviderName.FDSN.value)
        if provider is None or not hasattr(provider, "fetch_stations_async"):
            return Result.fail(
                ProviderError(
                    ProviderName.FDSN.value,
                    LookupError("FDSN provider is not enabled."),
                    "FDSN station query is unavailable",
                )
            )
        return await provider.fetch_stations_async(**params)

    def search_fdsn_waveforms(self, **params: Any) -> Result:
        """Fetch decoded traces (``waveform.Trace`` list) from the FDSN dataselect service."""
        provider = self.registry.get(ProviderName.FDSN.value)
        if provider is None or not hasattr(provider, "fetch_waveforms_sync"):
            return Result.fail(
                ProviderError(
                    ProviderName.FDSN.value,
                    LookupError("FDSN provider is not enabled."),
                    "FDSN waveform query is unavailable",
                )
            )
        return provider.fetch_waveforms_sync(**params)

    async def search_fdsn_waveforms_async(self, **params: Any) -> Result:
        """Asynchronously fetch decoded traces from the FDSN dataselect service."""
        provider = self.registry.get(ProviderName.FDSN.value)
        if provider is None or not hasattr(provider, "fetch_waveforms_async"):
            return Result.fail(
                ProviderError(
                    ProviderName.FDSN.value,
                    LookupError("FDSN provider is not enabled."),
                    "FDSN waveform query is unavailable",
                )
            )
        return await provider.fetch_waveforms_async(**params)

    def run_sync(
        self,
        criteria: SearchCriteria,
        strategy_name: str,
    ) -> Result[PipelineResult, PipelineError]:
        """Pipeline'ı senkron çalıştır."""
        return self.query.run_sync(criteria, strategy_name)

    def re_selection(
        self,
        df: pd.DataFrame,
        strategy_name: str,
        new_criteria: SearchCriteria,
    ) -> Result[PipelineResult, PipelineError]:
        """Var olan veri üzerinde yeniden strateji uygula."""
        return self.query.re_selection(df, strategy_name, new_criteria)

    # ── Download (WaveformDownloadService'e delege) ────────────────

    def download_waveforms(
        self,
        result_df: pd.DataFrame,
        **kwargs: Any,
    ) -> Result[bool, ProviderError]:
        """Toplu dalga formu indirme."""
        return self.downloader.download_batch(result_df, **kwargs)

    def download_single_waveform(
        self,
        filename: str,
        event_id: str,
        station_code: str,
        **kwargs: Any,
    ) -> Result[bool, ProviderError]:
        """Tekil dalga formu indirme."""
        return self.downloader.download_single(filename, event_id, station_code, **kwargs)

run_async(criteria, strategy_name) async

Pipeline'ı asenkron çalıştır.

Source code in src/selection_service/core/earthquake_api.py 
81
82
83
84
85
86
87
async def run_async(
    self,
    criteria: SearchCriteria,
    strategy_name: str,
) -> Result[PipelineResult, PipelineError]:
    """Pipeline'ı asenkron çalıştır."""
    return await self.query.run_async(criteria, strategy_name)

search_fdsn_stations(**params)

Search station/channel metadata on the configured FDSN service.

Source code in src/selection_service/core/earthquake_api.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def search_fdsn_stations(self, **params: Any) -> Result:
    """Search station/channel metadata on the configured FDSN service."""
    provider = self.registry.get(ProviderName.FDSN.value)
    if provider is None or not hasattr(provider, "fetch_stations_sync"):
        return Result.fail(
            ProviderError(
                ProviderName.FDSN.value,
                LookupError("FDSN provider is not enabled."),
                "FDSN station query is unavailable",
            )
        )
    return provider.fetch_stations_sync(**params)

search_fdsn_stations_async(**params) async

Asynchronously search FDSN station/channel metadata.

Source code in src/selection_service/core/earthquake_api.py 
102
103
104
105
106
107
108
109
110
111
112
113
async def search_fdsn_stations_async(self, **params: Any) -> Result:
    """Asynchronously search FDSN station/channel metadata."""
    provider = self.registry.get(ProviderName.FDSN.value)
    if provider is None or not hasattr(provider, "fetch_stations_async"):
        return Result.fail(
            ProviderError(
                ProviderName.FDSN.value,
                LookupError("FDSN provider is not enabled."),
                "FDSN station query is unavailable",
            )
        )
    return await provider.fetch_stations_async(**params)

search_fdsn_waveforms(**params)

Fetch decoded traces (waveform.Trace list) from the FDSN dataselect service.

Source code in src/selection_service/core/earthquake_api.py 
115
116
117
118
119
120
121
122
123
124
125
126
def search_fdsn_waveforms(self, **params: Any) -> Result:
    """Fetch decoded traces (``waveform.Trace`` list) from the FDSN dataselect service."""
    provider = self.registry.get(ProviderName.FDSN.value)
    if provider is None or not hasattr(provider, "fetch_waveforms_sync"):
        return Result.fail(
            ProviderError(
                ProviderName.FDSN.value,
                LookupError("FDSN provider is not enabled."),
                "FDSN waveform query is unavailable",
            )
        )
    return provider.fetch_waveforms_sync(**params)

search_fdsn_waveforms_async(**params) async

Asynchronously fetch decoded traces from the FDSN dataselect service.

Source code in src/selection_service/core/earthquake_api.py 
128
129
130
131
132
133
134
135
136
137
138
139
async def search_fdsn_waveforms_async(self, **params: Any) -> Result:
    """Asynchronously fetch decoded traces from the FDSN dataselect service."""
    provider = self.registry.get(ProviderName.FDSN.value)
    if provider is None or not hasattr(provider, "fetch_waveforms_async"):
        return Result.fail(
            ProviderError(
                ProviderName.FDSN.value,
                LookupError("FDSN provider is not enabled."),
                "FDSN waveform query is unavailable",
            )
        )
    return await provider.fetch_waveforms_async(**params)

run_sync(criteria, strategy_name)

Pipeline'ı senkron çalıştır.

Source code in src/selection_service/core/earthquake_api.py 
141
142
143
144
145
146
147
def run_sync(
    self,
    criteria: SearchCriteria,
    strategy_name: str,
) -> Result[PipelineResult, PipelineError]:
    """Pipeline'ı senkron çalıştır."""
    return self.query.run_sync(criteria, strategy_name)

re_selection(df, strategy_name, new_criteria)

Var olan veri üzerinde yeniden strateji uygula.

Source code in src/selection_service/core/earthquake_api.py 
149
150
151
152
153
154
155
156
def re_selection(
    self,
    df: pd.DataFrame,
    strategy_name: str,
    new_criteria: SearchCriteria,
) -> Result[PipelineResult, PipelineError]:
    """Var olan veri üzerinde yeniden strateji uygula."""
    return self.query.re_selection(df, strategy_name, new_criteria)

download_waveforms(result_df, **kwargs)

Toplu dalga formu indirme.

Source code in src/selection_service/core/earthquake_api.py 
160
161
162
163
164
165
166
def download_waveforms(
    self,
    result_df: pd.DataFrame,
    **kwargs: Any,
) -> Result[bool, ProviderError]:
    """Toplu dalga formu indirme."""
    return self.downloader.download_batch(result_df, **kwargs)

download_single_waveform(filename, event_id, station_code, **kwargs)

Tekil dalga formu indirme.

Source code in src/selection_service/core/earthquake_api.py 
168
169
170
171
172
173
174
175
176
def download_single_waveform(
    self,
    filename: str,
    event_id: str,
    station_code: str,
    **kwargs: Any,
) -> Result[bool, ProviderError]:
    """Tekil dalga formu indirme."""
    return self.downloader.download_single(filename, event_id, station_code, **kwargs)

selection_service.core.pipeline.PipelineResult dataclass

Source code in src/selection_service/core/pipeline.py 
34
35
36
37
38
39
40
41
@dataclass
class PipelineResult:
    selected_df    : pd.DataFrame
    scored_df      : pd.DataFrame
    report         : Dict[str, Any]
    execution_time : float
    failed_providers: List[str] = field(default_factory=list)
    logs           : List[str]  = field(default_factory=list)

selection_service.core.pipeline.PipelineContext dataclass

Source code in src/selection_service/core/pipeline.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
@dataclass
class PipelineContext:
    providers       : List[IDataFetcher]    # ← IDataProvider → IDataFetcher
    strategy        : ISelectionStrategy
    search_criteria : SearchCriteria
    data            : Optional[List[pd.DataFrame]] = None
    combined_df     : Optional[pd.DataFrame]       = None
    strategy_input_df: Optional[pd.DataFrame]      = None
    deduplication   : Dict[str, int]               = field(default_factory=dict)
    selected_df     : Optional[pd.DataFrame]       = None
    scored_df       : Optional[pd.DataFrame]       = None
    failed_providers: List[str]                    = field(default_factory=list)
    logs            : List[str]                    = field(default_factory=list)
    start_time      : float                        = field(default_factory=time.time)

selection_service.core.pipeline.PipelineReporter

Rapor oluşturma işlemlerinden sorumlu sınıf.

Source code in src/selection_service/core/pipeline.py 
 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
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
class PipelineReporter:
    """Rapor oluşturma işlemlerinden sorumlu sınıf."""

    def generate_report(self, context: PipelineContext) -> Dict[str, Any]:
        if context.selected_df is None or context.selected_df.empty:
            return {
                "status": "warning",
                "message": "No records selected",
                "compliance": self._compliance(context),
            }

        return {
            "status": "success",
            "search_criteria": (
                context.search_criteria.model_dump()
                if hasattr(context.search_criteria, "model_dump")
                else context.search_criteria
            ),
            "selected_count": len(context.selected_df),
            "total_considered": (
                len(context.scored_df) if context.scored_df is not None else 0
            ),
            "strategy": context.strategy.get_name(),
            "providers": [p.get_name() for p in context.providers],
            "records": context.selected_df.to_dict("records"),
            "statistics": self._calculate_statistics(context.selected_df),
            "selection_summary": self._selection_summary(context.scored_df),
            "score_breakdown": self._selected_score_breakdown(context.selected_df),
            "error_metrics": self._selected_error_metrics(context.selected_df),
            "compliance": self._compliance(context),
            "deduplication": dict(context.deduplication),
        }

    def _compliance(self, context: PipelineContext) -> Dict[str, Any]:
        """Seçim sayısı ve olay başına kayıt sınırı kontrolü.

        TBDY 2018: her yön için 11 kayıt (11*2 = 22) ve aynı depremden en
        fazla 3 kayıt/kayıt takımı. Sınırlar strateji yapılandırmasından
        (``num_records``, ``max_per_event``) okunur.
        """
        config = getattr(context.strategy, "config", None)
        selected = context.selected_df
        count = 0 if selected is None else len(selected)
        result: Dict[str, Any] = {"selected_count": count, "warnings": []}
        required = getattr(config, "num_records", None)
        limit = getattr(config, "max_per_event", None)
        if not isinstance(required, int) or not isinstance(limit, int):
            return result

        per_event: Dict[str, int] = {}
        if selected is not None and not selected.empty and "EVENT" in selected.columns:
            per_event = selected["EVENT"].value_counts().to_dict()
        max_per_event = max(per_event.values(), default=0)

        result.update(
            required_count=required,
            shortfall=max(required - count, 0),
            max_per_event_limit=limit,
            max_selected_per_event=max_per_event,
            max_per_event_ok=max_per_event <= limit,
        )
        if count < required:
            result["warnings"].append(
                f"Yetersiz kayıt: {count}/{required} seçilebildi. Arama "
                "kriterlerini (büyüklük, uzaklık, Vs30, mekanizma) genişletin."
            )
        if max_per_event > limit:
            result["warnings"].append(
                f"Aynı depremden {max_per_event} kayıt seçildi (sınır {limit})."
            )
        result["compliant"] = not result["warnings"]
        return result

    def _calculate_statistics(self, df: pd.DataFrame) -> Dict[str, Any]:
        stats: Dict[str, Any] = {
            "magnitude_range": (df["MAGNITUDE"].min(), df["MAGNITUDE"].max()),
            "score_range": (df["SCORE"].min(), df["SCORE"].max()),
        }
        if "RJB(km)" in df.columns:
            stats["distance_range"] = (df["RJB(km)"].min(), df["RJB(km)"].max())
        return stats

    def _selection_summary(self, df: pd.DataFrame | None) -> Dict[str, Any]:
        """Summarize selected/rejected records and rejection reasons."""
        if df is None or df.empty or "SELECTION_STATUS" not in df.columns:
            return {"status_counts": {}, "rejection_reasons": {}}

        status_counts = df["SELECTION_STATUS"].value_counts().to_dict()
        if "SELECTION_REASON" not in df.columns:
            return {"status_counts": status_counts, "rejection_reasons": {}}

        rejected = df[df["SELECTION_STATUS"] == "rejected"]
        return {
            "status_counts": status_counts,
            "rejection_reasons": rejected["SELECTION_REASON"].value_counts().to_dict(),
        }

    def _selected_score_breakdown(self, df: pd.DataFrame | None) -> List[Dict[str, Any]]:
        """Expose compact criterion-level score details for selected records."""
        if df is None or df.empty or "SCORE_BREAKDOWN" not in df.columns:
            return []

        id_columns = [c for c in ("PROVIDER", "RSN", "EVENT", "STATION", "SCORE") if c in df.columns]
        rows: List[Dict[str, Any]] = []
        for _, record in df.iterrows():
            item = {col: record.get(col) for col in id_columns}
            item["criteria"] = record.get("SCORE_BREAKDOWN", [])
            item["selection_reason"] = record.get("SELECTION_REASON", "")
            rows.append(item)
        return rows

    def _selected_error_metrics(self, df: pd.DataFrame | None) -> List[Dict[str, Any]]:
        """Expose constraint strategy error metrics for selected records."""
        if df is None or df.empty or "ERROR_METRICS" not in df.columns:
            return []

        id_columns = [
            c for c in ("PROVIDER", "RSN", "EVENT", "STATION", "ERROR_TOTAL", "SCORE")
            if c in df.columns
        ]
        rows: List[Dict[str, Any]] = []
        for _, record in df.iterrows():
            item = {col: record.get(col) for col in id_columns}
            item["metrics"] = record.get("ERROR_METRICS", [])
            item["hard_filters"] = record.get("HARD_FILTERS", [])
            item["selection_reason"] = record.get("SELECTION_REASON", "")
            rows.append(item)
        return rows

selection_service.core.pipeline.EarthquakePipeline

Railway Oriented Pipeline Engine.

Her adım Result[T, E] döndürür; bir adım başarısız olursa sonraki adımlar çalıştırılmaz (short-circuit).

Source code in src/selection_service/core/pipeline.py 
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
class EarthquakePipeline:
    """Railway Oriented Pipeline Engine.

    Her adım Result[T, E] döndürür; bir adım başarısız olursa
    sonraki adımlar çalıştırılmaz (short-circuit).
    """

    def __init__(self) -> None:
        self.reporter = PipelineReporter()

    # ── Async execution ────────────────────────────────────────────

    async def execute_async(
        self, context: PipelineContext
    ) -> Result[PipelineResult, PipelineError]:
        context.start_time = time.time()
        pipeline_flow = self._compose_async(
            self._validate_inputs,
            self._fetch_data_async,
            self._combine_data,
            self._apply_strategy,
            self._finalize_result,
        )
        return await pipeline_flow(context)

    # ── Sync execution ─────────────────────────────────────────────

    def execute_sync(
        self, context: PipelineContext
    ) -> Result[PipelineResult, PipelineError]:
        context.start_time = time.time()
        pipeline_flow = self._compose_sync(
            self._validate_inputs,
            self._fetch_data_sync,
            self._combine_data,
            self._apply_strategy,
            self._finalize_result,
        )
        return pipeline_flow(context)

    # ── Pipeline adımları ──────────────────────────────────────────

    @result_decorator
    def _validate_inputs(self, context: PipelineContext) -> PipelineContext:
        """Adım 1: Girdi kontrolü."""
        if not context.providers:
            raise PipelineError("Validation", None, "No providers specified")
        return context

    @async_result_decorator
    async def _fetch_data_async(self, context: PipelineContext) -> PipelineContext:
        """Adım 2 (Async): Paralel veri çekme."""

        async def _fetch_safe(
            provider: IDataFetcher,
        ) -> Result[pd.DataFrame, ProviderError]:
            try:
                crit = provider.map_criteria(context.search_criteria)
                return await provider.fetch_data_async(crit)
            except Exception as e:
                return Result.fail(ProviderError(provider.get_name(), e))

        tasks = [_fetch_safe(p) for p in context.providers]
        results = await asyncio.gather(*tasks)

        valid_data = []
        for i, res in enumerate(results):
            p_name = context.providers[i].get_name()
            if res.success and res.value is not None and not res.value.empty:
                valid_data.append(res.value)
                context.logs.append(f"[OK] {p_name} fetched {len(res.value)} records")
            else:
                context.failed_providers.append(p_name)
                reason = str(res.error) if not res.success else "Boş veri döndü"
                context.logs.append(f"[FAIL] {p_name}: {reason}")
                logger.warning("[%s] Veri alınamadı: %s", p_name, reason)

        if not valid_data:
            failed = ", ".join(context.failed_providers) or "bilinmiyor"
            raise NoDataError(
                f"Hiçbir sağlayıcıdan veri alınamadı. "
                f"Başarısız: [{failed}]. "
                "Detaylar için yukarıdaki uyarı mesajlarını inceleyin."
            )

        context.data = valid_data
        return context

    @result_decorator
    def _fetch_data_sync(self, context: PipelineContext) -> PipelineContext:
        """Adım 2 (Sync): Sıralı veri çekme."""
        valid_data = []
        for provider in context.providers:
            p_name = provider.get_name()
            try:
                crit = provider.map_criteria(context.search_criteria)
                res = provider.fetch_data_sync(crit)

                if res.success and res.value is not None and not res.value.empty:
                    valid_data.append(res.value)
                    context.logs.append(f"[OK] {p_name} fetched {len(res.value)} records")
                else:
                    context.failed_providers.append(p_name)
                    reason = str(res.error) if res.error else "Boş veri döndü"
                    context.logs.append(f"[FAIL] {p_name}: {reason}")
                    logger.warning("[%s] Veri alınamadı: %s", p_name, reason)

            except Exception as e:
                context.failed_providers.append(p_name)
                context.logs.append(f"[ERROR] {p_name}: {e}")
                logger.error("[%s] Beklenmedik hata: %s", p_name, e, exc_info=True)

        if not valid_data:
            failed = ", ".join(context.failed_providers) or "bilinmiyor"
            raise NoDataError(
                f"Hiçbir sağlayıcıdan veri alınamadı. "
                f"Başarısız: [{failed}]. "
                "Detaylar için yukarıdaki uyarı mesajlarını inceleyin."
            )

        context.data = valid_data
        return context

    @result_decorator
    def _combine_data(self, context: PipelineContext) -> PipelineContext:
        """Adım 3: Veri birleştirme ve temizleme.

        Önemli: dropna(axis=1, how="all") uygulanmaz.
        Bir provider'ın tüm satırları için None olan sütunlar (ör. PEER'dan
        gelen ENDPOINTSOURCE) diğer provider'ın verilerinde dolu olabilir.
        Sütun silme işlemi bu değerleri de yok edeceğinden uygulanmıyor.
        Bunun yerine eksik string kolonlar "", sayısal kolonlar 0 ile doldurulur;
        ENDPOINTSOURCE gibi URL/None kolonları None olarak bırakılır.
        """
        valid_dfs = [df for df in context.data if not df.empty]

        if not valid_dfs:
            raise NoDataError("No valid columns in retrieved data")

        combined = pd.concat(valid_dfs, ignore_index=True)

        # Seçim algoritması eksik veriyi gerçek değer gibi puanlamasın diye
        # 0 doldurmadan önceki hali ayrıca saklanır; çıktılar yine 0 ile
        # doldurulur (aşağı akış hesaplamaları NaN ile hata veriyor).
        strategy_input = self._mark_missing(combined)

        # EVENT_GROUP her çalıştırmada eklenir (çıktı şeması provider sayısından
        # bağımsız olsun). Birden fazla provider varsa aynı deprem/kayıt
        # tekrarları da tespit edilir.
        context.deduplication = {
            "merged_event_groups": 0,
            "duplicate_records_removed": 0,
        }
        dedup = self._dedup_config(context)
        if {"PROVIDER", "EVENT"} <= set(combined.columns):
            if dedup.enabled:
                groups = assign_event_groups(
                    combined,
                    max_event_distance_km=dedup.max_event_distance_km,
                    max_mag_diff=dedup.max_mag_diff,
                )
            else:
                groups = combined["PROVIDER"].astype(str) + "|" + combined["EVENT"].astype(str)
            combined[EVENT_GROUP_COLUMN] = groups
            strategy_input[EVENT_GROUP_COLUMN] = groups
            if dedup.enabled and combined["PROVIDER"].nunique() > 1:
                duplicates = find_duplicate_records(
                    strategy_input,
                    max_station_distance_km=dedup.max_station_distance_km,
                    prefer_provider=dedup.prefer_provider,
                )
                merged = int(groups[groups.str.fullmatch(r"G\d+")].nunique())
                context.deduplication = {
                    "merged_event_groups": merged,
                    "duplicate_records_removed": int(duplicates.sum()),
                }
                if duplicates.any() or merged:
                    context.logs.append(
                        f"Dedup: {merged} olay grubu birleştirildi, "
                        f"{int(duplicates.sum())} tekrar kayıt seçim dışı bırakıldı."
                    )
                strategy_input = strategy_input[~duplicates]
        context.strategy_input_df = strategy_input

        # Sayısal kolonları 0 ile doldur
        num_cols = combined.select_dtypes(include=["number"]).columns
        combined[num_cols] = combined[num_cols].fillna(0)

        # String kolonları "" ile doldur — URL/link kolonları (ENDPOINTSOURCE) hariç
        str_cols = combined.select_dtypes(include=["object"]).columns
        url_cols = {"ENDPOINTSOURCE"}  # None kalması gereken kolonlar
        fill_str_cols = [c for c in str_cols if c not in url_cols]
        combined[fill_str_cols] = combined[fill_str_cols].fillna("")

        context.combined_df = combined
        context.logs.append(f"Combined total: {len(combined)} records")
        return context

    @staticmethod
    def _dedup_config(context: PipelineContext) -> DedupConfig:
        """Stratejinin ``config.dedup`` ayarı; yoksa varsayılanlar."""
        config = getattr(context.strategy, "config", None)
        dedup = getattr(config, "dedup", None)
        return dedup if isinstance(dedup, DedupConfig) else DedupConfig()

    @staticmethod
    def _mark_missing(df: pd.DataFrame) -> pd.DataFrame:
        """0 değeri "bilinmiyor" anlamına gelen kolonları NaN'a çevir."""
        marked = df.copy()
        for col in ZERO_MEANS_MISSING:
            if col in marked.columns:
                values = pd.to_numeric(marked[col], errors="coerce")
                marked[col] = values.mask(values <= 0)
        return marked

    @staticmethod
    def _fill_numeric_nulls(df: pd.DataFrame) -> pd.DataFrame:
        """Seçimden sonra sayısal boşlukları 0 ile doldur (çıktı sözleşmesi)."""
        if df is None or df.empty:
            return df
        filled = df.copy()
        num_cols = filled.select_dtypes(include=["number"]).columns
        filled[num_cols] = filled[num_cols].fillna(0)
        return filled

    @result_decorator
    def _apply_strategy(self, context: PipelineContext) -> PipelineContext:
        """Adım 4: Seçim stratejisi uygula."""
        if context.combined_df.empty:
            raise NoDataError("Combined data is empty")

        strategy_input = (
            context.strategy_input_df
            if context.strategy_input_df is not None
            else context.combined_df
        )
        selected, scored = context.strategy.select_and_score(
            strategy_input, context.search_criteria
        )

        context.selected_df = self._fill_numeric_nulls(selected)
        context.scored_df = self._fill_numeric_nulls(scored)
        context.logs.append(
            f"Strategy '{context.strategy.get_name()}' applied. "
            f"Selected: {len(selected)}"
        )
        for warning in PipelineReporter()._compliance(context)["warnings"]:
            context.logs.append(f"[WARN] {warning}")
            logger.warning(warning)
        return context

    @result_decorator
    def _finalize_result(self, context: PipelineContext) -> PipelineResult:
        """Adım 5: Sonuç paketleme."""
        execution_time = time.time() - context.start_time
        report = self.reporter.generate_report(context)

        return PipelineResult(
            selected_df=context.selected_df,
            scored_df=context.scored_df,
            report=report,
            execution_time=execution_time,
            failed_providers=context.failed_providers,
            logs=context.logs,
        )

    # ── Railway composers ──────────────────────────────────────────

    def _compose_async(self, *funcs: Callable) -> Callable:
        async def composed(input_ctx: PipelineContext) -> Result:
            current = Result.ok(input_ctx)
            for func in funcs:
                if not current.success:
                    break
                if inspect.iscoroutinefunction(func):
                    current = await func(current.value)
                else:
                    current = func(current.value)
            return current
        return composed

    def _compose_sync(self, *funcs: Callable) -> Callable:
        def composed(input_ctx: PipelineContext) -> Result:
            current = Result.ok(input_ctx)
            for func in funcs:
                if not current.success:
                    break
                current = func(current.value)
            return current
        return composed

selection_service.core.config

convert_mechanism_to_text(df, mechanism_col='MECHANISM')

Mekanizma sütununu sayısal değerlerden metin karşılıklarına dönüştür

Source code in src/selection_service/core/config.py 
160
161
162
163
164
def convert_mechanism_to_text(df: pd.DataFrame, mechanism_col: str = 'MECHANISM') -> pd.DataFrame:
    """Mekanizma sütununu sayısal değerlerden metin karşılıklarına dönüştür"""
    df = df.copy()
    df[mechanism_col] = df[mechanism_col].map(MECHANISM_MAP).fillna("Unknown")
    return df

convert_mechanism_to_numeric(df, mechanism_col='MECHANISM')

Mekanizma sütununu metin değerlerinden sayısal karşılıklarına dönüştür

Source code in src/selection_service/core/config.py 
166
167
168
169
170
def convert_mechanism_to_numeric(df: pd.DataFrame, mechanism_col: str = 'MECHANISM') -> pd.DataFrame:
    """Mekanizma sütununu metin değerlerinden sayısal karşılıklarına dönüştür"""
    df = df.copy()
    df[mechanism_col] = df[mechanism_col].map(REVERSE_MECHANISM_MAP).fillna(-999).astype(int)
    return df

get_mechanism_text(numeric_value)

Sayısal mekanizma değerini metin karşılığına çevir

Source code in src/selection_service/core/config.py 
172
173
174
def get_mechanism_text(numeric_value: int) -> str:
    """Sayısal mekanizma değerini metin karşılığına çevir"""
    return MECHANISM_MAP.get(numeric_value, 'Unknown')

get_mechanism_numeric(text_value)

Metin mekanizma değerini sayısal karşılığına çevir

Source code in src/selection_service/core/config.py 
176
177
178
def get_mechanism_numeric(text_value: str) -> int:
    """Metin mekanizma değerini sayısal karşılığına çevir"""
    return REVERSE_MECHANISM_MAP.get(text_value, -999)

selection_service.core.error_handle

PipelineError

Bases: Exception

Base pipeline error

Source code in src/selection_service/core/error_handle.py 
10
11
12
class PipelineError(Exception):
    """Base pipeline error"""
    pass

ValidationError

Bases: PipelineError

Validation failed

Source code in src/selection_service/core/error_handle.py 
15
16
17
class ValidationError(PipelineError):
    """Validation failed"""
    pass

NoDataError

Bases: PipelineError

No data found error

Source code in src/selection_service/core/error_handle.py 
20
21
22
class NoDataError(PipelineError):
    """No data found error"""
    pass

StrategyError

Bases: PipelineError

Strategy application error

Source code in src/selection_service/core/error_handle.py 
25
26
27
class StrategyError(PipelineError):
    """Strategy application error"""
    pass

ProviderError

Bases: PipelineError

Data provider error

Source code in src/selection_service/core/error_handle.py 
30
31
32
33
34
35
36
37
class ProviderError(PipelineError):
    """Data provider error"""
    def __init__(self, provider_name: str, original_error: Exception,
                 message: str = None):
        self.provider_name = provider_name
        self.original_error = original_error
        self.message = message or f"{provider_name} error: {original_error}"
        super().__init__(self.message)

NetworkError

Bases: ProviderError

Network related errors

Source code in src/selection_service/core/error_handle.py 
40
41
42
class NetworkError(ProviderError):
    """Network related errors"""
    pass

DataProcessingError

Bases: ProviderError

Data processing errors

Source code in src/selection_service/core/error_handle.py 
45
46
47
class DataProcessingError(ProviderError):
    """Data processing errors"""
    pass

AfadEmptyResultError

Bases: NoDataError

AFAD API'den hiç kayıt dönmedi.

Olası sebepler
  • Arama kriterleri çok kısıtlayıcı (tarih aralığı, büyüklük sınırı, bbox)
  • Belirtilen bölgede/tarih aralığında kayıtlı deprem yok
  • API geçici olarak hizmet dışı

Attributes:

Name Type Description
criteria_hint

Hangi kriterin dar olabileceğine dair ipucu (opsiyonel)
Source code in src/selection_service/core/error_handle.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
class AfadEmptyResultError(NoDataError):
    """AFAD API'den hiç kayıt dönmedi.

    Olası sebepler:
      - Arama kriterleri çok kısıtlayıcı (tarih aralığı, büyüklük sınırı, bbox)
      - Belirtilen bölgede/tarih aralığında kayıtlı deprem yok
      - API geçici olarak hizmet dışı

    Attributes:
        criteria_hint: Hangi kriterin dar olabileceğine dair ipucu (opsiyonel)
    """

    def __init__(self, message: str, criteria_hint: str | None = None) -> None:
        self.criteria_hint = criteria_hint
        full_msg = message
        if criteria_hint:
            full_msg += f" [Kriter ipucu: {criteria_hint}]"
        super().__init__(full_msg)

AfadMappingError

Bases: DataProcessingError

AFAD verisi geldi ancak kolon eşlemesi sonrası kayıt kalmadı.

Olası sebepler
  • API yanıt şeması değişmiş (kolon adları farklı)
  • Mapper yanlış yapılandırılmış
  • Tüm satırlar kritik bir kolonda None/NaN içeriyordu

Attributes:

Name Type Description
raw_count

Mapper öncesi kayıt sayısı
Source code in src/selection_service/core/error_handle.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
class AfadMappingError(DataProcessingError):
    """AFAD verisi geldi ancak kolon eşlemesi sonrası kayıt kalmadı.

    Olası sebepler:
      - API yanıt şeması değişmiş (kolon adları farklı)
      - Mapper yanlış yapılandırılmış
      - Tüm satırlar kritik bir kolonda None/NaN içeriyordu

    Attributes:
        raw_count: Mapper öncesi kayıt sayısı
    """

    def __init__(self, raw_count: int) -> None:
        self.raw_count = raw_count
        super().__init__(
            provider_name="AFAD",
            original_error=None,
            message=(
                f"AFAD: {raw_count} ham kayıt alındı ancak kolon eşlemesi "
                "sonrası hiç kayıt kalmadı. "
                "API yanıt şeması değişmiş veya mapper yanlış yapılandırılmış olabilir."
            ),
        )