style: update logging level

.github/actions/setup/action.yaml

@@ -23,6 +23,8 @@ runs:
     - name: Setup MPI
       if: ${{ inputs.mpi == 'true' }}
       uses: mpi4py/setup-mpi@v1
+      with:
+        mpi: openmpi
     - name: Install project no MPI
       if: ${{ inputs.mpi == 'false' }}
       shell: sh

.github/workflows/precommit.yaml

@@ -1,4 +1,4 @@
-name: pre-commit checks
+name: Pre-commit checks
 
 on:
   pull_request:

.github/workflows/test.yaml

@@ -19,26 +19,12 @@ jobs:
       with:
         python-version: ${{env.python_version}}
         mpi: 'true'
-    - name: pytest
+    - name: Run pytest
       run: pytest tests/. integration_tests/. --cov=noisepy.seis --cov=noisepy.monitoring
     - name: Upload coverage reports to Codecov
       uses: codecov/codecov-action@v3
       env:
         CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
-  unit_tests_win:
-    runs-on: windows-latest
-    steps:
-    - name: Checkout Repo
-      uses: actions/checkout@v4
-    - name: Setup NoisePy
-      uses: ./.github/actions/setup
-      with:
-        python-version: ${{env.python_version}}
-        mpi : 'true'
-    - name: pytest
-      run: |
-        set PYTHONPATH=src
-        pytest tests/. integration_tests/.
   s1_s2:
     strategy:
       fail-fast: true

pyproject.toml

@@ -50,7 +50,7 @@ dependencies = [
     "PyYAML==6.0",
     "pydantic-yaml==1.0",
     "psutil>=5.9.5,<6.0.0",
-    "noisepy-seis-io>=0.3.0",
+    "noisepy-seis-io>=0.3.5",
     "scipy==1.12.0"
 ]
 
