start SOM plugin

[LuisSanchez25]

This plugin will be the start for me to modify amstrax in a way that we can use to process data with the SOM classification

[tunnell]

This looks very dangerous. What are you trying to do? I would redefine the context creation code for now

[LuisSanchez25]

This is the same way we set-up this context in straxen, why do you say its dangerous?

[tunnell]

The PR is already in github, but the problem is the following. Similarly to how it was done in straxen I made a new plugin -> PeaksSOM and a context under which this plugin can function, I had to change some of the details due to the different structures between the two but I am running into a mailbox issue coming from strax which claims the argument 'records' were given to compute. However, the PeaksSOM is trying to first compute what was previously under "Peaks". And the peaks plugin does have a compute argument with records however it doesnt make sense that the current SOM plugin flags this error since there was a computation of 'records' before I made any changes.

I will include the output error as well as an example of how I triggered it:

user = 'lsanchez'
import sys

# The following is bad but its how its done here
sys.path.insert(0, f'/data/xenon/xams_v2/users/{user}/GITHUB/ls_dir/amstrax/')
sys.path.insert(0, f'/data/xenon/xams_v2/users/{user}/GITHUB/strax/')

st = amstrax.contexts.xmas_som(output_folder=f'/data/xenon/xams_v2/users/{user}/data/', init_rundb=False)

st.set_config(dict(
    # elife=5000
    peak_min_area = 3,
    s1_min_area=1,
    s2_min_area=100,
    s2_min_width=750,
    s1_max_width=140,
    s1_max_area_fraction_top=1,
    s1_min_channels=1,
    s2_min_channels=1,
    gain_to_pe_array=[1/185.3, 1/239.0, 1/212.0, 1/228.2, 1/200.0, 1.], # 18-12-2023, runs 4226 and 4229
    )
)

run_ids_na_short = [f'{i:06}' for i in range(4233, 4290+1)] # 20 second na22 runs

run_ids = run_ids_na

peaks = st.get_array(run_ids, ('peaks', 'peak_basics', 'peak_positions'))

TypeError                                 Traceback (most recent call last)
Cell In[5], line 3
      1 run_ids = run_ids_na
----> 3 peaks = st.get_array(run_ids, ('peaks', 'peak_basics', 'peak_positions'))
      4 #records = st.get_array(run_ids, ('records'))
      5 #events = st.get_array(run_ids, ('events', 'event_basics', 'event_info'))

File /data/xenon/xams_v2/users/lsanchez/GITHUB/strax/strax/context.py:1607, in Context.get_array(self, run_id, targets, save, max_workers, **kwargs)
   1604     raise RuntimeError("Cannot allow_multiple with get_array/get_df")
   1606 if len(run_ids) > 1:
-> 1607     results = strax.multi_run(
   1608         self.get_array,
   1609         run_ids,
   1610         targets=targets,
   1611         log=self.log,
   1612         save=save,
   1613         max_workers=max_workers,
   1614         **kwargs,
   1615     )
   1616 else:
   1617     source = self.get_iter(
   1618         run_ids[0], targets, save=save, max_workers=max_workers, **kwargs
   1619     )

File /data/xenon/xams_v2/users/lsanchez/GITHUB/strax/strax/utils.py:601, in multi_run(exec_function, run_ids, max_workers, throw_away_result, multi_run_progress_bar, ignore_errors, log, *args, **kwargs)
    599         failures.append(_run_id)
    600         continue
--> 601     raise f.exception()
    603 if throw_away_result:
    604     continue

File /data/xenon/xams_v2/anaconda/xams/lib/python3.8/concurrent/futures/thread.py:57, in _WorkItem.run(self)
     54     return
     56 try:
---> 57     result = self.fn(*self.args, **self.kwargs)
     58 except BaseException as exc:
     59     self.future.set_exception(exc)

File /data/xenon/xams_v2/users/lsanchez/GITHUB/strax/strax/context.py:1620, in Context.get_array(self, run_id, targets, save, max_workers, **kwargs)
   1616 else:
   1617     source = self.get_iter(
   1618         run_ids[0], targets, save=save, max_workers=max_workers, **kwargs
   1619     )
-> 1620     results = [x.data for x in source]
   1622 results = np.concatenate(results)
   1623 return results

File /data/xenon/xams_v2/users/lsanchez/GITHUB/strax/strax/context.py:1620, in <listcomp>(.0)
   1616 else:
   1617     source = self.get_iter(
   1618         run_ids[0], targets, save=save, max_workers=max_workers, **kwargs
   1619     )
-> 1620     results = [x.data for x in source]
   1622 results = np.concatenate(results)
   1623 return results

File /data/xenon/xams_v2/users/lsanchez/GITHUB/strax/strax/context.py:1486, in Context.get_iter(self, run_id, targets, save, max_workers, time_range, seconds_range, time_within, time_selection, selection, selection_str, keep_columns, drop_columns, allow_multiple, progress_bar, _chunk_number, **kwargs)
   1478     generator.throw(
   1479         OutsideException(
   1480             "Terminating due to an exception originating from outside "
   1481             "strax's get_iter (which we cannot retrieve)."
   1482         )
   1483     )
   1485 except Exception as e:
