Ana içeriğe geç

Providers API

selection_service.providers.interfaces

providers/interfaces.py

IDataProvider tek monolitik interface'i, sorumluluk bazlı iki ayrı Protocol'e bölünmüştür:

IDataFetcher — veri çekme & kriterleri dönüştürme IWaveformDownloader — dalga formu indirme

Bu ayrım iki SOLID ihlalini düzeltir
  • ISP (Interface Segregation): PEER gibi download desteklemeyen provider'lar artık download metodlarını implement etmek zorunda değil.
  • LSP (Liskov Substitution): IDataFetcher bekleyen kod, download fırlatmayan PEER ile güvenle çalışır.

Not: eski IDataProvider v2.0.0'da kaldırıldı.

IDataFetcher

Bases: Protocol

Deprem kayıtlarını bir veri kaynağından çeken provider sözleşmesi.

Tüm provider'lar (AFAD, PEER, FDSN, …) bu interface'i uygular. Download desteği olan provider'lar ek olarak IWaveformDownloader'ı da uygular.

Protokol notları
  • map_criteria : SearchCriteria → provider'a özgü Dict dönüşümü.
  • fetch_* : Dönüş tipi her zaman Result[DataFrame, ProviderError]. Başarısız durumda Result.fail() ile sarmalanmış hata dönülür, exception fırlatılmaz.
Source code in src/selection_service/providers/interfaces.py 
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
@runtime_checkable
class IDataFetcher(Protocol):
    """
    Deprem kayıtlarını bir veri kaynağından çeken provider sözleşmesi.

    Tüm provider'lar (AFAD, PEER, FDSN, …) bu interface'i uygular.
    Download desteği olan provider'lar ek olarak IWaveformDownloader'ı da uygular.

    Protokol notları:
      - map_criteria  : SearchCriteria → provider'a özgü Dict dönüşümü.
      - fetch_*       : Dönüş tipi her zaman Result[DataFrame, ProviderError].
                        Başarısız durumda Result.fail() ile sarmalanmış hata dönülür,
                        exception fırlatılmaz.
    """

    def get_name(self) -> str:
        """Provider'ın benzersiz adı (ör. 'AFAD', 'PEER').

        Bu ad cache anahtarı ve log mesajlarında kullanılır;
        büyük/küçük harf tutarlılığına dikkat edin.
        """
        ...

    def map_criteria(self, criteria: SearchCriteria) -> Dict[str, Any]:
        """Ortak SearchCriteria modelini provider'a özgü parametrelere dönüştür.

        Args:
            criteria: Kullanıcının girdiği arama kriterleri.

        Returns:
            Provider API'sinin beklediği parametre dict'i.
        """
        ...

    async def fetch_data_async(
        self, criteria: Dict[str, Any]
    ) -> Result[pd.DataFrame, ProviderError]:
        """Veriyi asenkron olarak çek.

        Args:
            criteria: map_criteria() çıktısı (provider'a özgü dict).

        Returns:
            Result.ok(df)   — başarı; en az bir kayıt içeren DataFrame.
            Result.fail(e)  — hata; ProviderError ile sarmalanmış.
        """
        ...

    def fetch_data_sync(
        self, criteria: Dict[str, Any]
    ) -> Result[pd.DataFrame, ProviderError]:
        """Veriyi senkron olarak çek.

        Args:
            criteria: map_criteria() çıktısı (provider'a özgü dict).

        Returns:
            Result.ok(df)   — başarı; en az bir kayıt içeren DataFrame.
            Result.fail(e)  — hata; ProviderError ile sarmalanmış.
        """
        ...

get_name()

Provider'ın benzersiz adı (ör. 'AFAD', 'PEER').

Bu ad cache anahtarı ve log mesajlarında kullanılır; büyük/küçük harf tutarlılığına dikkat edin.

Source code in src/selection_service/providers/interfaces.py 
49
50
51
52
53
54
55
def get_name(self) -> str:
    """Provider'ın benzersiz adı (ör. 'AFAD', 'PEER').

    Bu ad cache anahtarı ve log mesajlarında kullanılır;
    büyük/küçük harf tutarlılığına dikkat edin.
    """
    ...

map_criteria(criteria)

Ortak SearchCriteria modelini provider'a özgü parametrelere dönüştür.

Parameters:

Name Type Description Default
criteria SearchCriteria

Kullanıcının girdiği arama kriterleri.

 required

Returns:

Type Description
Dict[str, Any]

Provider API'sinin beklediği parametre dict'i.
Source code in src/selection_service/providers/interfaces.py 
57
58
59
60
61
62
63
64
65
66
def map_criteria(self, criteria: SearchCriteria) -> Dict[str, Any]:
    """Ortak SearchCriteria modelini provider'a özgü parametrelere dönüştür.

    Args:
        criteria: Kullanıcının girdiği arama kriterleri.

    Returns:
        Provider API'sinin beklediği parametre dict'i.
    """
    ...

fetch_data_async(criteria) async

Veriyi asenkron olarak çek.

Parameters:

Name Type Description Default
criteria Dict[str, Any]