@@ -86,7 +86,7 @@ sql = [
     "SQLite3-0611",
 ]
 mpi = [
-    "mpi4py>=3.1.4,<4.0.0",
+    "mpi4py",
 ]
 aws = [
     "boto3>=1.26.0,<2.0.0",

src/noisepy/seis/__init__.py

@@ -27,6 +27,8 @@
 """
 
 logging.basicConfig(
-    level=logging.INFO, format="%(asctime)s %(thread)d %(levelname)s %(module)s.%(funcName)s(): %(message)s"
+    level=logging.INFO,
+    format="%(asctime)s | %(levelname)s | %(module)s.%(funcName)s() | %(message)s",
+    datefmt="%Y-%m-%d %H:%M:%S",
 )
 logger = logging.getLogger(__name__)

src/noisepy/seis/correlate.py

@@ -84,9 +84,9 @@ def filter_by_lat(s: Channel, d: Channel) -> bool:
     # Force config validation
     fft_params = ConfigParameters.model_validate(dict(fft_params), strict=True)
 
-    tlog = TimeLogger(logger, logging.INFO, prefix="CC Main")
+    tlog = TimeLogger(logger=logger, level=logging.DEBUG, prefix="CC MAIN")
     t_s1_total = tlog.reset()
-    logger.info(f"Starting Cross-Correlation with {os.cpu_count()} cores")
+    logger.info(f"Starting cross-correlation with {os.cpu_count()} cores")
 
     def init() -> List:
         # set variables to broadcast
@@ -120,7 +120,7 @@ def cc_timespan(
     pair_filter: Callable[[Channel, Channel], bool] = lambda src, rec: True,
 ) -> List[Tuple[Station, Station]]:
     executor = ThreadPoolExecutor()
-    tlog = TimeLogger(logger, logging.INFO, prefix="CC Main")
+    tlog = TimeLogger(logger=logger, level=logging.DEBUG, prefix="CC MAIN")
     """
     LOADING NOISE DATA AND DO FFT
     """
@@ -129,7 +129,7 @@ def cc_timespan(
     t_chunk = tlog.reset()  # for tracking overall chunk processing time
     all_channels = raw_store.get_channels(ts)
     all_channel_count = len(all_channels)
-    tlog.log(f"get {all_channel_count} channels")
+    tlog.log(f"getting {all_channel_count} channels")
     all_channels = list(filter(lambda c: c.station.valid(), all_channels))
     all_stations = set([c.station for c in all_channels])
     if all_channel_count > len(all_channels):
@@ -144,15 +144,14 @@ def cc_timespan(
 
     stations = set([station for pair in station_pairs for station in pair])
     _ = list(executor.map(lambda s: cc_store.contains(s, s, ts), stations))
-    tlog.log(f"check for {len(stations)} stations already done (warm up cache)")
     station_pair_dones = list(executor.map(lambda p: cc_store.contains(p[0], p[1], ts), station_pairs))
 
     missing_pairs = [pair for pair, done in zip(station_pairs, station_pair_dones) if not done]
     # get a set of unique stations from the list of pairs
     missing_stations = set([station for pair in missing_pairs for station in pair])
     # Filter the channels to only the missing stations
     missing_channels = list(filter(lambda c: c.station in missing_stations, all_channels))
-    tlog.log("check for stations already done")
+    tlog.log("checking for stations already done")
 
     logger.info(
         f"Still need to process: {len(missing_stations)}/{len(all_stations)} stations, "
@@ -161,7 +160,7 @@ def cc_timespan(
         f"for {ts}"
     )
     if len(missing_channels) == 0:
-        logger.warning(f"{ts} already completed")
+        logger.info(f"{ts} already completed")
         return []
 
     ch_data_tuples = _read_channels(
@@ -173,10 +172,10 @@ def cc_timespan(
         logger.warning(f"No data available for {ts}")
         return missing_pairs
 
-    tlog.log(f"Read channel data: {len(ch_data_tuples)} channels")
+    tlog.log(f"reading {len(ch_data_tuples)} channels")
     ch_data_tuples_pre = preprocess_all(executor, ch_data_tuples, raw_store, fft_params, ts)
     del ch_data_tuples
-    tlog.log(f"Preprocess: {len(ch_data_tuples_pre)} channels")
+    tlog.log(f"preprocessing {len(ch_data_tuples_pre)} channels")
     if len(ch_data_tuples_pre) == 0:
         logger.warning(f"No data available for {ts} after preprocessing")
         return missing_pairs
@@ -198,7 +197,7 @@ def cc_timespan(
     ch_data_tuples_pre.clear()
     del ch_data_tuples_pre
     gc.collect()
-    fft_datas = get_results(fft_refs, "Compute ffts")
+    fft_datas = get_results(fft_refs, "Computing ffts")
     for ix_ch, fft_data in enumerate(fft_datas):
         if fft_data.fft.size > 0:
             ffts[ix_ch] = fft_data
@@ -237,10 +236,10 @@ def cc_timespan(
             save_exec,
         )
         tasks.append(t)
-    compute_results = get_results(tasks, "Cross correlation")
+    compute_results = get_results(tasks, "Cross-correlating")
     _, save_tasks = zip(*compute_results)
     save_tasks = [t for t in save_tasks if t]
-    _ = get_results(save_tasks, "Save correlations")
+    _ = get_results(save_tasks, "Saving correlations")
     failed_pairs = [
         pair[0]
         for pair, (comp_res, save_task) in zip(work_items, compute_results)
@@ -298,7 +297,7 @@ def stations_cross_correlation(
     cc_store: CrossCorrelationDataStore,
     executor: Executor,
 ) -> Tuple[bool, Future]:
-    tlog = TimeLogger(logger, logging.DEBUG)
+    tlog = TimeLogger(logger=logger, level=logging.DEBUG, prefix="CC STATION")
     datas = []
     try:
         if cc_store.contains(src, rec, ts):
@@ -378,7 +377,7 @@ def preprocess_all(
     channels = list(zip(*ch_data))[0]
     stream_refs = [executor.submit(preprocess, raw_store, t[0], t[1], fft_params, ts) for t in ch_data]
     del ch_data
-    new_streams = get_results(stream_refs, "Pre-process")
+    new_streams = get_results(stream_refs, "Pre-processing")
     # Log if any streams were removed during pre-processing
     for ch, st in zip(channels, new_streams):
         if len(st) == 0:
@@ -479,7 +478,7 @@ def _read_channels(
     single_freq: bool = True,
 ) -> List[Tuple[Channel, ChannelData]]:
     ch_data_refs = [executor.submit(_safe_read_data, store, ts, ch) for ch in channels]
-    ch_data = get_results(ch_data_refs, "Read channel data")
+    ch_data = get_results(ch_data_refs, "Reading data")
     tuples = list(filter(lambda tup: tup[1].data.size > 0, zip(channels, ch_data)))
 
     return _filter_channel_data(tuples, sampling_rate, single_freq)
@@ -503,9 +502,9 @@ def _filter_channel_data(
         return []
     if single_freq:
         closest_freq = _get_closest_freq(frequencies, sampling_rate)
-        logger.info(f"Picked {closest_freq} as the closest sampling_rate to {sampling_rate}. ")
+        logger.debug(f"Picked {closest_freq} as the closest sampling_rate to {sampling_rate}. ")
         filtered_tuples = list(filter(lambda tup: tup[1].sampling_rate == closest_freq, tuples))
-        logger.info(f"Filtered to {len(filtered_tuples)}/{len(tuples)} channels with sampling rate == {closest_freq}")
+        logger.debug(f"Filtered to {len(filtered_tuples)}/{len(tuples)} channels with sampling rate == {closest_freq}")
     else:
         filtered_tuples = list(filter(lambda tup: tup[1].sampling_rate >= sampling_rate, tuples))
         # for each station, pick the closest >= to sampling_rate
@@ -545,5 +544,5 @@ def check_memory(params: ConfigParameters, nsta: int) -> int:
             "Require %5.3fG memory but only %5.3fG provided)! Reduce inc_hours to avoid this issue!"
             % (memory_size, MAX_MEM)
         )
-    logger.info(f"Require {memory_size:5.2f}gb memory for cross correlations")
+    logger.debug(f"Require {memory_size:5.2f}GB memory for correlations")
     return nseg_chunk

src/noisepy/seis/fdsn_download.py

@@ -82,7 +82,7 @@ def download(direc: str, prepro_para: ConfigParameters) -> None:
     client = Client(prepro_para.client_url_key)
     executor = ThreadPoolExecutor()
 
-    tlog = TimeLogger(logger, logging.INFO)
+    tlog = TimeLogger(logger=logger, level=logging.DEBUG, prefix="Download")
     t_tot = tlog.reset()
     dlist = os.path.join(direc, "station.csv")  # CSV file for station location info
     prepro_para.respdir = os.path.join(
@@ -158,10 +158,10 @@ def download(direc: str, prepro_para: ConfigParameters) -> None:
                 else:
                     location.append("*")
                 nsta += 1
-    tlog.log("Getting inventory")
+    tlog.log("getting inventory")
     # rough estimation on memory needs (assume float32 dtype)
     nsec_chunk = prepro_para.inc_hours / 24 * 86400
-    nseg_chunk = int(np.floor((nsec_chunk - prepro_para.cc_len) / prepro_para.step)) + 1
+    nseg_chunk = int(np.floor((nsec_chunk - prepro_para.cc_len) / prepro_para.step))
     npts_chunk = int(nseg_chunk * prepro_para.cc_len * prepro_para.sampling_rate)
     memory_size = nsta * npts_chunk * 4 / 1024**3
     if memory_size > MAX_MEM:

src/noisepy/seis/noise_module.py

@@ -179,7 +179,7 @@ def preprocess_raw(
                 raise ValueError("The response found in the inventory is empty (no stages)! abort!  %s" % st[0])
             else:
                 try:
-                    logger.info("removing response for %s using inv" % st[0])
+                    logger.debug("removing response for %s using inv" % st[0])
                     st[0].attach_response(inv)
                     st[0].remove_response(output=rm_resp_out, pre_filt=pre_filt, water_level=60)
                 except Exception as e:
@@ -188,14 +188,14 @@ def preprocess_raw(
                     return st
 
         elif rm_resp == RmResp.SPECTRUM:  # TODO: to be implement
-            logger.info("remove response using spectrum")
+            logger.debug("remove response using spectrum")
             specfile = glob.glob(os.path.join(respdir, "*" + station + "*"))
             if len(specfile) == 0:
                 raise ValueError("no response sepctrum found for %s" % station)
             st = resp_spectrum(st, specfile[0], sampling_rate, pre_filt)
 
         elif rm_resp == RmResp.RESP:  # TODO: to be implement
-            logger.info("remove response using RESP files")
+            logger.debug("remove response using RESP files")
             resp = glob.glob(os.path.join(respdir, "RESP." + station + "*"))
             if len(resp) == 0:
                 raise ValueError("no RESP files found for %s" % station)

src/noisepy/seis/stack.py

@@ -52,7 +52,7 @@ def stack_cross_correlations(
 
     # Use 'spawn' to avoid issues with multiprocessing on linux and 'fork'
     executor = ProcessPoolExecutor(mp_context=get_context("spawn"))
-    tlog = TimeLogger(logger=logger, level=logging.INFO)
+    tlog = TimeLogger(logger=logger, level=logging.DEBUG, prefix="STACK MAIN")
     t_tot = tlog.reset()
 
     stations = set(fft_params.stations)
@@ -85,10 +85,10 @@ def initializer():
     pairs_node = [pairs_all[i] for i in scheduler.get_indices(pairs_all)]
 
     tasks = [executor.submit(stack_store_pair, p[0], p[1], cc_store, stack_store, fft_params) for p in pairs_node]
-    results = get_results(tasks, "Stacking Pairs")
+    results = get_results(tasks, "Stacking pairs")
     executor.shutdown()
     scheduler.synchronize()
-    tlog.log("step 2 in total", t_tot)
+    tlog.log("Step 2 in total", t_tot)
     if not all(results):
         failed = [p for p, r in zip(pairs_node, results) if not r]
         failed_str = "\n".join(map(str, failed))
@@ -108,15 +108,15 @@ def stack_store_pair(
     try:
         ts = DateTimeRange(fft_params.start_date, fft_params.end_date)
         if stack_store.contains(src_sta, rec_sta, ts):
-            logger.info(f"Stack already exists for {src_sta}-{rec_sta}/{ts}")
+            logger.debug(f"Stack already exists for {src_sta}-{rec_sta}/{ts}")
             return True
-        logger.info(f"Stacking {src_sta}_{rec_sta}/{ts}")
+        logger.debug(f"Stacking {src_sta}_{rec_sta}/{ts}")
         timespans = cc_store.get_timespans(src_sta, rec_sta)
         stacks = stack_pair(src_sta, rec_sta, timespans, cc_store, fft_params)
         if len(stacks) == 0:
             logger.warning(f"No stacks for {src_sta}_{rec_sta}")
             return False
-        tlog = TimeLogger(logger=logger, level=logging.INFO)
+        tlog = TimeLogger(logger=logger, level=logging.DEBUG, prefix="STACK PAIR")
         stack_store.append(ts, src_sta, rec_sta, stacks)
         tlog.log(f"writing stack pair {(src_sta, rec_sta)}")
         return True
@@ -132,7 +132,7 @@ def stack_pair(
     cc_store: CrossCorrelationDataStore,
     fft_params: ConfigParameters,
 ) -> List[Stack]:
-    tlog = TimeLogger(logger=logger, level=logging.INFO)
+    tlog = TimeLogger(logger=logger, level=logging.DEBUG, prefix="STACK PAIR")
     # check if it is auto-correlation
     if src_sta == rec_sta:
         fauto = 1
@@ -169,7 +169,7 @@ def stack_pair(
     iseg = 0
     for ts in timespans:
         if ts.end_datetime > fft_params.end_date or ts.start_datetime < fft_params.start_date:
-            logger.warning(
+            logger.debug(
                 f"Skipping {ts} for {src_sta}-{rec_sta} because it is outside the requested time range "
                 f"({fft_params.start_date} - {fft_params.end_date})"
             )
@@ -307,7 +307,7 @@ def append_stacks(comp: str, tparameters: Dict[str, Any], stack_data: List[Tuple
                     (StackMethod.ROBUST, bigstack_rotated2[icomp]),
                 ]
                 append_stacks(comp, tparameters, stacks)
-    tlog.log(f"stack/rotate all station pairs {(src_sta,rec_sta)}", t_load)
+    tlog.log(f"stacking/rotating all station pairs {(src_sta,rec_sta)}", t_load)
     return stack_results