-> 1486     generator.throw(e)
   1487     raise ValueError(f"Failed to process chunk {n_chunks}!")
   1489 if not seen_a_chunk:

File /data/xenon/xams_v2/users/lsanchez/GITHUB/strax/strax/context.py:1452, in Context.get_iter(self, run_id, targets, save, max_workers, time_range, seconds_range, time_within, time_selection, selection, selection_str, keep_columns, drop_columns, allow_multiple, progress_bar, _chunk_number, **kwargs)
   1450 pbar.strax_print_time = time.perf_counter()
   1451 pbar.mbs = []
-> 1452 for n_chunks, result in enumerate(strax.continuity_check(generator), 1):
   1453     seen_a_chunk = True
   1454     if not isinstance(result, strax.Chunk):

File /data/xenon/xams_v2/users/lsanchez/GITHUB/strax/strax/chunk.py:298, in continuity_check(chunk_iter)
    295 last_runid = None
    297 last_subrun = {"run_id": None}
--> 298 for chunk in chunk_iter:
    299     if chunk.run_id != last_runid:
    300         last_end = None

File /data/xenon/xams_v2/users/lsanchez/GITHUB/strax/strax/processor.py:312, in ThreadedMailboxProcessor.iter(self)
    306 if exc is not None:
    307     # Reraise exception. This is outside the except block
    308     # to avoid the 'during handling of this exception, another
    309     # exception occurred' stuff from confusing the traceback
    310     # which is printed for the user
    311     self.log.debug("Reraising exception")
--> 312     raise exc.with_traceback(traceback)
    314 # Check the savers for any exception that occurred during saving
    315 # These are thrown back to the mailbox, but if that has already closed
    316 # it doesn't trigger a crash...
    317 # TODO: add savers inlined by parallelsourceplugin
    318 # TODO: need to look at plugins too if we ever implement true
    319 # multi-target mode
    320 for k, saver_list in self.components.savers.items():

File /data/xenon/xams_v2/users/lsanchez/GITHUB/strax/strax/mailbox.py:274, in Mailbox._send_from(self, iterable)
    268                 raise MailboxReadTimeout(
    269                     f"{self} could not progress beyond {i}, "
    270                     "no driving subscriber requested it."
    271                 )
    273 try:
--> 274     x = next(iterable)
    275 except StopIteration:
    276     # No need to send this yet, close will do that
    277     break

File /data/xenon/xams_v2/users/lsanchez/GITHUB/strax/strax/plugins/plugin.py:505, in Plugin.iter(self, iters, executor)
    503             yield new_future
    504         else:
--> 505             yield from self._iter_compute(chunk_i=chunk_i, **inputs_merged)
    507 except IterDone:
    508     # Check all sources are exhausted.
    509     # This is more than a check though -- it ensure the content of
    510     # all sources are requested all the way (including the final
    511     # Stopiteration), as required by lazy-mode processing requires
    512     for d in iters.keys():

File /data/xenon/xams_v2/users/lsanchez/GITHUB/strax/strax/plugins/plugin.py:532, in Plugin._iter_compute(self, chunk_i, **inputs_merged)
    530 def _iter_compute(self, chunk_i, **inputs_merged):
    531     """Either yields or returns strax chunks from the input."""
--> 532     yield self.do_compute(chunk_i=chunk_i, **inputs_merged)

File /data/xenon/xams_v2/users/lsanchez/GITHUB/strax/strax/plugins/plugin.py:619, in Plugin.do_compute(self, chunk_i, **kwargs)
    617     kwargs["start"] = start
    618     kwargs["end"] = end
--> 619 result = self.compute(**kwargs)
    621 return self._fix_output(result, start, end)

TypeError: compute() got an unexpected keyword argument 'records'

[tunnell]

@LuisSanchez25 We cannot run this without the xams_som_v1.npz… please add.

[JelleAalbers]

This should help:

    def compute(self, records, start, end):
        # Current classification
        peaks_classifcation = super().compute(records, start, end)

[LuisSanchez25]

@LuisSanchez25 We cannot run this without the xams_som_v1.npz… please add.

Ah I assumed people here would have access to the stoomboot machine which is where the file is saved, I can add it in the PR directly

[coveralls]

Pull Request Test Coverage Report for Build 9118375458

Details

• 54 of 165 (32.73%) changed or added relevant lines in 5 files are covered.
• 1 unchanged line in 1 file lost coverage.
• Overall coverage decreased (-0.5%) to 41.758%

---

Changes Missing Coverage	Covered Lines	Changed/Added Lines	%
amstrax/contexts.py	1	4	25.0%
amstrax/plugins/peaks/merged_peaks.py	16	21	76.19%
amstrax/plugins/peaks/peaks_som.py	32	135	23.7%

Files with Coverage Reduction	New Missed Lines	%
amstrax/contexts.py	1	73.33%

Totals
Change from base Build 7220191619:	-0.5%
Covered Lines:	1259
Relevant Lines:	3015

---

💛 - Coveralls

[LuisSanchez25]

This should help:

    def compute(self, records, start, end):
        # Current classification
        peaks_classifcation = super().compute(records, start, end)

Hey @JelleAalbers thanks a lot for your help! I was misunderstanding how the super function works, I think I can fix the rest from here thanks a lot!