map_criteria() çıktısı (provider'a özgü dict).

 required

Returns:

Type Description
Result[DataFrame, ProviderError]

Result.ok(df) — başarı; en az bir kayıt içeren DataFrame.
Result[DataFrame, ProviderError]

Result.fail(e) — hata; ProviderError ile sarmalanmış.
Source code in src/selection_service/providers/interfaces.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
async def fetch_data_async(
    self, criteria: Dict[str, Any]
) -> Result[pd.DataFrame, ProviderError]:
    """Veriyi asenkron olarak çek.

    Args:
        criteria: map_criteria() çıktısı (provider'a özgü dict).

    Returns:
        Result.ok(df)   — başarı; en az bir kayıt içeren DataFrame.
        Result.fail(e)  — hata; ProviderError ile sarmalanmış.
    """
    ...

fetch_data_sync(criteria)

Veriyi senkron olarak çek.

Parameters:

Name Type Description Default
criteria Dict[str, Any]

map_criteria() çıktısı (provider'a özgü dict).

 required

Returns:

Type Description
Result[DataFrame, ProviderError]

Result.ok(df) — başarı; en az bir kayıt içeren DataFrame.
Result[DataFrame, ProviderError]

Result.fail(e) — hata; ProviderError ile sarmalanmış.
Source code in src/selection_service/providers/interfaces.py 
82
83
84
85
86
87
88
89
90
91
92
93
94
def fetch_data_sync(
    self, criteria: Dict[str, Any]
) -> Result[pd.DataFrame, ProviderError]:
    """Veriyi senkron olarak çek.

    Args:
        criteria: map_criteria() çıktısı (provider'a özgü dict).

    Returns:
        Result.ok(df)   — başarı; en az bir kayıt içeren DataFrame.
        Result.fail(e)  — hata; ProviderError ile sarmalanmış.
    """
    ...

IWaveformDownloader

Bases: Protocol

Dalga formu dosyalarını indiren provider sözleşmesi.

Yalnızca download desteği olan provider'lar (şu an: AFAD) bu interface'i uygular. PEER ve FDSN bu interface'i uygulamak zorunda değildir.

Tasarım kararı
  • Her iki metot da Result[bool, ProviderError] döndürür; başarıda Result.ok(True), hata durumunda Result.fail(e).
  • Metotlar @result_decorator ile sarmalanabilir (bkz. AfadProvider).
Source code in src/selection_service/providers/interfaces.py 
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
@runtime_checkable
class IWaveformDownloader(Protocol):
    """
    Dalga formu dosyalarını indiren provider sözleşmesi.

    Yalnızca download desteği olan provider'lar (şu an: AFAD) bu interface'i
    uygular. PEER ve FDSN bu interface'i uygulamak zorunda değildir.

    Tasarım kararı:
      - Her iki metot da Result[bool, ProviderError] döndürür;
        başarıda Result.ok(True), hata durumunda Result.fail(e).
      - Metotlar @result_decorator ile sarmalanabilir (bkz. AfadProvider).
    """

    def download_single_waveforms(
        self,
        filename: str,
        **kwargs: Any,
    ) -> Result[bool, ProviderError]:
        """Tek bir dalga formu dosyasını indir ve diske kaydet.

        Args:
            filename: İndirilecek dosyanın adı (ör. 'waveform_12345.mseed').
            **kwargs:
                event_id      (int,  optional): Klasör yapısı için deprem ID'si.
                station_code  (str,  optional): Dosya adı için istasyon kodu.
                file_type     (str,  optional): 'ap' (default).
                file_status   (str,  optional): 'Acc' (default).
                    Seçenekler: RawAcc | Acc | Vel | Disp |
                                ResSpecAcc | ResSpecVel | ResSpecDisp | FFT | Husid
                export_type   (str,  optional): 'mseed' (default).
                    Seçenekler: asc2 | mseed | asd
                user_name     (str,  optional): 'GuestUser' (default).

        Returns:
            Result.ok(True)  — dosya başarıyla indirildi ve çıkarıldı.
            Result.fail(e)   — ağ, dosya sistemi veya zip hatası.
        """
        ...

    def download_waveforms_batch(
        self,
        filenames: List[str],
        **kwargs: Any,
    ) -> Dict[str, Any]:
        """Birden fazla dalga formu dosyasını toplu indir.

        Dosyalar batch_size'a göre gruplara ayrılır; her grup tek HTTP
        isteğiyle indirilir, ZIP açılır ve diske kaydedilir.

        Args:
            filenames: İndirilecek dosya adları listesi.
            **kwargs:
                batch_size    (int,  optional): Batch başına max dosya sayısı. ≤10, default 10.
                event_id      (int | List[int], optional): Deprem ID veya ID listesi.
                file_type     (str,  optional): 'ap' (default).
                file_status   (str,  optional): 'Acc' (default).
                export_type   (str,  optional): 'mseed' (default).
                user_name     (str,  optional): 'GuestUser' (default).

        Returns:
            {
                'total': int,
                'downloaded': int,
                'batches': List[dict],
            }
        """
        ...

download_single_waveforms(filename, **kwargs)

Tek bir dalga formu dosyasını indir ve diske kaydet.

Parameters:

Name Type Description Default
filename str

İndirilecek dosyanın adı (ör. 'waveform_12345.mseed').

 required
**kwargs Any

event_id (int, optional): Klasör yapısı için deprem ID'si. station_code (str, optional): Dosya adı için istasyon kodu. file_type (str, optional): 'ap' (default). file_status (str, optional): 'Acc' (default). Seçenekler: RawAcc | Acc | Vel | Disp | ResSpecAcc | ResSpecVel | ResSpecDisp | FFT | Husid export_type (str, optional): 'mseed' (default). Seçenekler: asc2 | mseed | asd user_name (str, optional): 'GuestUser' (default).

 {}

Returns:

Type Description
Result[bool, ProviderError]

Result.ok(True) — dosya başarıyla indirildi ve çıkarıldı.
Result[bool, ProviderError]

Result.fail(e) — ağ, dosya sistemi veya zip hatası.
Source code in src/selection_service/providers/interfaces.py 
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
def download_single_waveforms(
    self,
    filename: str,
    **kwargs: Any,
) -> Result[bool, ProviderError]:
    """Tek bir dalga formu dosyasını indir ve diske kaydet.

    Args:
        filename: İndirilecek dosyanın adı (ör. 'waveform_12345.mseed').
        **kwargs:
            event_id      (int,  optional): Klasör yapısı için deprem ID'si.
            station_code  (str,  optional): Dosya adı için istasyon kodu.
            file_type     (str,  optional): 'ap' (default).
            file_status   (str,  optional): 'Acc' (default).
                Seçenekler: RawAcc | Acc | Vel | Disp |
                            ResSpecAcc | ResSpecVel | ResSpecDisp | FFT | Husid
            export_type   (str,  optional): 'mseed' (default).
                Seçenekler: asc2 | mseed | asd
            user_name     (str,  optional): 'GuestUser' (default).

    Returns:
        Result.ok(True)  — dosya başarıyla indirildi ve çıkarıldı.
        Result.fail(e)   — ağ, dosya sistemi veya zip hatası.
    """
    ...

download_waveforms_batch(filenames, **kwargs)

Birden fazla dalga formu dosyasını toplu indir.

Dosyalar batch_size'a göre gruplara ayrılır; her grup tek HTTP isteğiyle indirilir, ZIP açılır ve diske kaydedilir.

Parameters:

Name Type Description Default
filenames List[str]

İndirilecek dosya adları listesi.

 required
**kwargs Any

batch_size (int, optional): Batch başına max dosya sayısı. ≤10, default 10. event_id (int | List[int], optional): Deprem ID veya ID listesi. file_type (str, optional): 'ap' (default). file_status (str, optional): 'Acc' (default). export_type (str, optional): 'mseed' (default). user_name (str, optional): 'GuestUser' (default).

 {}

Returns:

Type Description
Dict[str, Any]

{ 'total': int, 'downloaded': int, 'batches': List[dict],
Dict[str, Any]

}
Source code in src/selection_service/providers/interfaces.py 
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
def download_waveforms_batch(
    self,
    filenames: List[str],
    **kwargs: Any,
) -> Dict[str, Any]:
    """Birden fazla dalga formu dosyasını toplu indir.

    Dosyalar batch_size'a göre gruplara ayrılır; her grup tek HTTP
    isteğiyle indirilir, ZIP açılır ve diske kaydedilir.

    Args:
        filenames: İndirilecek dosya adları listesi.
        **kwargs:
            batch_size    (int,  optional): Batch başına max dosya sayısı. ≤10, default 10.
            event_id      (int | List[int], optional): Deprem ID veya ID listesi.
            file_type     (str,  optional): 'ap' (default).
            file_status   (str,  optional): 'Acc' (default).
            export_type   (str,  optional): 'mseed' (default).
            user_name     (str,  optional): 'GuestUser' (default).

    Returns:
        {
            'total': int,
            'downloaded': int,
            'batches': List[dict],
        }
    """
    ...

supports_download(provider)

Provider'ın dalga formu indirmeyi destekleyip desteklemediğini kontrol et.

Kullanım örneği

if supports_download(provider): provider.download_waveforms_batch(filenames) else: logger.info(f"{provider.get_name()} download desteklemiyor, atlanıyor.")

Parameters:

Name Type Description Default
provider IDataFetcher

Kontrol edilecek IDataFetcher örneği.

 required

Returns:

Type Description
bool

True — provider aynı zamanda IWaveformDownloader protokolünü karşılıyor.
bool

False — provider yalnızca veri çekmeyi destekliyor.
Source code in src/selection_service/providers/interfaces.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
def supports_download(provider: IDataFetcher) -> bool:
    """Provider'ın dalga formu indirmeyi destekleyip desteklemediğini kontrol et.

    Kullanım örneği:
        if supports_download(provider):
            provider.download_waveforms_batch(filenames)
        else:
            logger.info(f"{provider.get_name()} download desteklemiyor, atlanıyor.")

    Args:
        provider: Kontrol edilecek IDataFetcher örneği.

    Returns:
        True  — provider aynı zamanda IWaveformDownloader protokolünü karşılıyor.
        False — provider yalnızca veri çekmeyi destekliyor.
    """
    return isinstance(provider, IWaveformDownloader)

selection_service.providers.providers_factory.ProviderFactory

Provider factory — somut provider örnekleri oluşturur.

Dönüş tipi IDataFetcher'dır; download desteği olan provider'ları ayırt etmek için supports_download(provider) kullanın.

Source code in src/selection_service/providers/providers_factory.py 
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
class ProviderFactory:
    """Provider factory — somut provider örnekleri oluşturur.

    Dönüş tipi IDataFetcher'dır; download desteği olan provider'ları
    ayırt etmek için supports_download(provider) kullanın.
    """

    # Singleton benzeri paylaşımlı cache — tüm provider'lar aynı cache'i kullanır.
    _cache_manager: CacheManager = CacheManager()

    @staticmethod
    def create_provider(
        provider_type: ProviderName,
        use_cache: bool = True,
        **kwargs: Any,
    ) -> IDataFetcher:
        """Provider tipine göre uygun sağlayıcı örneği oluştur.

        Args:
            provider_type: ProviderName enum değeri (AFAD, PEER, …).
            use_cache:     True ise sonuçlar CachedProviderProxy ile önbelleklenir.
            **kwargs:      Provider'a özgü ek parametreler.

        Returns:
            IDataFetcher örneği (gerekirse IWaveformDownloader da karşılar).

        Raises:
            ValueError: Bilinmeyen provider_type.
        """
        mapper = ColumnMapperFactory.create_mapper(provider_type, **kwargs)

        if provider_type == ProviderName.AFAD:
            provider: IDataFetcher = AFADDataProvider(column_mapper=mapper)
        elif provider_type == ProviderName.PEER:
            provider = PeerWest2Provider(column_mapper=mapper, **kwargs)
        elif provider_type == ProviderName.FDSN:
            provider = FDSNProvider(column_mapper=mapper, **kwargs)
        elif provider_type == ProviderName.ESM:
            provider = ESMProvider(column_mapper=mapper)
        else:
            raise ValueError(f"Unknown provider: {provider_type}")

        if use_cache:
            return CachedProviderProxy(provider, ProviderFactory._cache_manager)

        return provider

create_provider(provider_type, use_cache=True, **kwargs) staticmethod

Provider tipine göre uygun sağlayıcı örneği oluştur.

Parameters:

Name Type Description Default
provider_type ProviderName

ProviderName enum değeri (AFAD, PEER, …).

 required
use_cache bool

True ise sonuçlar CachedProviderProxy ile önbelleklenir.

 True
**kwargs Any

Provider'a özgü ek parametreler.

 {}

Returns:

Type Description
IDataFetcher

IDataFetcher örneği (gerekirse IWaveformDownloader da karşılar).

Raises:

Type Description
ValueError

Bilinmeyen provider_type.
Source code in src/selection_service/providers/providers_factory.py 
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
@staticmethod
def create_provider(
    provider_type: ProviderName,
    use_cache: bool = True,
    **kwargs: Any,
) -> IDataFetcher:
    """Provider tipine göre uygun sağlayıcı örneği oluştur.

    Args:
        provider_type: ProviderName enum değeri (AFAD, PEER, …).
        use_cache:     True ise sonuçlar CachedProviderProxy ile önbelleklenir.
        **kwargs:      Provider'a özgü ek parametreler.

    Returns:
        IDataFetcher örneği (gerekirse IWaveformDownloader da karşılar).

    Raises:
        ValueError: Bilinmeyen provider_type.
    """
    mapper = ColumnMapperFactory.create_mapper(provider_type, **kwargs)

    if provider_type == ProviderName.AFAD:
        provider: IDataFetcher = AFADDataProvider(column_mapper=mapper)
    elif provider_type == ProviderName.PEER:
        provider = PeerWest2Provider(column_mapper=mapper, **kwargs)
    elif provider_type == ProviderName.FDSN:
        provider = FDSNProvider(column_mapper=mapper, **kwargs)
    elif provider_type == ProviderName.ESM:
        provider = ESMProvider(column_mapper=mapper)
    else:
        raise ValueError(f"Unknown provider: {provider_type}")

    if use_cache:
        return CachedProviderProxy(provider, ProviderFactory._cache_manager)

    return provider

selection_service.providers.providers_factory.CachedProviderProxy

Bases: IDataFetcher

Şeffaf cache katmanı — hem async hem sync çağrıları önbellekler.

Decorator / Proxy örüntüsü: - fetch_data_async / fetch_data_sync için önce cache'e bakılır. - Cache miss ise asıl provider çağrılır ve sonuç cache'e yazılır. - Diğer tüm metot çağrıları getattr aracılığıyla asıl provider'a iletilir (download metodları dahil).

Source code in src/selection_service/providers/providers_factory.py 
 28
 29
 30
 31
 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
class CachedProviderProxy(IDataFetcher):
    """Şeffaf cache katmanı — hem async hem sync çağrıları önbellekler.

    Decorator / Proxy örüntüsü:
      - fetch_data_async / fetch_data_sync için önce cache'e bakılır.
      - Cache miss ise asıl provider çağrılır ve sonuç cache'e yazılır.
      - Diğer tüm metot çağrıları __getattr__ aracılığıyla
        asıl provider'a iletilir (download metodları dahil).
    """

    def __init__(self, provider: IDataFetcher, cache_manager: CacheManager) -> None:
        self._provider = provider
        self._cache = cache_manager

    # ──────────────────────────────────────────────────────────────
    # IDataFetcher sözleşmesi — zorunlu delegasyonlar
    # ──────────────────────────────────────────────────────────────

    def get_name(self) -> str:
        return self._provider.get_name()

    def map_criteria(self, criteria: Any) -> dict:
        return self._provider.map_criteria(criteria)

    # ──────────────────────────────────────────────────────────────
    # Async fetch — cache destekli
    # ──────────────────────────────────────────────────────────────

    async def fetch_data_async(self, criteria: Any) -> Result:
        """Cache'den oku; bulamazsan provider'ı çağır ve sonucu yaz."""
        cached = self._try_read_cache(criteria)
        if cached is not None:
            return Result.ok(cached)

        result = await self._provider.fetch_data_async(criteria)
        self._try_write_cache(criteria, result)
        return result

    # ──────────────────────────────────────────────────────────────
    # Sync fetch — cache destekli (Adım 4 düzeltmesi)
    # ──────────────────────────────────────────────────────────────

    def fetch_data_sync(self, criteria: Any) -> Result:
        """Senkron cache desteği — önceki sürümde eksikti.

        Mevcut (önceki) davranış: her sync çağrı cache'i bypass ediyordu.
        Düzeltilmiş davranış: async ile aynı cache mantığı uygulanır.
        """
        cached = self._try_read_cache(criteria)
        if cached is not None:
            return Result.ok(cached)

        result = self._provider.fetch_data_sync(criteria)
        self._try_write_cache(criteria, result)
        return result

    # ──────────────────────────────────────────────────────────────
    # Proxy — diğer metotları (download vb.) asıl provider'a ilet
    # ──────────────────────────────────────────────────────────────

    def __getattr__(self, name: str) -> Any:
        """fetch_ ve get_name dışındaki her çağrıyı asıl provider'a aktar.

        Bu sayede CachedProviderProxy'nin IWaveformDownloader metodlarını
        (download_single_waveforms, download_waveforms_batch) da şeffaf biçimde
        iletmesi sağlanır — proxy'yi her yeni metot için güncellemek gerekmez.
        """
        return getattr(self._provider, name)

    # ──────────────────────────────────────────────────────────────
    # Private yardımcılar
    # ──────────────────────────────────────────────────────────────

    def _try_read_cache(self, criteria: Any):
        """Cache'den DataFrame oku; hata veya miss durumunda None döndür."""
        try:
            cached_df = self._cache.get(self._provider.get_name(), criteria)
            if cached_df is not None and not cached_df.empty:
                logger.info(
                    "[CACHE HIT] %s verisi diskten alındı.", self._provider.get_name()
                )
                return cached_df
        except Exception as e:
            logger.warning(
                "[CACHE READ WARNING] %s — API'ye gidiliyor: %s",
                self._provider.get_name(), e,
            )
        return None

    def _try_write_cache(self, criteria: Any, result: Result) -> None:
        """Başarılı sonucu cache'e yaz; hata durumunda sessizce geç."""
        if result.success and result.value is not None and not result.value.empty:
            try:
                self._cache.set(self._provider.get_name(), criteria, result.value)
            except Exception as e:
                logger.warning(
                    "[CACHE WRITE WARNING] %s: %s", self._provider.get_name(), e
                )

fetch_data_async(criteria) async

Cache'den oku; bulamazsan provider'ı çağır ve sonucu yaz.

Source code in src/selection_service/providers/providers_factory.py 
56
57
58
59
60
61
62
63
64
async def fetch_data_async(self, criteria: Any) -> Result:
    """Cache'den oku; bulamazsan provider'ı çağır ve sonucu yaz."""
    cached = self._try_read_cache(criteria)
    if cached is not None:
        return Result.ok(cached)

    result = await self._provider.fetch_data_async(criteria)
    self._try_write_cache(criteria, result)
    return result

fetch_data_sync(criteria)

Senkron cache desteği — önceki sürümde eksikti.

Mevcut (önceki) davranış: her sync çağrı cache'i bypass ediyordu. Düzeltilmiş davranış: async ile aynı cache mantığı uygulanır.

Source code in src/selection_service/providers/providers_factory.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
def fetch_data_sync(self, criteria: Any) -> Result:
    """Senkron cache desteği — önceki sürümde eksikti.

    Mevcut (önceki) davranış: her sync çağrı cache'i bypass ediyordu.
    Düzeltilmiş davranış: async ile aynı cache mantığı uygulanır.
    """
    cached = self._try_read_cache(criteria)
    if cached is not None:
        return Result.ok(cached)

    result = self._provider.fetch_data_sync(criteria)
    self._try_write_cache(criteria, result)
    return result

__getattr__(name)

fetch_ ve get_name dışındaki her çağrıyı asıl provider'a aktar.

Bu sayede CachedProviderProxy'nin IWaveformDownloader metodlarını (download_single_waveforms, download_waveforms_batch) da şeffaf biçimde iletmesi sağlanır — proxy'yi her yeni metot için güncellemek gerekmez.

Source code in src/selection_service/providers/providers_factory.py 
88
89
90
91
92
93
94
95
def __getattr__(self, name: str) -> Any:
    """fetch_ ve get_name dışındaki her çağrıyı asıl provider'a aktar.

    Bu sayede CachedProviderProxy'nin IWaveformDownloader metodlarını
    (download_single_waveforms, download_waveforms_batch) da şeffaf biçimde
    iletmesi sağlanır — proxy'yi her yeni metot için güncellemek gerekmez.
    """
    return getattr(self._provider, name)

selection_service.providers.peer_provider.PeerWest2Provider

Bases: IDataFetcher

PEER NGA-West2 veri sağlayıcı.

Yalnızca IDataFetcher uygular; dalga formu indirme desteği yoktur. Download gereken senaryolarda supports_download(provider) → False döner ve çağıran kod bu durumu yönetir (bkz. EarthquakeApi.download_waveforms).

Source code in src/selection_service/providers/peer_provider.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
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
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
class PeerWest2Provider(IDataFetcher):
    """PEER NGA-West2 veri sağlayıcı.

    Yalnızca IDataFetcher uygular; dalga formu indirme desteği yoktur.
    Download gereken senaryolarda supports_download(provider) → False döner
    ve çağıran kod bu durumu yönetir (bkz. EarthquakeApi.download_waveforms).
    """

    def __init__(self, column_mapper: Type[IColumnMapper], **kwargs: Any) -> None:
        self.column_mapper = column_mapper
        self.name = ProviderName.PEER.value
        self.flatfile_df = load_csv("NGA-West2_flatfile.csv")
        self.mapped_df: pd.DataFrame | None = None
        self.response_df: pd.DataFrame | None = None

    # ──────────────────────────────────────────────────────────────
    # IDataFetcher sözleşmesi
    # ──────────────────────────────────────────────────────────────

    def get_name(self) -> str:
        return str(self.name)

    def map_criteria(self, criteria: SearchCriteria) -> Dict[str, Any]:
        """SearchCriteria → PEER filtre parametrelerine dönüştür."""
        return criteria.to_peer_params()

    @async_result_decorator
    async def fetch_data_async(self, criteria: Dict[str, Any]) -> pd.DataFrame:
        """NGA-West2 verilerini asenkron olarak çek.

        CSV okuma CPU-bound olduğundan thread pool'a atılır;
        event loop bloke edilmez.
        """
        try:
            # get_running_loop: Python 3.10+ uyumlu (get_event_loop deprecated)
            loop = asyncio.get_running_loop()
            self.mapped_df = await loop.run_in_executor(
                None,
                partial(self.column_mapper.map_columns, self.flatfile_df.copy()),
            )
            filtered_df = await loop.run_in_executor(
                None,
                partial(self._apply_filters, self.mapped_df, criteria),
            )
            filtered_df["PROVIDER"] = str(self.name)
            return self._normalize_mechanism(filtered_df)
        except Exception as e:
            raise ProviderError(
                self.name, e, f"PEER async data fetch failed: {e}"
            )

    @result_decorator
    def fetch_data_sync(self, criteria: Dict[str, Any]) -> pd.DataFrame:
        """NGA-West2 verilerini senkron olarak çek."""
        try:
            df = self.flatfile_df.copy()
            self.mapped_df = self.column_mapper.map_columns(df=df)
            self.mapped_df = self._apply_filters(self.mapped_df, criteria)
            self.mapped_df["PROVIDER"] = str(self.name)
            self.mapped_df = self._normalize_mechanism(self.mapped_df)
            logger.info("PEER'dan %d kayıt alındı.", len(self.mapped_df))
            return self.mapped_df
        except Exception as e:
            raise ProviderError(
                self.name, e, f"PEER sync data fetch failed: {e}"
            )

    # ──────────────────────────────────────────────────────────────
    # Private yardımcılar
    # ──────────────────────────────────────────────────────────────

    def _normalize_mechanism(self, df: pd.DataFrame) -> pd.DataFrame:
        """Sayısal mekanizma sütununu metin karşılığına çevir."""
        if df["MECHANISM"].dtype in (np.int64, np.float64, int, float):
            return convert_mechanism_to_text(df)
        return df

    def _apply_filters(
        self,
        df: pd.DataFrame,
        criteria: Dict[str, Any],
    ) -> pd.DataFrame:
        """Arama kriterlerine göre DataFrame'i filtrele.

        Her kriter None ise o filtre uygulanmaz; bu sayede kısmi
        kriterlerle çalışmak mümkündür.
        """
        if df.empty:
            return df

        try:
            filtered = df.copy()

            # --- Numeric range filters (DRY: helper ile) ---
            numeric_filters: list[tuple[str, str, str]] = [
                # (criteria_min_key, criteria_max_key, df_column)
                ("min_magnitude",  "max_magnitude",  "MAGNITUDE"),
                ("min_Rjb",        "max_Rjb",        "RJB(km)"),
                ("min_Rrup",       "max_Rrup",        "RRUP(km)"),
                ("min_vs30",       "max_vs30",        "VS30(m/s)"),
                ("min_depth",      "max_depth",       "HYPO_DEPTH(km)"),
                ("min_pga",        "max_pga",         "PGA(cm2/sec)"),
                ("min_pgv",        "max_pgv",         "PGV(cm/sec)"),
                ("min_pgd",        "max_pgd",         "PGD(cm)"),
                ("year_start",     "year_end",        "YEAR"),
                ("min_latitude",   "max_latitude",    "HYPO_LAT"),
                ("min_longitude",  "max_longitude",   "HYPO_LON"),
            ]
            for min_key, max_key, col in numeric_filters:
                filtered = self._apply_range(filtered, criteria, min_key, max_key, col)

            filtered = self._apply_circle(filtered, criteria)

            # --- Kategorik filtre ---
            mechanisms = self._mechanism_filter_values(filtered, criteria)
            if mechanisms:
                filtered = filtered[
                    filtered["MECHANISM"].isin(mechanisms)
                ]

            return filtered

        except Exception as e:
            raise DataProcessingError(self.name, e, "Filter application failed")

    @staticmethod
    def _mechanism_filter_values(
        df: pd.DataFrame,
        criteria: Dict[str, Any],
    ) -> list[Any]:
        """Return mechanism filter values compatible with the DataFrame dtype."""
        values = list(criteria.get("mechanisms") or [])
        fault_type = criteria.get("fault_type")
        if fault_type is not None:
            values.append(fault_type)
        if not values or "MECHANISM" not in df.columns:
            return []

        normalized: list[Any] = []
        for value in values:
            normalized.append(value)
            if isinstance(value, str) and value in REVERSE_MECHANISM_MAP:
                normalized.append(REVERSE_MECHANISM_MAP[value])
            elif isinstance(value, (int, float, np.integer, np.floating)):
                normalized.append(MECHANISM_MAP.get(int(value), "Unknown"))

        return list(dict.fromkeys(normalized))

    @staticmethod
    def _apply_circle(df: pd.DataFrame, criteria: Dict[str, Any]) -> pd.DataFrame:
        """Dairesel arama: hiposantr, (lat, lon) merkezine radius km içinde olmalı."""
        lat0 = criteria.get("circleLatitude")
        lon0 = criteria.get("circleLongitude")
        radius = criteria.get("circleRadius")
        if None in (lat0, lon0, radius) or not {"HYPO_LAT", "HYPO_LON"} <= set(df.columns):
            return df
        lat = np.radians(pd.to_numeric(df["HYPO_LAT"], errors="coerce"))
        lon = np.radians(pd.to_numeric(df["HYPO_LON"], errors="coerce"))
        a = (
            np.sin((lat - np.radians(lat0)) / 2) ** 2
            + np.cos(np.radians(lat0)) * np.cos(lat) * np.sin((lon - np.radians(lon0)) / 2) ** 2
        )
        distance_km = 2 * 6371.0088 * np.arcsin(np.sqrt(a))
        return df[distance_km <= radius]

    @staticmethod
    def _apply_range(
        df: pd.DataFrame,
        criteria: Dict[str, Any],
        min_key: str,
        max_key: str,
        col: str,
    ) -> pd.DataFrame:
        """Tek bir sayısal aralık filtresi uygula (None-safe)."""
        if col not in df.columns:
            return df
        min_val = criteria.get(min_key)
        max_val = criteria.get(max_key)
        if min_val is not None:
            df = df[df[col] >= min_val]
        if max_val is not None:
            df = df[df[col] <= max_val]
        return df

map_criteria(criteria)

SearchCriteria → PEER filtre parametrelerine dönüştür.

Source code in src/selection_service/providers/peer_provider.py 
63
64
65
def map_criteria(self, criteria: SearchCriteria) -> Dict[str, Any]:
    """SearchCriteria → PEER filtre parametrelerine dönüştür."""
    return criteria.to_peer_params()

fetch_data_async(criteria) async

NGA-West2 verilerini asenkron olarak çek.

CSV okuma CPU-bound olduğundan thread pool'a atılır; event loop bloke edilmez.

Source code in src/selection_service/providers/peer_provider.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
@async_result_decorator
async def fetch_data_async(self, criteria: Dict[str, Any]) -> pd.DataFrame:
    """NGA-West2 verilerini asenkron olarak çek.

    CSV okuma CPU-bound olduğundan thread pool'a atılır;
    event loop bloke edilmez.
    """
    try:
        # get_running_loop: Python 3.10+ uyumlu (get_event_loop deprecated)
        loop = asyncio.get_running_loop()
        self.mapped_df = await loop.run_in_executor(
            None,
            partial(self.column_mapper.map_columns, self.flatfile_df.copy()),
        )
        filtered_df = await loop.run_in_executor(
            None,
            partial(self._apply_filters, self.mapped_df, criteria),
        )
        filtered_df["PROVIDER"] = str(self.name)
        return self._normalize_mechanism(filtered_df)
    except Exception as e:
        raise ProviderError(
            self.name, e, f"PEER async data fetch failed: {e}"
        )

fetch_data_sync(criteria)

NGA-West2 verilerini senkron olarak çek.

Source code in src/selection_service/providers/peer_provider.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
@result_decorator
def fetch_data_sync(self, criteria: Dict[str, Any]) -> pd.DataFrame:
    """NGA-West2 verilerini senkron olarak çek."""
    try:
        df = self.flatfile_df.copy()
        self.mapped_df = self.column_mapper.map_columns(df=df)
        self.mapped_df = self._apply_filters(self.mapped_df, criteria)
        self.mapped_df["PROVIDER"] = str(self.name)
        self.mapped_df = self._normalize_mechanism(self.mapped_df)
        logger.info("PEER'dan %d kayıt alındı.", len(self.mapped_df))
        return self.mapped_df
    except Exception as e:
        raise ProviderError(
            self.name, e, f"PEER sync data fetch failed: {e}"
        )

selection_service.providers.afad_provider.AFADDataProvider

Bases: IDataFetcher, IWaveformDownloader

AFAD (Tadas) veri sağlayıcı.

IDataFetcher → fetch_data_async / fetch_data_sync / map_criteria / get_name IWaveformDownloader → download_single_waveforms / download_waveforms_batch

Source code in src/selection_service/providers/afad_provider.py 
 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
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
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
class AFADDataProvider(IDataFetcher, IWaveformDownloader):
    """AFAD (Tadas) veri sağlayıcı.

    IDataFetcher   → fetch_data_async / fetch_data_sync / map_criteria / get_name
    IWaveformDownloader → download_single_waveforms / download_waveforms_batch
    """

    def __init__(
        self,
        column_mapper: Type[IColumnMapper],
        timeout: int = 30,
        api_client: AfadApiClient | None = None,
        file_manager: AfadFileManager | None = None,
    ) -> None:
        self.name = ProviderName.AFAD.value
        self.column_mapper = column_mapper
        self.timeout = timeout

        # DIP: Dışarıdan injection desteklenir; verilmezse default oluşturulur.
        self.api_client = api_client or AfadApiClient(timeout=self.timeout)
        self.file_manager = file_manager or AfadFileManager(base_dir="Afad_events")

        self.mapped_df: pd.DataFrame | None = None
        self.response_df: pd.DataFrame | None = None

    # ──────────────────────────────────────────────────────────────
    # IDataFetcher sözleşmesi
    # ──────────────────────────────────────────────────────────────

    def get_name(self) -> str:
        return str(self.name)

    def map_criteria(self, criteria: SearchCriteria) -> Dict[str, Any]:
        params = criteria.to_afad_params()
        bounds = criteria.afad_distance_bounds()
        if bounds:
            params[DISTANCE_BOUNDS_KEY] = bounds
        return params

    @staticmethod
    def _split_request(criteria: Dict[str, Any]) -> tuple[Dict[str, Any], Dict[str, Any]]:
        """AFAD istek govdesini ve istemci tarafi mesafe sinirlarini ayir."""
        request = dict(criteria)
        bounds = request.pop(DISTANCE_BOUNDS_KEY, None) or {}
        return request, bounds

    @async_result_decorator
    async def fetch_data_async(self, criteria: Dict[str, Any]) -> pd.DataFrame:
        """Asenkron veri çekme.

        AfadEmptyResultError ve AfadMappingError doğrudan yeniden fırlatılır;
        diğer hatalar ProviderError'a sarılır.
        """
        try:
            request, bounds = self._split_request(criteria)
            data = await self.api_client.search_waveforms_async(request)
            return self._process_response_data(data, bounds)
        except (AfadEmptyResultError, AfadMappingError):
            raise
        except ProviderError:
            raise
        except Exception as e:
            logger.error("AFAD asenkron veri çekme sırasında beklenmedik hata: %s", e, exc_info=True)
            raise ProviderError(
                self.name, e, f"AFAD bağlantı/veri hatası: {e}"
            ) from e

    @result_decorator
    def fetch_data_sync(self, criteria: Dict[str, Any]) -> pd.DataFrame:
        """Senkron veri çekme.

        AfadEmptyResultError ve AfadMappingError doğrudan yeniden fırlatılır;
        bu sayede Pipeline katmanı hata türüne göre farklı mesaj üretebilir.
        Diğer beklenmedik hatalar ProviderError'a sarılır.
        """
        try:
            request, bounds = self._split_request(criteria)
            data = self.api_client.search_waveforms_sync(request)
            return self._process_response_data(data, bounds)
        except (AfadEmptyResultError, AfadMappingError):
            raise  # Zaten açıklayıcı — Pipeline'a olduğu gibi ilet
        except ProviderError:
            raise
        except Exception as e:
            logger.error("AFAD senkron veri çekme sırasında beklenmedik hata: %s", e, exc_info=True)
            raise ProviderError(
                self.name, e, f"AFAD bağlantı/veri hatası: {e}"
            ) from e

    # ──────────────────────────────────────────────────────────────
    # IWaveformDownloader sözleşmesi
    # ──────────────────────────────────────────────────────────────

    @result_decorator
    def download_single_waveforms(self, filename: str, **kwargs: Any) -> bool:
        """Tekil dalga formu indir, kaydet ve çıkar."""
        payload = self._prepare_download_payload([filename], **kwargs)
        event_id = kwargs.get("event_id", int(time.time()))
        station_code = kwargs.get("station_code", "unknown")

        content = self.api_client.download_waveform(payload)
        zip_name = f"waveforms_{event_id}_{station_code}.zip"
        extracted = self.file_manager.store_download(
            content, event_id, zip_name,
            kwargs.get("export_type", "mseed"),
            raw_filename=filename,
        )
        if not extracted:
            raise ProviderError("AFAD", None, f"No files extracted for {filename}")
        return True

    @result_decorator
    def download_waveforms_batch(
        self, filenames: List[str], **kwargs: Any
    ) -> Dict[str, Any]:
        """Toplu indirme: filenames'i batch'lere böler ve her batch'i indirir.

        Args:
            filenames: İndirilecek dosya isimleri.
            **kwargs:
                batch_size  (int):  Batch başına max dosya sayısı (≤10, default 10).
                event_id    (int | List[int]): Klasör yapısı için deprem ID'si.
                export_type (str):  'mseed' | 'asc2' | 'asd'  (default 'mseed').
                file_type   (str):  'ap' (default).
                file_status (str):  'Acc' (default).
                user_name   (str):  'GuestUser' (default).

        Returns:
            {'total': int, 'downloaded': int, 'batches': List[dict]}
        """
        batch_size = min(kwargs.get("batch_size", 10), 10)
        event_id = self._download_event_id(**kwargs)
        export_type = kwargs.get("export_type", "mseed")
        retry_kwargs = {k: v for k, v in kwargs.items() if k not in ("event_id", "event_ids")}

        results: Dict[str, Any] = {
            "total": len(filenames),
            "downloaded": 0,
            "batches": [],
        }

        batches = [
            filenames[i : i + batch_size]
            for i in range(0, len(filenames), batch_size)
        ]
        logger.info("%d dosya, %d parti halinde indirilecek.", len(filenames), len(batches))

        for idx, batch_files in enumerate(batches, 1):
            try:
                payload = self._prepare_download_payload(batch_files, **kwargs)
                logger.debug("Batch %d payload hazırlandı: %s", idx, payload)

                content = self.api_client.download_waveform(payload)
                logger.debug("Batch %d indirildi, boyut: %d bytes", idx, len(content))

                zip_name = f"part_{idx}.zip"
                # Tek dosyalık partide AFAD içeriği ZIP'lemeden gönderebilir.
                raw_name = batch_files[0] if len(batch_files) == 1 else None
                extracted = self.file_manager.store_download(
                    content, event_id, zip_name, export_type, raw_filename=raw_name
                )
                count = len(extracted)
                missing = self._missing_files(batch_files, extracted)
                logger.debug("Batch %d çıkarıldı, %d dosya bulundu.", idx, count)

                results["downloaded"] += count
                results["batches"].append({
                    "batch": idx,
                    "requested": len(batch_files),
                    "count": count,
                    "missing": sorted(missing),
                    "success": not missing,
                    "extracted_files": list(extracted),
                })
                logger.info("Batch %d tamamlandı: %d dosya.", idx, count)

                if missing:
                    retry_count = self._handle_retry(
                        batch_files, extracted, event_id, **retry_kwargs
                    )
                    results["downloaded"] += retry_count
                    results["batches"][-1]["retry_recovered"] = retry_count

                time.sleep(2)

            except Exception as e:
                logger.error("Batch %d başarısız: %s", idx, e, exc_info=True)
                # Tek bir sorunlu kayıt (AFAD {"Result":101}) tüm partiyi düşürebilir;
                # dosyaları tek tek deneyerek sağlam olanları kurtar.
                recovered = self._handle_retry(batch_files, [], event_id, **retry_kwargs)
                results["downloaded"] += recovered
                results["batches"].append({
                    "batch": idx,
                    "requested": len(batch_files),
                    "success": False,
                    "error": str(e),
                    "retry_recovered": recovered,
                })

        return results

    # ──────────────────────────────────────────────────────────────
    # Ek public metotlar (IDataFetcher / IWaveformDownloader dışı)
    # ──────────────────────────────────────────────────────────────

    @result_decorator
    def get_event_details(self, event_ids: List[int]) -> pd.DataFrame:
        """Belirtilen deprem ID'leri için detay bilgisi çek."""
        all_details = []
        for event_id in event_ids:
            detail = self.api_client.get_event_details(event_id)
            if detail:
                all_details.append(detail[0] if isinstance(detail, list) else detail)
            time.sleep(0.1)
        return pd.DataFrame(all_details)

    # ──────────────────────────────────────────────────────────────
    # Private yardımcılar
    # ──────────────────────────────────────────────────────────────

    def _process_response_data(
        self, data: List[Dict], distance_bounds: Dict[str, Any] | None = None
    ) -> pd.DataFrame:
        """API yanıtını standart DataFrame'e çevir (DRY — async + sync ortak).

        Boş yanıt durumlarını sessizce geçmek yerine sebebi açıklayan
        özel hata fırlatır; çağıran katman (Pipeline) bu hataları
        kullanıcıya anlamlı mesaj olarak iletir.

        ``distance_bounds`` verilirse (ham kolon -> (min, max)) mesafe
        filtreleri burada uygulanır; değeri boş olan kayıt sınır varken elenir.

        Raises:
            AfadEmptyResultError: API sıfır kayıt döndürdüğünde veya mesafe
                filtresinden sonra kayıt kalmadığında.
            AfadMappingError: Mapper sonrası kayıt sayısı sıfıra düştüğünde.
        """
        raw_count = len(data) if data else 0
        logger.debug("AFAD API yanıtı: %d ham kayıt.", raw_count)

        if raw_count == 0:
            logger.warning(
                "AFAD API sıfır kayıt döndürdü. "
                "Arama kriteri (tarih, büyüklük, bölge) kontrol edilmeli."
            )
            raise AfadEmptyResultError(
                "AFAD API'den sonuç dönmedi.",
                criteria_hint="Tarih aralığını genişletin, büyüklük alt sınırını düşürün "
                              "veya bbox/çember arama alanını büyütün.",
            )

        self.response_df = self._apply_distance_bounds(pd.DataFrame(data), distance_bounds)
        if self.response_df.empty:
            logger.warning(
                "AFAD %d kayıt döndürdü; mesafe filtresinden sonra kayıt kalmadı.",
                raw_count,
            )
            raise AfadEmptyResultError(
                "AFAD kayıtlarından hiçbiri mesafe sınırlarına uymuyor.",
                criteria_hint="Repi/Rhyp/Rjb/Rrup aralığını genişletin.",
            )

        self.mapped_df = self.column_mapper.map_columns(df=self.response_df)
        self.mapped_df["PROVIDER"] = str(self.name)

        if "RSN" in self.mapped_df.columns:
            self.mapped_df["ENDPOINTSOURCE"] = (
                "https://tadas.afad.gov.tr/waveform-detail/"
                + self.mapped_df["RSN"].astype(str)
            )

        mapped_count = len(self.mapped_df)
        if mapped_count == 0:
            logger.error(
                "AFAD mapper sonrası 0 kayıt kaldı (ham kayıt: %d). "
                "API yanıt şeması değişmiş olabilir.",
                raw_count,
            )
            raise AfadMappingError(raw_count=raw_count)

        logger.info("AFAD'dan %d kayıt alındı (ham: %d).", mapped_count, raw_count)
        return self.mapped_df

    @staticmethod
    def _apply_distance_bounds(
        df: pd.DataFrame, distance_bounds: Dict[str, Any] | None
    ) -> pd.DataFrame:
        """Mesafe sınırlarını (dahil) AFAD yanıtına uygula."""
        if not distance_bounds or df.empty:
            return df
        mask = pd.Series(True, index=df.index)
        for column, (min_val, max_val) in distance_bounds.items():
            if column not in df.columns:
                values = pd.Series(float("nan"), index=df.index)
            else:
                values = pd.to_numeric(df[column], errors="coerce")
            if min_val is not None:
                mask &= values >= min_val
            if max_val is not None:
                mask &= values <= max_val
        return df[mask].reset_index(drop=True)

    def _prepare_download_payload(
        self, filenames: List[str], **kwargs: Any
    ) -> Dict[str, Any]:
        """AFAD download API payload'unu hazırla."""
        clean_filenames = [name for name in filenames if name]
        return {
            "filename": clean_filenames,
            "file_type": [kwargs.get("file_type", "ap")] * len(clean_filenames),
            "file_status": kwargs.get("file_status", "Acc"),
            "export_type": kwargs.get("export_type", "mseed"),
            "user_name": kwargs.get("user_name", "GuestUser"),
            "call": "afad",
        }

    def _download_event_id(self, **kwargs: Any) -> Any:
        event_id = kwargs.get("event_id")
        if event_id is not None:
            return event_id
        event_ids = kwargs.get("event_ids")
        if isinstance(event_ids, list) and event_ids:
            return event_ids[0]
        return int(time.time())

    def _missing_files(self, requested: List[str], extracted: List[str]) -> set[str]:
        # AFAD çıktı adları kayıt adına ek alır (ör. '<ad>_ap_Acc.mseed').
        extracted_names = [os.path.basename(path) for path in extracted]
        return {
            filename for filename in requested
            if filename and not any(name.startswith(filename) for name in extracted_names)
        }

    def _handle_retry(
        self,
        requested: List[str],
        extracted: List[str],
        event_id: Any,
        **kwargs: Any,
    ) -> int:
        """Eksik dosyalar için tek tek retry dene; başarıyla indirilen sayısını döndür."""
        missing = self._missing_files(requested, extracted)
        if not missing:
            return 0

        logger.warning("Eksik dosyalar, retry başlatılıyor: %s", missing)
        recovered = 0
        for filename in sorted(missing):
            try:
                result = self.download_single_waveforms(filename, event_id=event_id, **kwargs)
                if isinstance(result, Result):
                    if not result.success:
                        raise result.error or ProviderError("AFAD", None, "Retry failed")
                elif result is False:
                    raise ProviderError("AFAD", None, "Retry returned False")
                logger.info("Retry başarılı: %s indirildi.", filename)
                recovered += 1
            except Exception as e:
                logger.error("Retry başarısız: %s indirilemedi: %s", filename, e, exc_info=True)
        return recovered

fetch_data_async(criteria) async

Asenkron veri çekme.

AfadEmptyResultError ve AfadMappingError doğrudan yeniden fırlatılır; diğer hatalar ProviderError'a sarılır.

Source code in src/selection_service/providers/afad_provider.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
@async_result_decorator
async def fetch_data_async(self, criteria: Dict[str, Any]) -> pd.DataFrame:
    """Asenkron veri çekme.

    AfadEmptyResultError ve AfadMappingError doğrudan yeniden fırlatılır;
    diğer hatalar ProviderError'a sarılır.
    """
    try:
        request, bounds = self._split_request(criteria)
        data = await self.api_client.search_waveforms_async(request)
        return self._process_response_data(data, bounds)
    except (AfadEmptyResultError, AfadMappingError):
        raise
    except ProviderError:
        raise
    except Exception as e:
        logger.error("AFAD asenkron veri çekme sırasında beklenmedik hata: %s", e, exc_info=True)
        raise ProviderError(
            self.name, e, f"AFAD bağlantı/veri hatası: {e}"
        ) from e

fetch_data_sync(criteria)

Senkron veri çekme.

AfadEmptyResultError ve AfadMappingError doğrudan yeniden fırlatılır; bu sayede Pipeline katmanı hata türüne göre farklı mesaj üretebilir. Diğer beklenmedik hatalar ProviderError'a sarılır.

Source code in src/selection_service/providers/afad_provider.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
@result_decorator
def fetch_data_sync(self, criteria: Dict[str, Any]) -> pd.DataFrame:
    """Senkron veri çekme.

    AfadEmptyResultError ve AfadMappingError doğrudan yeniden fırlatılır;
    bu sayede Pipeline katmanı hata türüne göre farklı mesaj üretebilir.
    Diğer beklenmedik hatalar ProviderError'a sarılır.
    """
    try:
        request, bounds = self._split_request(criteria)
        data = self.api_client.search_waveforms_sync(request)
        return self._process_response_data(data, bounds)
    except (AfadEmptyResultError, AfadMappingError):
        raise  # Zaten açıklayıcı — Pipeline'a olduğu gibi ilet
    except ProviderError:
        raise
    except Exception as e:
        logger.error("AFAD senkron veri çekme sırasında beklenmedik hata: %s", e, exc_info=True)
        raise ProviderError(
            self.name, e, f"AFAD bağlantı/veri hatası: {e}"
        ) from e

download_single_waveforms(filename, **kwargs)

Tekil dalga formu indir, kaydet ve çıkar.

Source code in src/selection_service/providers/afad_provider.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
@result_decorator
def download_single_waveforms(self, filename: str, **kwargs: Any) -> bool:
    """Tekil dalga formu indir, kaydet ve çıkar."""
    payload = self._prepare_download_payload([filename], **kwargs)
    event_id = kwargs.get("event_id", int(time.time()))
    station_code = kwargs.get("station_code", "unknown")

    content = self.api_client.download_waveform(payload)
    zip_name = f"waveforms_{event_id}_{station_code}.zip"
    extracted = self.file_manager.store_download(
        content, event_id, zip_name,
        kwargs.get("export_type", "mseed"),
        raw_filename=filename,
    )
    if not extracted:
        raise ProviderError("AFAD", None, f"No files extracted for {filename}")
    return True

download_waveforms_batch(filenames, **kwargs)

Toplu indirme: filenames'i batch'lere böler ve her batch'i indirir.

Parameters:

Name Type Description Default
filenames List[str]

İndirilecek dosya isimleri.

 required
**kwargs Any

batch_size (int): Batch başına max dosya sayısı (≤10, default 10). event_id (int | List[int]): Klasör yapısı için deprem ID'si. export_type (str): 'mseed' | 'asc2' | 'asd' (default 'mseed'). file_type (str): 'ap' (default). file_status (str): 'Acc' (default). user_name (str): 'GuestUser' (default).

 {}

Returns:

Type Description
Dict[str, Any]

{'total': int, 'downloaded': int, 'batches': List[dict]}
Source code in src/selection_service/providers/afad_provider.py 
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
192
193
194
195
196
197
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
@result_decorator
def download_waveforms_batch(
    self, filenames: List[str], **kwargs: Any
) -> Dict[str, Any]:
    """Toplu indirme: filenames'i batch'lere böler ve her batch'i indirir.

    Args:
        filenames: İndirilecek dosya isimleri.
        **kwargs:
            batch_size  (int):  Batch başına max dosya sayısı (≤10, default 10).
            event_id    (int | List[int]): Klasör yapısı için deprem ID'si.
            export_type (str):  'mseed' | 'asc2' | 'asd'  (default 'mseed').
            file_type   (str):  'ap' (default).
            file_status (str):  'Acc' (default).
            user_name   (str):  'GuestUser' (default).

    Returns:
        {'total': int, 'downloaded': int, 'batches': List[dict]}
    """
    batch_size = min(kwargs.get("batch_size", 10), 10)
    event_id = self._download_event_id(**kwargs)
    export_type = kwargs.get("export_type", "mseed")
    retry_kwargs = {k: v for k, v in kwargs.items() if k not in ("event_id", "event_ids")}

    results: Dict[str, Any] = {
        "total": len(filenames),
        "downloaded": 0,
        "batches": [],
    }

    batches = [
        filenames[i : i + batch_size]
        for i in range(0, len(filenames), batch_size)
    ]
    logger.info("%d dosya, %d parti halinde indirilecek.", len(filenames), len(batches))

    for idx, batch_files in enumerate(batches, 1):
        try:
            payload = self._prepare_download_payload(batch_files, **kwargs)
            logger.debug("Batch %d payload hazırlandı: %s", idx, payload)

            content = self.api_client.download_waveform(payload)
            logger.debug("Batch %d indirildi, boyut: %d bytes", idx, len(content))

            zip_name = f"part_{idx}.zip"
            # Tek dosyalık partide AFAD içeriği ZIP'lemeden gönderebilir.
            raw_name = batch_files[0] if len(batch_files) == 1 else None
            extracted = self.file_manager.store_download(
                content, event_id, zip_name, export_type, raw_filename=raw_name
            )
            count = len(extracted)
            missing = self._missing_files(batch_files, extracted)
            logger.debug("Batch %d çıkarıldı, %d dosya bulundu.", idx, count)

            results["downloaded"] += count
            results["batches"].append({
                "batch": idx,
                "requested": len(batch_files),
                "count": count,
                "missing": sorted(missing),
                "success": not missing,
                "extracted_files": list(extracted),
            })
            logger.info("Batch %d tamamlandı: %d dosya.", idx, count)

            if missing:
                retry_count = self._handle_retry(
                    batch_files, extracted, event_id, **retry_kwargs
                )
                results["downloaded"] += retry_count
                results["batches"][-1]["retry_recovered"] = retry_count

            time.sleep(2)

        except Exception as e:
            logger.error("Batch %d başarısız: %s", idx, e, exc_info=True)
            # Tek bir sorunlu kayıt (AFAD {"Result":101}) tüm partiyi düşürebilir;
            # dosyaları tek tek deneyerek sağlam olanları kurtar.
            recovered = self._handle_retry(batch_files, [], event_id, **retry_kwargs)
            results["downloaded"] += recovered
            results["batches"].append({
                "batch": idx,
                "requested": len(batch_files),
                "success": False,
                "error": str(e),
                "retry_recovered": recovered,
            })

    return results

get_event_details(event_ids)

Belirtilen deprem ID'leri için detay bilgisi çek.

Source code in src/selection_service/providers/afad_provider.py 
242
243
244
245
246
247
248
249
250
251
@result_decorator
def get_event_details(self, event_ids: List[int]) -> pd.DataFrame:
    """Belirtilen deprem ID'leri için detay bilgisi çek."""
    all_details = []
    for event_id in event_ids:
        detail = self.api_client.get_event_details(event_id)
        if detail:
            all_details.append(detail[0] if isinstance(detail, list) else detail)
        time.sleep(0.1)
    return pd.DataFrame(all_details)

selection_service.providers.esm_provider.ESMProvider

Bases: IDataFetcher, IWaveformDownloader

ESM flatfile aramasi ve eventdata ile dalga formu indirme.

Source code in src/selection_service/providers/esm_provider.py 
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
192
193
194
195
196
197
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
class ESMProvider(IDataFetcher, IWaveformDownloader):
    """ESM flatfile aramasi ve eventdata ile dalga formu indirme."""

    def __init__(
        self,
        column_mapper: Type[IColumnMapper],
        timeout: int = 60,
        base_url: str = ESM_BASE_URL,
        session: Any | None = None,
        file_manager: AfadFileManager | None = None,
        **_: Any,
    ) -> None:
        self.name = ProviderName.ESM.value
        self.column_mapper = column_mapper
        self.timeout = timeout
        self.base_url = base_url.rstrip("/")
        self.session = session or requests.Session()
        self.file_manager = file_manager or AfadFileManager(base_dir="ESM_events")

    # ──────────────────────────────────────────────────────────────
    # IDataFetcher sozlesmesi
    # ──────────────────────────────────────────────────────────────

    def get_name(self) -> str:
        return str(self.name)

    def map_criteria(self, criteria: SearchCriteria) -> Dict[str, Any]:
        return {"query": criteria.to_esm_params(), "filters": criteria.esm_client_filters()}

    async def fetch_data_async(self, criteria: Dict[str, Any]):
        import asyncio

        return await asyncio.to_thread(self.fetch_data_sync, criteria)

    @result_decorator
    def fetch_data_sync(self, criteria: Dict[str, Any]) -> pd.DataFrame:
        query = dict(criteria.get("query") or {})
        query.setdefault("include-fields", ",".join(FLATFILE_FIELDS))
        try:
            response = self.session.get(
                f"{self.base_url}{FLATFILE_PATH}", params=query, timeout=self.timeout
            )
        except requests.RequestException as exc:
            raise ProviderError(self.name, exc, f"ESM baglanti hatasi: {exc}") from exc
        # FDSN benzeri servisler "veri yok" icin 204 (bazen 404) dondurur.
        if response.status_code in (204, 404):
            return pd.DataFrame()
        if response.status_code >= 400:
            raise ProviderError(
                self.name, None, f"ESM flatfile HTTP {response.status_code}: {response.text[:200]}"
            )
        raw = parse_flatfile(response.text)
        if raw.empty:
            return raw
        mapped = self.column_mapper.map_columns(raw)
        # Repi/Rhyp ortak semada yok; filtre icin gecici olarak eklenir (indeks korunur).
        mapped["REPI(km)"] = _number(raw.get("epi_dist"))
        mapped["RHYP(km)"] = _number(raw.get("hyp_dist"))
        filtered = apply_client_filters(mapped, criteria.get("filters") or {})
        filtered = filtered.drop(columns=["REPI(km)", "RHYP(km)"])
        logger.info("ESM'den %d kayit alindi (%d filtre sonrasi).", len(mapped), len(filtered))
        return filtered.reset_index(drop=True)

    # ──────────────────────────────────────────────────────────────
    # IWaveformDownloader sozlesmesi
    # ──────────────────────────────────────────────────────────────

    @result_decorator
    def download_single_waveforms(self, filename: str, **kwargs: Any) -> bool:
        """Bir kaydin dalga formlarini indirir.

        ``filename`` aramadaki ``FILE_NAME_H1`` degeridir: ``AG.ISTASYON.KONUM.KANAL``
        (kanal iki harf, ornegin HN/HG; uc bilesen birlikte iner). ``event_id`` ESM
        deprem kimligidir.
        """
        event_id = kwargs.pop("event_id", None)
        if not event_id:
            raise ProviderError(self.name, None, "ESM indirmesi icin event_id gerekli")
        params = self.eventdata_params(filename, str(event_id), **kwargs)
        try:
            response = self.session.get(
                f"{self.base_url}{EVENTDATA_PATH}", params=params, timeout=max(self.timeout, 120)
            )
        except requests.RequestException as exc:
            raise ProviderError(self.name, exc, f"ESM indirme hatasi: {exc}") from exc
        if response.status_code in (204, 404):
            raise ProviderError(self.name, None, f"ESM'de {filename} icin dalga formu yok")
        if response.status_code >= 400:
            raise ProviderError(
                self.name, None, f"ESM eventdata HTTP {response.status_code}: {response.text[:200]}"
            )
        safe_event = "".join(ch if ch.isalnum() or ch in "-_" else "_" for ch in str(event_id))
        safe_name = "".join(ch if ch.isalnum() or ch in "-_" else "_" for ch in filename)
        extension = "zip" if response.content[:4] == b"PK\x03\x04" else params["format"]
        extracted = self.file_manager.store_download(
            response.content, safe_event, f"esm_{safe_event}_{safe_name}.zip",
            params["format"], raw_filename=f"{safe_name}.{extension}",
        )
        if not extracted:
            raise ProviderError(self.name, None, f"No files extracted for {filename}")
        return True

    @result_decorator
    def download_waveforms_batch(self, filenames: List[str], **kwargs: Any) -> Dict[str, Any]:
        event_ids = kwargs.pop("event_ids", None) or kwargs.pop("event_id", None)
        if not isinstance(event_ids, (list, tuple)):
            event_ids = [event_ids] * len(filenames)
        downloaded, failures = 0, []
        for filename, event_id in zip(filenames, event_ids):
            result = self.download_single_waveforms(filename, event_id=event_id, **kwargs)
            if result.success:
                downloaded += 1
            else:
                failures.append(f"{filename}: {result.error}")
        if filenames and downloaded == 0:
            raise ProviderError(self.name, None, "; ".join(failures) or "ESM indirmesi basarisiz")
        return {"total": len(filenames), "downloaded": downloaded, "failures": failures}

    @staticmethod
    def eventdata_params(filename: str, event_id: str, **kwargs: Any) -> Dict[str, Any]:
        """FILE_NAME_H1 ve AFAD tarzi indirme seceneklerinden eventdata sorgusu."""
        parts = (str(filename).split(".") + ["", "", "", ""])[:4]
        network, station, location, channel = (part.strip() for part in parts)
        if not network or not station:
            raise ProviderError(ProviderName.ESM.value, None, f"Gecersiz ESM kayit adi: {filename}")
        export_type = str(kwargs.get("export_type") or "ascii").lower()
        status = str(kwargs.get("file_status") or "Acc")
        processing, data_type = FILE_STATUS.get(status, FILE_STATUS["Acc"])
        params = {
            "eventid": event_id,
            "network": network,
            "station": station,
            "channel": f"{channel[:2]}?" if channel else "*",
            "format": EXPORT_FORMATS.get(export_type, "ascii"),
            "processing-type": kwargs.get("processing_type") or processing,
            "data-type": kwargs.get("data_type") or data_type,
        }
        # Bos konum gonderilmez: "--" degerini servis bir secenek sanip HTTP 400 doner.
        if location:
            params["location"] = location
        return params

download_single_waveforms(filename, **kwargs)

Bir kaydin dalga formlarini indirir.

filename aramadaki FILE_NAME_H1 degeridir: AG.ISTASYON.KONUM.KANAL (kanal iki harf, ornegin HN/HG; uc bilesen birlikte iner). event_id ESM deprem kimligidir.

Source code in src/selection_service/providers/esm_provider.py 
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
192
193
194
195
196
197
198
199
200
@result_decorator
def download_single_waveforms(self, filename: str, **kwargs: Any) -> bool:
    """Bir kaydin dalga formlarini indirir.

    ``filename`` aramadaki ``FILE_NAME_H1`` degeridir: ``AG.ISTASYON.KONUM.KANAL``
    (kanal iki harf, ornegin HN/HG; uc bilesen birlikte iner). ``event_id`` ESM
    deprem kimligidir.
    """
    event_id = kwargs.pop("event_id", None)
    if not event_id:
        raise ProviderError(self.name, None, "ESM indirmesi icin event_id gerekli")
    params = self.eventdata_params(filename, str(event_id), **kwargs)
    try:
        response = self.session.get(
            f"{self.base_url}{EVENTDATA_PATH}", params=params, timeout=max(self.timeout, 120)
        )
    except requests.RequestException as exc:
        raise ProviderError(self.name, exc, f"ESM indirme hatasi: {exc}") from exc
    if response.status_code in (204, 404):
        raise ProviderError(self.name, None, f"ESM'de {filename} icin dalga formu yok")
    if response.status_code >= 400:
        raise ProviderError(
            self.name, None, f"ESM eventdata HTTP {response.status_code}: {response.text[:200]}"
        )
    safe_event = "".join(ch if ch.isalnum() or ch in "-_" else "_" for ch in str(event_id))
    safe_name = "".join(ch if ch.isalnum() or ch in "-_" else "_" for ch in filename)
    extension = "zip" if response.content[:4] == b"PK\x03\x04" else params["format"]
    extracted = self.file_manager.store_download(
        response.content, safe_event, f"esm_{safe_event}_{safe_name}.zip",
        params["format"], raw_filename=f"{safe_name}.{extension}",
    )
    if not extracted:
        raise ProviderError(self.name, None, f"No files extracted for {filename}")
    return True

eventdata_params(filename, event_id, **kwargs) staticmethod

FILE_NAME_H1 ve AFAD tarzi indirme seceneklerinden eventdata sorgusu.

Source code in src/selection_service/providers/esm_provider.py 
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
@staticmethod
def eventdata_params(filename: str, event_id: str, **kwargs: Any) -> Dict[str, Any]:
    """FILE_NAME_H1 ve AFAD tarzi indirme seceneklerinden eventdata sorgusu."""
    parts = (str(filename).split(".") + ["", "", "", ""])[:4]
    network, station, location, channel = (part.strip() for part in parts)
    if not network or not station:
        raise ProviderError(ProviderName.ESM.value, None, f"Gecersiz ESM kayit adi: {filename}")
    export_type = str(kwargs.get("export_type") or "ascii").lower()
    status = str(kwargs.get("file_status") or "Acc")
    processing, data_type = FILE_STATUS.get(status, FILE_STATUS["Acc"])
    params = {
        "eventid": event_id,
        "network": network,
        "station": station,
        "channel": f"{channel[:2]}?" if channel else "*",
        "format": EXPORT_FORMATS.get(export_type, "ascii"),
        "processing-type": kwargs.get("processing_type") or processing,
        "data-type": kwargs.get("data_type") or data_type,
    }
    # Bos konum gonderilmez: "--" degerini servis bir secenek sanip HTTP 400 doner.
    if location:
        params["location"] = location
    return params

selection_service.providers.fdsn_provider.FDSNProvider

Bases: IDataFetcher, IWaveformDownloader

FDSN olay aramasi; istege bagli kayit aramasi ve dalga formu indirme.

Source code in src/selection_service/providers/fdsn_provider.py 
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
192
193
194
195
196
197
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
class FDSNProvider(IDataFetcher, IWaveformDownloader):
    """FDSN olay aramasi; istege bagli kayit aramasi ve dalga formu indirme."""

    def __init__(
        self,
        column_mapper: IColumnMapper,
        service: str = "USGS",
        data_service: str = "IRIS",
        base_url: str | None = None,
        data_base_url: str | None = None,
        timeout: int = 30,
        session: Any | None = None,
        record_search: bool = False,
        channels: Tuple[str, ...] = DEFAULT_CHANNELS,
        max_events: int = 20,
        default_radius_km: float = 150.0,
        pre_event_s: float = 30.0,
        file_manager: AfadFileManager | None = None,
        **_: Any,
    ) -> None:
        self.column_mapper = column_mapper
        self.service = str(service).upper()
        self.data_service = str(data_service).upper()
        self.event_url = _node_url(base_url or service)
        self.data_url = _node_url(data_base_url or data_service)
        self.timeout = timeout
        self.session = session or requests.Session()
        self.record_search = record_search
        self.channels = tuple(code.upper()[:2] for code in channels)
        self.max_events = max_events
        self.default_radius_km = default_radius_km
        self.pre_event_s = pre_event_s
        self.file_manager = file_manager or AfadFileManager(base_dir="FDSN_events")
        # Kayit aramasinda gorulen depremler; indirmede zaman penceresi icin kullanilir.
        self._events: Dict[str, Dict[str, Any]] = {}
        self._distances: Dict[Tuple[str, str], float] = {}

    # ──────────────────────────────────────────────────────────────
    # IDataFetcher sozlesmesi
    # ──────────────────────────────────────────────────────────────

    def get_name(self) -> str:
        """Kayit defteri ve onbellek anahtari tum dugumler icin ayni kalir."""
        return ProviderName.FDSN.value

    def map_criteria(self, criteria: SearchCriteria) -> Dict[str, Any]:
        params = {key: value for key, value in criteria.to_fdsn_params().items() if value is not None}
        if not self.record_search:
            return params
        filters = {key: value for key, value in criteria.esm_client_filters().items()
                   if key not in UNSUPPORTED_FILTERS}
        distances = [filters.get(key) for key in ("max_Repi", "max_Rjb", "max_Rrup", "max_Rhyp")]
        radius = max([value for value in distances if value is not None], default=self.default_radius_km)
        return {
            "events": params,
            "stations": {"network": criteria.network, "station": criteria.station_code},
            "radius_km": float(radius),
            "filters": filters,
        }

    async def fetch_data_async(self, criteria: Dict[str, Any]) -> Result:
        return await asyncio.to_thread(self.fetch_data_sync, criteria)

    def fetch_data_sync(self, criteria: Dict[str, Any]) -> Result:
        try:
            if self.record_search:
                return Result.ok(self._search_records(criteria))
            return Result.ok(self._events_frame(self._query_events(criteria)))
        except Exception as exc:
            return Result.fail(ProviderError(self.get_name(), exc, f"FDSN event query failed: {exc}"))

    # ──────────────────────────────────────────────────────────────
    # Istasyon ve dalga formu sorgulari
    # ──────────────────────────────────────────────────────────────

    async def fetch_stations_async(self, **params: Any) -> Result:
        return await asyncio.to_thread(self.fetch_stations_sync, **params)

    def fetch_stations_sync(self, **params: Any) -> Result:
        """Kanal basina bir satir (ag, istasyon, konum, kanal, konum, duyarlilik)."""
        try:
            return Result.ok(self._stations_frame(self._query_stations(params)))
        except Exception as exc:
            return Result.fail(ProviderError(self.get_name(), exc, f"FDSN station query failed: {exc}"))

    async def fetch_waveforms_async(self, **params: Any) -> Result:
        return await asyncio.to_thread(self.fetch_waveforms_sync, **params)

    def fetch_waveforms_sync(self, **params: Any) -> Result:
        """Dataselect sorgusu; cozulmus :class:`~selection_service.waveform.Trace` listesi doner."""
        try:
            return Result.ok(read_mseed(self._dataselect(params)))
        except Exception as exc:
            return Result.fail(ProviderError(self.get_name(), exc, f"FDSN waveform query failed: {exc}"))

    # ──────────────────────────────────────────────────────────────
    # IWaveformDownloader sozlesmesi
    # ──────────────────────────────────────────────────────────────

    @result_decorator
    def download_single_waveforms(self, filename: str, **kwargs: Any) -> bool:
        """Bir kaydin uc bilesenini indirir.

        ``filename`` aramadaki ``FILE_NAME_H1`` degeridir (``AG.ISTASYON.KONUM.BANT``,
        ornegin ``TK.4614..HN``). ``export_type`` ``mseed`` ise ham miniSEED
        (sayim) yazilir; ``asc2``/``ascii`` icin kanal duyarliligiyla fiziksel
        birime cevrilmis ESM benzeri ASCII dosyalari yazilir. Yalniz ivme
        (``RawAcc``/``Acc``) istenebilir; islenmis seriler FDSN'de yoktur.
        """
        event_id = str(kwargs.get("event_id") or "").strip()
        if not event_id:
            raise ProviderError(self.get_name(), None, "FDSN indirmesi icin event_id gerekli")
        status = str(kwargs.get("file_status") or "Acc")
        if status not in ("RawAcc", "Acc"):
            raise ProviderError(self.get_name(), None,
                                f"FDSN yalniz ham ivme kaydi verir; {status} desteklenmiyor")
        network, station, location, band = self._split_filename(filename)
        event = self._event(event_id)
        distance = self._distances.get((event_id, filename))
        start = event["time"] - timedelta(seconds=self.pre_event_s)
        end = event["time"] + timedelta(seconds=self._post_event_s(distance))
        query = {"network": network, "station": station, "location": location or "--",
                 "channel": f"{band}?", "starttime": _fdsn_time(start), "endtime": _fdsn_time(end)}
        raw = self._dataselect(query)
        target = Path(self.file_manager.base_dir) / f"event_{_safe(event_id)}"
        target.mkdir(parents=True, exist_ok=True)
        label = _safe(f"{network}.{station}.{location}.{band}")
        export_type = str(kwargs.get("export_type") or "asc2").lower()
        if export_type == "mseed":
            (target / f"fdsn_{label}.mseed").write_bytes(raw)
            return True
        traces = self.calibrate(read_mseed(raw), self._query_stations(query))
        if not traces:
            raise ProviderError(self.get_name(), None, f"{filename} icin dalga formu yok")
        write_ascii(traces, target, f"fdsn_{label}", header={
            "EVENT_ID": event_id, "EVENT_DATE_TIME": _fdsn_time(event["time"]),
            "EVENT_LATITUDE_DEGREE": event.get("latitude"), "EVENT_LONGITUDE_DEGREE": event.get("longitude"),
            "EVENT_DEPTH_KM": event.get("depth"), "MAGNITUDE": event.get("magnitude"),
            "DATA_SOURCE": self.data_url,
        })
        return True

    @result_decorator
    def download_waveforms_batch(self, filenames: List[str], **kwargs: Any) -> Dict[str, Any]:
        event_ids = kwargs.pop("event_ids", None) or kwargs.pop("event_id", None)
        if not isinstance(event_ids, (list, tuple)):
            event_ids = [event_ids] * len(filenames)
        downloaded, failures = 0, []
        for filename, event_id in zip(filenames, event_ids):
            result = self.download_single_waveforms(filename, event_id=event_id, **kwargs)
            if result.success:
                downloaded += 1
            else:
                failures.append(f"{filename}: {result.error}")
        if filenames and downloaded == 0:
            raise ProviderError(self.get_name(), None, "; ".join(failures) or "FDSN indirmesi basarisiz")
        return {"total": len(filenames), "downloaded": downloaded, "failures": failures}

    @staticmethod
    def calibrate(traces: List[Trace], channels: List[Dict[str, str]]) -> List[Trace]:
        """Sayimlari kanal duyarliligina boler; ivmeyi cm/s^2 olarak dondurur.

        Duyarlilik (``Scale``) referans frekansinda gecerli tek bir kazanctir; ivme
        olcerlerin duz bandinda alet tepkisinin giderilmesine esdegerdir.
        """
        scales = {}
        for row in channels:
            key = ".".join(row.get(name, "") for name in ("network", "station", "location", "channel"))
            scales[key] = (_float(row.get("scale")), str(row.get("scaleunits", "")).upper())
        calibrated = []
        for trace in traces:
            scale, units = scales.get(trace.id, (math.nan, ""))
            if not math.isfinite(scale) or scale == 0:
                logger.warning("%s icin duyarlilik yok; kayit sayim olarak birakildi", trace.id)
                calibrated.append(trace)
                continue
            data = trace.data / scale
            if units.replace(" ", "") in ACCELERATION_UNITS:
                data, units = data * 100.0, "cm/s^2"
            calibrated.append(Trace(trace.network, trace.station, trace.location, trace.channel,
                                    trace.starttime, trace.sampling_rate, data, units or "unknown",
                                    {"SENSITIVITY": scale}))
        return calibrated

    # ──────────────────────────────────────────────────────────────
    # HTTP
    # ──────────────────────────────────────────────────────────────

    def _get(self, url: str, params: Dict[str, Any]) -> Optional[requests.Response]:
        query = {key: value for key, value in params.items() if value not in (None, "")}
        try:
            response = self.session.get(url, params=query, timeout=self.timeout)
        except requests.RequestException as exc:
            raise ProviderError(self.get_name(), exc, f"FDSN baglanti hatasi: {exc}") from exc
        if response.status_code in (204, 404):
            return None
        if response.status_code >= 400:
            raise ProviderError(self.get_name(), None,
                                f"FDSN HTTP {response.status_code}: {response.text[:200]}")
        return response

    def _query_events(self, params: Dict[str, Any]) -> List[Dict[str, Any]]:
        response = self._get(f"{self.event_url}/fdsnws/event/1/query", {**params, "format": "text"})
        events = []
        for row in parse_text_table(response.text if response is not None else ""):
            moment = _parse_time(row.get("time"))
            event_id = row.get("eventid", "")
            if moment is None or not event_id:
                continue
            event = {
                "id": event_id, "time": moment,
                "latitude": _float(row.get("latitude")), "longitude": _float(row.get("longitude")),
                "depth": _float(row.get("depth/km")),
                "magnitude": _float(row.get("magnitude")), "magnitude_type": row.get("magtype") or None,
                "name": row.get("eventlocationname") or None,
            }
            self._events[event_id] = event
            events.append(event)
        return events

    def _event(self, event_id: str) -> Dict[str, Any]:
        if event_id not in self._events:
            self._query_events({"eventid": event_id})
        if event_id not in self._events:
            raise ProviderError(self.get_name(), None, f"FDSN olayi bulunamadi: {event_id}")
        return self._events[event_id]

    def _query_stations(self, params: Dict[str, Any]) -> List[Dict[str, str]]:
        query = {"network": "*", "station": "*", "location": "*", "channel": "*", "level": "channel",
                 **params, "format": "text"}
        if str(query.get("level")).lower() != "channel":
            query["level"] = "channel"
        response = self._get(f"{self.data_url}/fdsnws/station/1/query", query)
        return parse_text_table(response.text if response is not None else "")

    def _dataselect(self, params: Dict[str, Any]) -> bytes:
        required = ("network", "station", "location", "channel", "starttime", "endtime")
        missing = [key for key in required if params.get(key) in (None, "")]
        if missing:
            raise ValueError("Missing FDSN waveform parameters: " + ", ".join(missing))
        query = dict(params)
        for key in ("starttime", "endtime"):
            if isinstance(query[key], datetime):
                query[key] = _fdsn_time(query[key])
        response = self._get(f"{self.data_url}/fdsnws/dataselect/1/query", query)
        if response is None or not response.content:
            raise ProviderError(self.get_name(), None, "FDSN veri merkezinde bu kayit icin veri yok")
        return response.content

    # ──────────────────────────────────────────────────────────────
    # Tablolar
    # ──────────────────────────────────────────────────────────────

    def _events_frame(self, events: List[Dict[str, Any]]) -> pd.DataFrame:
        records = [{
            "RSN": event["id"], "EVENT": event["id"], "YEAR": event["time"].year,
            "MAGNITUDE": event["magnitude"], "MAGNITUDE_TYPE": event["magnitude_type"],
            "HYPO_LAT": event["latitude"], "HYPO_LON": event["longitude"], "HYPO_DEPTH(km)": event["depth"],
            "ENDPOINTSOURCE": f"{self.event_url}/fdsnws/event/1/query?eventid={event['id']}",
        } for event in events]
        mapped = self.column_mapper.map_columns(pd.DataFrame(records))
        mapped["PROVIDER"] = f"FDSN_{self.service}"
        return mapped

    def _stations_frame(self, rows: List[Dict[str, str]]) -> pd.DataFrame:
        return pd.DataFrame([{
            "NETWORK": row.get("network"), "STATION": row.get("station"),
            "LOCATION": row.get("location"), "CHANNEL": row.get("channel"),
            "LATITUDE": _float(row.get("latitude")), "LONGITUDE": _float(row.get("longitude")),
            "ELEVATION(m)": _float(row.get("elevation")), "SAMPLE_RATE": _float(row.get("samplerate")),
            "SENSITIVITY": _float(row.get("scale")), "SENSITIVITY_UNITS": row.get("scaleunits") or None,
            "START_DATE": row.get("starttime") or None, "END_DATE": row.get("endtime") or None,
            "PROVIDER": f"FDSN_{self.data_service}",
        } for row in rows])

    def _search_records(self, criteria: Dict[str, Any]) -> pd.DataFrame:
        events = self._query_events(criteria.get("events") or {})
        events.sort(key=lambda item: -item["magnitude"] if math.isfinite(item["magnitude"]) else 0)
        if len(events) > self.max_events:
            logger.info("FDSN: %d depremden en buyuk %d tanesi icin istasyon aranacak",
                        len(events), self.max_events)
            events = events[: self.max_events]
        radius = float(criteria.get("radius_km") or self.default_radius_km)
        station_filter = criteria.get("stations") or {}
        records = []
        for event in events:
            if not (math.isfinite(event["latitude"]) and math.isfinite(event["longitude"])):
                continue
            try:
                channels = self._query_stations({
                    "network": station_filter.get("network") or "*",
                    "station": station_filter.get("station") or "*",
                    "channel": ",".join(f"{band}?" for band in self.channels),
                    "latitude": event["latitude"], "longitude": event["longitude"],
                    "maxradius": round(radius / KM_PER_DEGREE, 4),
                    "starttime": _fdsn_time(event["time"]),
                    "endtime": _fdsn_time(event["time"] + timedelta(seconds=1)),
                })
            except ProviderError as exc:
                logger.warning("FDSN istasyon sorgusu basarisiz (%s): %s", event["id"], exc)
                continue
            records.extend(self._event_records(event, channels))
        if not records:
            return self.column_mapper.map_columns(pd.DataFrame())
        frame = pd.DataFrame(records)
        mapped = self.column_mapper.map_columns(frame)
        mapped["PROVIDER"] = f"FDSN_{self.service}"
        mapped["REPI(km)"] = frame["REPI(km)"]
        mapped["RHYP(km)"] = frame["RRUP(km)"]
        filtered = apply_client_filters(mapped, criteria.get("filters") or {})
        filtered = filtered.drop(columns=["REPI(km)", "RHYP(km)"]).reset_index(drop=True)
        for _, row in filtered.iterrows():
            self._distances[(str(row["EVENT"]), str(row["FILE_NAME_H1"]))] = _float(row["RJB(km)"])
        logger.info("FDSN'den %d kayit alindi (%d filtre sonrasi).", len(frame), len(filtered))
        return filtered

    def _event_records(self, event: Dict[str, Any], channels: List[Dict[str, str]]) -> List[Dict[str, Any]]:
        best: Dict[Tuple[str, str], Tuple[Tuple[int, str], Dict[str, str]]] = {}
        for row in channels:
            band = row.get("channel", "")[:2].upper()
            if band not in self.channels:
                continue
            key = (row.get("network", ""), row.get("station", ""))
            rank = (self.channels.index(band), row.get("location", ""))
            if key not in best or rank < best[key][0]:
                best[key] = (rank, row)
        records = []
        depth = event["depth"] if math.isfinite(event["depth"]) else 0.0
        for (network, station), (_, row) in best.items():
            lat, lon = _float(row.get("latitude")), _float(row.get("longitude"))
            if not (math.isfinite(lat) and math.isfinite(lon)):
                continue
            repi = epicentral_km(event["latitude"], event["longitude"], lat, lon)
            filename = f"{network}.{station}.{row.get('location', '')}.{row.get('channel', '')[:2].upper()}"
            records.append({
                "RSN": f"{event['id']}|{filename}", "EVENT": event["id"], "YEAR": event["time"].year,
                "MAGNITUDE": event["magnitude"], "MAGNITUDE_TYPE": event["magnitude_type"],
                "HYPO_LAT": event["latitude"], "HYPO_LON": event["longitude"], "HYPO_DEPTH(km)": event["depth"],
                "STATION": station, "SSN": f"{network}.{station}", "STATION_ID": f"{network}.{station}",
                "STATION_LAT": lat, "STATION_LON": lon, "MECHANISM": "Unknown",
                "RJB(km)": round(repi, 3), "RRUP(km)": round(math.hypot(repi, depth), 3), "REPI(km)": repi,
                "FILE_NAME_H1": filename,
                "ENDPOINTSOURCE": f"{self.event_url}/fdsnws/event/1/query?eventid={event['id']}",
            })
        return records

    # ──────────────────────────────────────────────────────────────

    @staticmethod
    def _split_filename(filename: str) -> Tuple[str, str, str, str]:
        parts = (str(filename).split(".") + ["", "", "", ""])[:4]
        network, station, location, band = (part.strip() for part in parts)
        if not network or not station or len(band) < 2:
            raise ProviderError(ProviderName.FDSN.value, None, f"Gecersiz FDSN kayit adi: {filename}")
        return network, station, location, band[:2].upper()

    @staticmethod
    def _post_event_s(distance_km: Optional[float]) -> float:
        """S dalgasi ve kuyruk icin pencere: 120 s + mesafe / 3.5 km/s (en az 180 s)."""
        if distance_km is None or not math.isfinite(distance_km):
            return 300.0
        return max(180.0, 120.0 + distance_km / 3.5)

get_name()

Kayit defteri ve onbellek anahtari tum dugumler icin ayni kalir.

Source code in src/selection_service/providers/fdsn_provider.py 
166
167
168
def get_name(self) -> str:
    """Kayit defteri ve onbellek anahtari tum dugumler icin ayni kalir."""
    return ProviderName.FDSN.value

fetch_stations_sync(**params)

Kanal basina bir satir (ag, istasyon, konum, kanal, konum, duyarlilik).

Source code in src/selection_service/providers/fdsn_provider.py 
203
204
205
206
207
208
def fetch_stations_sync(self, **params: Any) -> Result:
    """Kanal basina bir satir (ag, istasyon, konum, kanal, konum, duyarlilik)."""
    try:
        return Result.ok(self._stations_frame(self._query_stations(params)))
    except Exception as exc:
        return Result.fail(ProviderError(self.get_name(), exc, f"FDSN station query failed: {exc}"))

fetch_waveforms_sync(**params)

Dataselect sorgusu; cozulmus :class:~selection_service.waveform.Trace listesi doner.

Source code in src/selection_service/providers/fdsn_provider.py 
213
214
215
216
217
218
def fetch_waveforms_sync(self, **params: Any) -> Result:
    """Dataselect sorgusu; cozulmus :class:`~selection_service.waveform.Trace` listesi doner."""
    try:
        return Result.ok(read_mseed(self._dataselect(params)))
    except Exception as exc:
        return Result.fail(ProviderError(self.get_name(), exc, f"FDSN waveform query failed: {exc}"))

download_single_waveforms(filename, **kwargs)

Bir kaydin uc bilesenini indirir.

filename aramadaki FILE_NAME_H1 degeridir (AG.ISTASYON.KONUM.BANT, ornegin TK.4614..HN). export_type mseed ise ham miniSEED (sayim) yazilir; asc2/ascii icin kanal duyarliligiyla fiziksel birime cevrilmis ESM benzeri ASCII dosyalari yazilir. Yalniz ivme (RawAcc/Acc) istenebilir; islenmis seriler FDSN'de yoktur.

Source code in src/selection_service/providers/fdsn_provider.py 
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
@result_decorator
def download_single_waveforms(self, filename: str, **kwargs: Any) -> bool:
    """Bir kaydin uc bilesenini indirir.

    ``filename`` aramadaki ``FILE_NAME_H1`` degeridir (``AG.ISTASYON.KONUM.BANT``,
    ornegin ``TK.4614..HN``). ``export_type`` ``mseed`` ise ham miniSEED
    (sayim) yazilir; ``asc2``/``ascii`` icin kanal duyarliligiyla fiziksel
    birime cevrilmis ESM benzeri ASCII dosyalari yazilir. Yalniz ivme
    (``RawAcc``/``Acc``) istenebilir; islenmis seriler FDSN'de yoktur.
    """
    event_id = str(kwargs.get("event_id") or "").strip()
    if not event_id:
        raise ProviderError(self.get_name(), None, "FDSN indirmesi icin event_id gerekli")
    status = str(kwargs.get("file_status") or "Acc")
    if status not in ("RawAcc", "Acc"):
        raise ProviderError(self.get_name(), None,
                            f"FDSN yalniz ham ivme kaydi verir; {status} desteklenmiyor")
    network, station, location, band = self._split_filename(filename)
    event = self._event(event_id)
    distance = self._distances.get((event_id, filename))
    start = event["time"] - timedelta(seconds=self.pre_event_s)
    end = event["time"] + timedelta(seconds=self._post_event_s(distance))
    query = {"network": network, "station": station, "location": location or "--",
             "channel": f"{band}?", "starttime": _fdsn_time(start), "endtime": _fdsn_time(end)}
    raw = self._dataselect(query)
    target = Path(self.file_manager.base_dir) / f"event_{_safe(event_id)}"
    target.mkdir(parents=True, exist_ok=True)
    label = _safe(f"{network}.{station}.{location}.{band}")
    export_type = str(kwargs.get("export_type") or "asc2").lower()
    if export_type == "mseed":
        (target / f"fdsn_{label}.mseed").write_bytes(raw)
        return True
    traces = self.calibrate(read_mseed(raw), self._query_stations(query))
    if not traces:
        raise ProviderError(self.get_name(), None, f"{filename} icin dalga formu yok")
    write_ascii(traces, target, f"fdsn_{label}", header={
        "EVENT_ID": event_id, "EVENT_DATE_TIME": _fdsn_time(event["time"]),
        "EVENT_LATITUDE_DEGREE": event.get("latitude"), "EVENT_LONGITUDE_DEGREE": event.get("longitude"),
        "EVENT_DEPTH_KM": event.get("depth"), "MAGNITUDE": event.get("magnitude"),
        "DATA_SOURCE": self.data_url,
    })
    return True

calibrate(traces, channels) staticmethod

Sayimlari kanal duyarliligina boler; ivmeyi cm/s^2 olarak dondurur.

Duyarlilik (Scale) referans frekansinda gecerli tek bir kazanctir; ivme olcerlerin duz bandinda alet tepkisinin giderilmesine esdegerdir.

Source code in src/selection_service/providers/fdsn_provider.py 
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
@staticmethod
def calibrate(traces: List[Trace], channels: List[Dict[str, str]]) -> List[Trace]:
    """Sayimlari kanal duyarliligina boler; ivmeyi cm/s^2 olarak dondurur.

    Duyarlilik (``Scale``) referans frekansinda gecerli tek bir kazanctir; ivme
    olcerlerin duz bandinda alet tepkisinin giderilmesine esdegerdir.
    """
    scales = {}
    for row in channels:
        key = ".".join(row.get(name, "") for name in ("network", "station", "location", "channel"))
        scales[key] = (_float(row.get("scale")), str(row.get("scaleunits", "")).upper())
    calibrated = []
    for trace in traces:
        scale, units = scales.get(trace.id, (math.nan, ""))
        if not math.isfinite(scale) or scale == 0:
            logger.warning("%s icin duyarlilik yok; kayit sayim olarak birakildi", trace.id)
            calibrated.append(trace)
            continue
        data = trace.data / scale
        if units.replace(" ", "") in ACCELERATION_UNITS:
            data, units = data * 100.0, "cm/s^2"
        calibrated.append(Trace(trace.network, trace.station, trace.location, trace.channel,
                                trace.starttime, trace.sampling_rate, data, units or "unknown",
                                {"SENSITIVITY": scale}))
    return calibrated

selection_service.waveform.mseed

miniSEED 2 okuyucu (yalniz numpy).

FDSN dataselect servisleri ham miniSEED dondurur. Bu modul kuvvetli yer hareketi kayitlarinda kullanilan kodlamalari cozer: 16/32 bit tamsayi, 32/64 bit kayan nokta, Steim1 ve Steim2. Her kanal (AG.ISTASYON.KONUM.KANAL) icin kayitlar zamana gore birlestirilir; kucuk bosluklar dogrusal interpolasyonla doldurulur, cakisan ornekler atilir.

Desteklenmeyenler: miniSEED 3, Blockette 1000 olmayan kayitlar, GEOSCOPE ve SRO gibi eski kodlamalar.

MSeedError

Bases: ValueError

miniSEED verisi cozulemedi.

Source code in src/selection_service/waveform/mseed.py 
25
26
class MSeedError(ValueError):
    """miniSEED verisi cozulemedi."""

Trace dataclass

Tek kanalin surekli ornekleri.

Tamsayi kodlanmis veri ham sayimdir (units="counts"); kayan nokta verinin birimi dosyada yazmaz ("unknown").

Source code in src/selection_service/waveform/mseed.py 
29
30
31
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
@dataclass
class Trace:
    """Tek kanalin surekli ornekleri.

    Tamsayi kodlanmis veri ham sayimdir (``units="counts"``); kayan nokta
    verinin birimi dosyada yazmaz (``"unknown"``).
    """

    network: str
    station: str
    location: str
    channel: str
    starttime: datetime
    sampling_rate: float
    data: np.ndarray
    units: str = "counts"
    meta: Dict[str, object] = field(default_factory=dict)

    @property
    def id(self) -> str:
        return f"{self.network}.{self.station}.{self.location}.{self.channel}"

    @property
    def delta(self) -> float:
        return 1.0 / self.sampling_rate

    @property
    def endtime(self) -> datetime:
        return self.starttime + timedelta(seconds=(len(self.data) - 1) * self.delta)

read_mseed(raw, max_gap_samples=10)

miniSEED baytlarini kanal basina tek :class:Trace olarak cozer.

max_gap_samples ornekten uzun bosluklar kaydi ayri parcalara boler; daha kisa bosluklar dogrusal interpolasyonla doldurulur.

Source code in src/selection_service/waveform/mseed.py 
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
def read_mseed(raw: bytes, max_gap_samples: int = 10) -> List[Trace]:
    """miniSEED baytlarini kanal basina tek :class:`Trace` olarak cozer.

    ``max_gap_samples`` ornekten uzun bosluklar kaydi ayri parcalara boler;
    daha kisa bosluklar dogrusal interpolasyonla doldurulur.
    """
    grouped: Dict[Tuple[str, str, str, str], List[_Record]] = {}
    for record in _records(bytes(raw)):
        grouped.setdefault(record.key, []).append(record)
    traces: List[Trace] = []
    for key, records in grouped.items():
        records.sort(key=lambda item: item.start)
        rate = records[0].rate
        if rate <= 0:
            continue
        units = "counts" if all(np.issubdtype(r.data.dtype, np.integer) for r in records) else "unknown"
        start = records[0].start
        pieces = [records[0].data.astype(np.float64)]
        expected = len(records[0].data)
        for record in records[1:]:
            offset = round((record.start - start).total_seconds() * rate)
            gap = offset - expected
            data = record.data.astype(np.float64)
            if gap > max_gap_samples or abs(record.rate - rate) > 1e-6 * rate:
                traces.append(_trace(key, start, rate, pieces, units))
                start, pieces, expected = record.start, [data], len(data)
                continue
            if gap > 0:
                previous = pieces[-1][-1]
                pieces.append(np.linspace(previous, data[0], gap + 2)[1:-1])
            elif gap < 0:
                data = data[-gap:]
            pieces.append(data)
            expected = offset + len(record.data)
        traces.append(_trace(key, start, rate, pieces, units))
    return traces

selection_service.waveform.ascii

Izlerin ESM benzeri ASCII dosyalarina yazilmasi.

Her bilesen icin bir dosya yazilir: ANAHTAR: deger baslik satirlari ve satir basina bir ornek. Bicim ESM ascii ciktisiyla ayni anahtarlari kullanir; boylece ESM dosyalarini okuyan uygulamalar FDSN kayitlarini da okuyabilir.

write_ascii(traces, directory, prefix, header=None)

Her izi <prefix>_<KANAL>.ASC olarak yazar ve yollari dondurur.

Source code in src/selection_service/waveform/ascii.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def write_ascii(traces: Iterable[Trace], directory: str | Path, prefix: str,
                header: Dict[str, object] | None = None) -> List[Path]:
    """Her izi ``<prefix>_<KANAL>.ASC`` olarak yazar ve yollari dondurur."""
    directory = Path(directory)
    directory.mkdir(parents=True, exist_ok=True)
    paths: List[Path] = []
    for trace in traces:
        lines = {
            "NETWORK": trace.network,
            "STATION_CODE": trace.station,
            "LOCATION": trace.location,
            "STREAM": trace.channel,
            "DATE_TIME_FIRST_SAMPLE_YYYYMMDD_HHMMSS": trace.starttime.strftime("%Y%m%d_%H%M%S.%f")[:-3],
            "SAMPLING_INTERVAL_S": f"{trace.delta:.6f}",
            "NDATA": len(trace.data),
            "UNITS": trace.units,
            **(header or {}),
            **trace.meta,
        }
        body = "\n".join(f"{value:.6e}" for value in np.asarray(trace.data, dtype=float))
        text = "\n".join(f"{key}: {value}" for key, value in lines.items()) + "\n" + body + "\n"
        path = directory / f"{prefix}_{trace.channel}.ASC"
        path.write_text(text, encoding="utf-8")
        paths.append(path)
    return paths

selection_service.providers.afad.afad_api_client.AfadApiClient

AFAD API isteklerini yöneten istemci sınıfı

Source code in src/selection_service/providers/afad/afad_api_client.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
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
class AfadApiClient:
    """AFAD API isteklerini yöneten istemci sınıfı"""

    BASE_URL = "https://ivmeservis.afad.gov.tr"
    PROCESS_URL = "https://ivmeprocessguest.afad.gov.tr"

    def __init__(self, timeout: int = 30):
        self.timeout = timeout
        self.headers = {
            'Accept': 'application/json, text/plain, */*',
            'Content-Type': 'application/json',
            'Origin': 'https://tadas.afad.gov.tr',
            'Referer': 'https://tadas.afad.gov.tr/',
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)',
            'Username': 'GuestUser',
            'IsGuest': 'true'
        }

    async def search_waveforms_async(self, criteria: Dict[str, Any]) -> Dict:
        url = f"{self.BASE_URL}/Waveforms/GetWaveforms"
        try:
            async with aiohttp.ClientSession(headers=self.headers) as session:
                request = session.post(url, json=criteria, timeout=self.timeout)
                if inspect.isawaitable(request):
                    request = await request
                async with request as response:
                    if response.status == 200:
                        response = await response.json()
                        if response is None or len(response) == 0:
                            raise NetworkError("AFAD", Exception("Empty response"), "AFAD returned empty data")
                        return response
                    error_text = await response.text()
                    raise NetworkError("AFAD", Exception(f"HTTP {response.status}: {error_text}"))
        except Exception as e:
            raise NetworkError("AFAD", e, "Async search failed")

    def search_waveforms_sync(self, criteria: Dict[str, Any]) -> Dict:
        url = f"{self.BASE_URL}/Waveforms/GetWaveforms"
        try:
            response = requests.post(url, json=criteria, headers=self.headers, timeout=self.timeout)
            if response.status_code == 200:
                return response.json()
            raise NetworkError("AFAD", Exception(f"HTTP {response.status_code}: {response.text}"))
        except Exception as e:
            raise NetworkError("AFAD", e, "Sync search failed")

    def download_waveform(self, payload: Dict[str, Any]) -> bytes:
        """Dalga formu indirme isteği
            payload örneği:
            {
                "EventId": 12345,
                "StationId": 67890,
                "ExportType": "asc2"
            }
        """
        url = f"{self.PROCESS_URL}/ExportData"
        try:
            response = requests.post(url, headers=self.headers, json=payload, timeout=50)
            response.raise_for_status()
            content = response.content or b""
            if not content:
                raise NetworkError("AFAD", Exception("Empty download response"))

            # AFAD can return an HTML or JSON error body with HTTP 200.
            # Reject those before they are saved as broken ZIP files.
            leading = content.lstrip()[:1]
            if leading in (b"<", b"{", b"["):
                text = getattr(response, "text", "")
                raise NetworkError(
                    "AFAD",
                    Exception(f"Unexpected download body: {text[:200]}"),
                    "Download response was not a binary archive",
                )

            return content
        except Exception as e:
            raise NetworkError("AFAD", e, "Download request failed")

    def get_event_details(self, event_id: int) -> Optional[Dict]:
        url = f"{self.BASE_URL}/Event/GetEventById/{event_id}"
        # Event detail için referer değişiyor, kopyasını alıp güncelle
        headers = self.headers.copy()
        headers['Referer'] = f'https://tadas.afad.gov.tr/event-detail/{event_id}'

        try:
            response = requests.get(url, headers=headers, timeout=self.timeout)
            if response.status_code == 200:
                return response.json()
            return None
        except Exception as e:
            raise NetworkError("AFAD", e, f"Event detail fetch failed for {event_id}")

download_waveform(payload)

Dalga formu indirme isteği payload örneği: { "EventId": 12345, "StationId": 67890, "ExportType": "asc2" }

Source code in src/selection_service/providers/afad/afad_api_client.py 
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
def download_waveform(self, payload: Dict[str, Any]) -> bytes:
    """Dalga formu indirme isteği
        payload örneği:
        {
            "EventId": 12345,
            "StationId": 67890,
            "ExportType": "asc2"
        }
    """
    url = f"{self.PROCESS_URL}/ExportData"
    try:
        response = requests.post(url, headers=self.headers, json=payload, timeout=50)
        response.raise_for_status()
        content = response.content or b""
        if not content:
            raise NetworkError("AFAD", Exception("Empty download response"))

        # AFAD can return an HTML or JSON error body with HTTP 200.
        # Reject those before they are saved as broken ZIP files.
        leading = content.lstrip()[:1]
        if leading in (b"<", b"{", b"["):
            text = getattr(response, "text", "")
            raise NetworkError(
                "AFAD",
                Exception(f"Unexpected download body: {text[:200]}"),
                "Download response was not a binary archive",
            )

        return content
    except Exception as e:
        raise NetworkError("AFAD", e, "Download request failed")

selection_service.providers.afad.afad_file_manager.AfadFileManager

Dosya ve ZIP işlemlerini yöneten sınıf

Source code in src/selection_service/providers/afad/afad_file_manager.py 
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 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
class AfadFileManager:
    """Dosya ve ZIP işlemlerini yöneten sınıf"""

    def __init__(self, base_dir: str = "Afad_events"):
        self.base_dir = base_dir

    def ensure_event_dir(self, event_id: int) -> str:
        path = os.path.join(self.base_dir, f"event_{event_id}")
        os.makedirs(path, exist_ok=True)
        return path

    def save_zip(self, content: bytes, event_id: int, filename: str) -> str:
        folder = self.ensure_event_dir(event_id)
        path = os.path.join(folder, filename)
        with open(path, 'wb') as f:
            f.write(content)
        return path

    def store_download(
        self,
        content: bytes,
        event_id: int,
        zip_name: str,
        export_type: str = "asc2",
        raw_filename: str | None = None,
    ) -> List[str]:
        """İndirilen içeriği kaydet; ZIP ise aç, değilse ham dosya olarak yaz.

        AFAD tek dosyalık isteklerde (ör. mseed) içeriği ZIP'lemeden gönderir.
        """
        if content[:4] == b"PK\x03\x04":
            zip_path = self.save_zip(content, event_id, zip_name)
            return self.extract_zip(zip_path, export_type)
        if not raw_filename:
            raise ProviderError("AFAD", None, "Download is not a ZIP archive and no target filename given")
        self._ensure_safe_member(raw_filename)
        if not os.path.splitext(raw_filename)[1] and export_type:
            raw_filename = f"{raw_filename}.{export_type}"
        return [self.save_zip(content, event_id, raw_filename)]

    def extract_zip(self, zip_path: str, export_type: str = "asc2") -> List[str]:
        """Güvenli zip çıkarma ve iç içe zip kontrolü"""
        extracted_files = []
        target_dir = os.path.dirname(zip_path)

        try:
            if not os.path.isfile(zip_path) or not zipfile.is_zipfile(zip_path):
                raise ProviderError("AFAD", None, "Downloaded file is not a valid ZIP archive")

            with zipfile.ZipFile(zip_path, 'r') as zf:
                if zf.testzip() is not None:
                    raise ProviderError("AFAD", None, "Corrupted ZIP file")

                for file_info in zf.infolist():
                    if file_info.is_dir():
                        continue
                    self._ensure_safe_member(file_info.filename)
                    # Dosya isminden station_id ayıklama ve klasörleme mantığı buraya eklenebilir
                    # Şimdilik basic extract yapıyoruz, Station klasörlemesi Provider içinde yönetilebilir
                    # veya buraya taşınabilir.
                    zf.extract(file_info, target_dir)
                    extracted_path = os.path.join(target_dir, file_info.filename)

                    # İç içe zip kontrolü
                    if file_info.filename.endswith('.zip') and export_type in ["asc", "asc2"]:
                         extracted_files.extend(self._extract_nested_zip(extracted_path, target_dir))
                    else:
                        extracted_files.append(extracted_path)

            if not extracted_files:
                raise ProviderError("AFAD", None, "ZIP archive did not contain extractable files")

        except Exception as e:
            raise ProviderError("AFAD", e, f"Extraction failed: {zip_path}")
        finally:
            # İşlem bitince zip'i sil
            if os.path.exists(zip_path):
                os.remove(zip_path)

        return extracted_files

    def _extract_nested_zip(self, nested_zip_path: str, target_dir: str) -> List[str]:
        files = []
        try:
            if not zipfile.is_zipfile(nested_zip_path):
                raise ProviderError("AFAD", None, "Nested archive is not a valid ZIP")
            with zipfile.ZipFile(nested_zip_path, 'r') as nz:
                if nz.testzip() is not None:
                    raise ProviderError("AFAD", None, "Corrupted nested ZIP file")
                for name in nz.namelist():
                    self._ensure_safe_member(name)
                nz.extractall(target_dir)
                files = [
                    os.path.join(target_dir, n)
                    for n in nz.namelist()
                    if not n.endswith("/")
                ]
            os.remove(nested_zip_path)
        except Exception as e:
            raise ProviderError("AFAD", e, f"Nested extraction failed: {nested_zip_path}")
        return files

    def _ensure_safe_member(self, member_name: str) -> None:
        normalized = os.path.normpath(member_name)
        if os.path.isabs(normalized) or normalized.startswith(".."):
            raise ProviderError("AFAD", None, f"Unsafe ZIP member path: {member_name}")

store_download(content, event_id, zip_name, export_type='asc2', raw_filename=None)

İndirilen içeriği kaydet; ZIP ise aç, değilse ham dosya olarak yaz.

AFAD tek dosyalık isteklerde (ör. mseed) içeriği ZIP'lemeden gönderir.

Source code in src/selection_service/providers/afad/afad_file_manager.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def store_download(
    self,
    content: bytes,
    event_id: int,
    zip_name: str,
    export_type: str = "asc2",
    raw_filename: str | None = None,
) -> List[str]:
    """İndirilen içeriği kaydet; ZIP ise aç, değilse ham dosya olarak yaz.

    AFAD tek dosyalık isteklerde (ör. mseed) içeriği ZIP'lemeden gönderir.
    """
    if content[:4] == b"PK\x03\x04":
        zip_path = self.save_zip(content, event_id, zip_name)
        return self.extract_zip(zip_path, export_type)
    if not raw_filename:
        raise ProviderError("AFAD", None, "Download is not a ZIP archive and no target filename given")
    self._ensure_safe_member(raw_filename)
    if not os.path.splitext(raw_filename)[1] and export_type:
        raw_filename = f"{raw_filename}.{export_type}"
    return [self.save_zip(content, event_id, raw_filename)]

extract_zip(zip_path, export_type='asc2')

Güvenli zip çıkarma ve iç içe zip kontrolü

Source code in src/selection_service/providers/afad/afad_file_manager.py 
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
def extract_zip(self, zip_path: str, export_type: str = "asc2") -> List[str]:
    """Güvenli zip çıkarma ve iç içe zip kontrolü"""
    extracted_files = []
    target_dir = os.path.dirname(zip_path)

    try:
        if not os.path.isfile(zip_path) or not zipfile.is_zipfile(zip_path):
            raise ProviderError("AFAD", None, "Downloaded file is not a valid ZIP archive")

        with zipfile.ZipFile(zip_path, 'r') as zf:
            if zf.testzip() is not None:
                raise ProviderError("AFAD", None, "Corrupted ZIP file")

            for file_info in zf.infolist():
                if file_info.is_dir():
                    continue
                self._ensure_safe_member(file_info.filename)
                # Dosya isminden station_id ayıklama ve klasörleme mantığı buraya eklenebilir
                # Şimdilik basic extract yapıyoruz, Station klasörlemesi Provider içinde yönetilebilir
                # veya buraya taşınabilir.
                zf.extract(file_info, target_dir)
                extracted_path = os.path.join(target_dir, file_info.filename)

                # İç içe zip kontrolü
                if file_info.filename.endswith('.zip') and export_type in ["asc", "asc2"]:
                     extracted_files.extend(self._extract_nested_zip(extracted_path, target_dir))
                else:
                    extracted_files.append(extracted_path)

        if not extracted_files:
            raise ProviderError("AFAD", None, "ZIP archive did not contain extractable files")

    except Exception as e:
        raise ProviderError("AFAD", e, f"Extraction failed: {zip_path}")
    finally:
        # İşlem bitince zip'i sil
        if os.path.exists(zip_path):
            os.remove(zip_path)

    return extracted_files

selection_service.providers.cache_manager.CacheManager

Source code in src/selection_service/providers/cache_manager.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
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
class CacheManager:
    def __init__(self, cache_dir: str = ".cache", expiry_hours: int = 24):
        self.cache_dir = cache_dir
        self.expiry_seconds = expiry_hours * 3600
        if not os.path.exists(cache_dir):
            os.makedirs(cache_dir)

    def _generate_key(self, provider_name: str, criteria: Any) -> str:
        """
        Kriterlerden benzersiz bir dosya adı üretir.
        Hem Pydantic modeli hem de Dictionary (dict) destekler.
        """
        try:
            if hasattr(criteria, 'model_dump_json'):
                # Pydantic v2
                criteria_str = criteria.model_dump_json()
            elif isinstance(criteria, dict):
                # Dictionary ise, anahtarları sırala ki hash her zaman aynı çıksın
                # default=str, datetime objelerini stringe çevirmek için gereklidir
                criteria_str = json.dumps(criteria, sort_keys=True, default=str)
            else:
                # Fallback
                criteria_str = str(criteria)

            raw_key = f"{provider_name}_{criteria_str}"
            return hashlib.md5(raw_key.encode()).hexdigest()
        except Exception as e:
            print(f"[CACHE KEY ERROR] Hash üretilemedi: {e}")
            # Hata durumunda timestamp kullanarak unique ama cachelenemeyen bir key üret
            # (Sistemi kırmamak için)
            return f"error_{int(time.time())}"

    def get(self, provider_name: str, criteria: Any) -> Optional[pd.DataFrame]:
        key = self._generate_key(provider_name, criteria)
        file_path = os.path.join(self.cache_dir, f"{key}.parquet")

        if not os.path.exists(file_path):
            return None

        # --- ZAMAN AŞIMI KONTROLÜ ---
        try:
            file_mod_time = os.path.getmtime(file_path)
            if (time.time() - file_mod_time) > self.expiry_seconds:
                print(f"[CACHE] {provider_name} verisi zaman aşımına uğradı. Siliniyor...")
                os.remove(file_path)
                return None
        except OSError:
            return None
        # ----------------------------

        try:
            # engine='pyarrow' veya 'fastparquet' kullanılabilir
            return pd.read_parquet(file_path, engine='pyarrow')
        except Exception as e:
            print(f"[CACHE READ ERROR] Okuma hatası ({key}): {e}")
            return None

    def set(self, provider_name: str, criteria: Any, df: pd.DataFrame):
        if df is None or df.empty:
            return

        key = self._generate_key(provider_name, criteria)
        file_path = os.path.join(self.cache_dir, f"{key}.parquet")

        try:
            # index=False önemli, gereksiz index sütunu oluşmasın
            df.to_parquet(file_path, engine='pyarrow', index=False)
            # print(f"[CACHE SAVED] {provider_name} -> {key}.parquet") 
        except Exception as e:
            print(f"[CACHE WRITE ERROR] Yazma hatası ({file_path}): {e}")