16224615683

Committed 11 Jul 2025 04:13PM UTC coverage: 48.564% (-2.1%) from 50.63%

Build # 16224615683

Build Type

Pull #46

github

Committed by

web-flow

Commit Message

Merge be40794fd into 484f19548

Pull Request Pull Request #46: Gp/feat/mission set

Run Details

49 of 145 branches covered (33.79%)

Branch coverage included in aggregate %.

78 of 185 new or added lines in 13 files covered. (42.16%)

10 existing lines in 4 files now uncovered.

526 of 1039 relevant lines covered (50.63%)

0.51 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

28.26

/src/socm/workflows/ml_null_tests/time_null_test.py

from datetime import timedelta
from typing import Dict, List, Optional, Union

import numpy as np
from sotodlib.core import Context

from socm.workflows.ml_null_tests import NullTestWorkflow


class TimeNullTestWorkflow(NullTestWorkflow):
    """
    A workflow for time null tests.
    """

    chunk_nobs: Optional[int] = None
    chunk_duration: Optional[timedelta] = None
    nsplits: int = 8
    name: str = "time_null_test_workflow"

    def _get_splits(self, ctx: Context, obs_info: Dict[str, Dict[str, Union[float, str]]]) -> List[List[str]]:
        """
        Distribute the observations across splits based on the context and observation IDs.
        """
        if self.chunk_nobs is None and self.chunk_duration is None:
            raise ValueError("Either chunk_nobs or duration must be set.")
        elif self.chunk_nobs is not None and self.chunk_duration is not None:
            raise ValueError("Only one of chunk_nobs or duration can be set.")
        elif self.chunk_nobs is None:
            # Decide the chunk size based on the duration. Each chunk needs to have the
            # observations that their start times are just less than chunk_duration.
            raise NotImplementedError("Splitting by duration is not implemented yet. Please set chunk_nobs.")
        sorted_ids = sorted(obs_info, key=lambda k: obs_info[k]["start_time"])
        # Group in chunks of 10 observations.
        obs_lists = np.array_split(sorted_ids, self.chunk_nobs)
        splits = [[] for _ in range(self.nsplits)]
        for i, obs_list in enumerate(obs_lists):
            splits[i % self.nsplits] += obs_list.tolist()

        return splits

    @classmethod
    def get_workflows(cls, desc=None) -> List[NullTestWorkflow]:
        """
        Create a list of NullTestWorkflows instances from the provided descriptions.
        """

        time_workflow = cls(**desc)

        workflows = []
        for split in time_workflow._splits:
            desc = time_workflow.model_dump(exclude_unset=True)
            desc["datasize"] = 0
            desc["query"] = "obs_id IN ("
            for oid in split:
                desc["query"] += f"'{oid}',"
            desc["query"] = desc["query"].rstrip(",")
            desc["query"] += ")"
            desc["chunk_nobs"] = 1
            desc["output_dir"] = f"{time_workflow.output_dir}/mission_split_{len(workflows) + 1}"
            workflow = NullTestWorkflow(**desc)
            workflows.append(workflow)

        return workflows

1	from datetime import timedelta	1✔
2	from typing import Dict, List, Optional, Union	1✔
3
4	import numpy as np	1✔
5	from sotodlib.core import Context	1✔
6
7	from socm.workflows.ml_null_tests import NullTestWorkflow	1✔
8
9
10	class TimeNullTestWorkflow(NullTestWorkflow):	1✔
11	"""
12	A workflow for time null tests.
13	"""
14
15	chunk_nobs: Optional[int] = None	1✔
16	chunk_duration: Optional[timedelta] = None	1✔
17	nsplits: int = 8	1✔
18	name: str = "time_null_test_workflow"	1✔
19
20	def _get_splits(self, ctx: Context, obs_info: Dict[str, Dict[str, Union[float, str]]]) -> List[List[str]]:	1✔
21	"""
22	Distribute the observations across splits based on the context and observation IDs.
23	"""
NEW 24	if self.chunk_nobs is None and self.chunk_duration is None:	×
NEW 25	raise ValueError("Either chunk_nobs or duration must be set.")	×
NEW 26	elif self.chunk_nobs is not None and self.chunk_duration is not None:	×
NEW 27	raise ValueError("Only one of chunk_nobs or duration can be set.")	×
NEW 28	elif self.chunk_nobs is None:	×
29	# Decide the chunk size based on the duration. Each chunk needs to have the
30	# observations that their start times are just less than chunk_duration.
NEW 31	raise NotImplementedError("Splitting by duration is not implemented yet. Please set chunk_nobs.")	×
NEW 32	sorted_ids = sorted(obs_info, key=lambda k: obs_info[k]["start_time"])	×
33	# Group in chunks of 10 observations.
NEW 34	obs_lists = np.array_split(sorted_ids, self.chunk_nobs)	×
NEW 35	splits = [[] for _ in range(self.nsplits)]	×
NEW 36	for i, obs_list in enumerate(obs_lists):	×
NEW 37	splits[i % self.nsplits] += obs_list.tolist()	×
38
NEW 39	return splits	×
40
41	@classmethod	1✔
42	def get_workflows(cls, desc=None) -> List[NullTestWorkflow]:	1✔
43	"""
44	Create a list of NullTestWorkflows instances from the provided descriptions.
45	"""
46
NEW 47	time_workflow = cls(**desc)	×
48
NEW 49	workflows = []	×
NEW 50	for split in time_workflow._splits:	×
NEW 51	desc = time_workflow.model_dump(exclude_unset=True)	×
NEW 52	desc["datasize"] = 0	×
NEW 53	desc["query"] = "obs_id IN ("	×
NEW 54	for oid in split:	×
NEW 55	desc["query"] += f"'{oid}',"	×
NEW 56	desc["query"] = desc["query"].rstrip(",")	×
NEW 57	desc["query"] += ")"	×
NEW 58	desc["chunk_nobs"] = 1	×
NEW 59	desc["output_dir"] = f"{time_workflow.output_dir}/mission_split_{len(workflows) + 1}"	×
NEW 60	workflow = NullTestWorkflow(**desc)	×
NEW 61	workflows.append(workflow)	×
62
NEW 63	return workflows	×

simonsobs / so_campaign_manager / 16224615683

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous