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

17.5

/src/socm/utils/misc.py

from sqlite3 import Cursor
from typing import Dict, Tuple
from uuid import uuid4


def generate_ids(namespace: str) -> str:
    return namespace + uuid4().hex[:8]


def dict_factory(cursor: Cursor, row: Tuple) -> Dict:
    d = {}
    for idx, col in enumerate(cursor.description):
        d[col[0]] = row[idx]
    return d


def linear_func(x: float, a: float, b: float) -> float:
    return a * x + b


def get_workflow_entries(campaign_dict, subcampaign_map=None):
    """
    Extract workflow entries from a campaign dictionary using a predefined mapping.

    Args:
        campaign_dict: A dictionary containing campaign configuration
        subcampaign_map: A dictionary mapping subcampaign names to lists of their workflow names
                         E.g., {"ml-null-test": ["mission-tests", "wafer-tests"]}

    Returns:
        Dictionary containing workflow entries
    """
    campaign_data = campaign_dict.get("campaign", {})

    # Default empty map if none provided
    if subcampaign_map is None:
        subcampaign_map = {}

    # Collect all workflows (direct and from subcampaigns)
    workflows = {}

    for key, value in campaign_data.items():
        # Skip non-dictionary values
        if not isinstance(value, dict):
            continue

        # Check if this is a known subcampaign
        if key in subcampaign_map:
            # Process known workflows for this subcampaign
            subcampaign_name = key
            subcampaign_workflows = subcampaign_map[key]
            
            # Create a copy of the subcampaign config without its workflows
            subcampaign_common_config = {k: v for k, v in value.items() 
                                        if k not in subcampaign_workflows}
            
            for workflow_name in subcampaign_workflows:
                if workflow_name in value:
                    # Start with the workflow's own config
                    workflow_config = value[workflow_name].copy()
                    
                    # Update with common subcampaign config
                    workflow_config.update(subcampaign_common_config)
                    
                    if isinstance(workflow_config, dict):
                        # Create combined key: subcampaign.workflow_name
                        workflows[f"{subcampaign_name}.{workflow_name}"] = workflow_config
        else:
            # Treat as regular workflow
            workflows[key] = value

    return workflows

1	from sqlite3 import Cursor	1✔
2	from typing import Dict, Tuple	1✔
3	from uuid import uuid4	1✔
4
5
6	def generate_ids(namespace: str) -> str:	1✔
7	return namespace + uuid4().hex[:8]	×
8
9
10	def dict_factory(cursor: Cursor, row: Tuple) -> Dict:	1✔
11	d = {}	×
12	for idx, col in enumerate(cursor.description):	×
13	d[col[0]] = row[idx]	×
14	return d	×
15
16
17	def linear_func(x: float, a: float, b: float) -> float:	1✔
18	return a * x + b	×
19
20
21	def get_workflow_entries(campaign_dict, subcampaign_map=None):	1✔
22	"""
23	Extract workflow entries from a campaign dictionary using a predefined mapping.
24
25	Args:
26	campaign_dict: A dictionary containing campaign configuration
27	subcampaign_map: A dictionary mapping subcampaign names to lists of their workflow names
28	E.g., {"ml-null-test": ["mission-tests", "wafer-tests"]}
29
30	Returns:
31	Dictionary containing workflow entries
32	"""
NEW 33	campaign_data = campaign_dict.get("campaign", {})	×
34
35	# Default empty map if none provided
NEW 36	if subcampaign_map is None:	×
NEW 37	subcampaign_map = {}	×
38
39	# Collect all workflows (direct and from subcampaigns)
NEW 40	workflows = {}	×
41
NEW 42	for key, value in campaign_data.items():	×
43	# Skip non-dictionary values
NEW 44	if not isinstance(value, dict):	×
NEW 45	continue	×
46
47	# Check if this is a known subcampaign
NEW 48	if key in subcampaign_map:	×
49	# Process known workflows for this subcampaign
NEW 50	subcampaign_name = key	×
NEW 51	subcampaign_workflows = subcampaign_map[key]	×
52
53	# Create a copy of the subcampaign config without its workflows
NEW 54	subcampaign_common_config = {k: v for k, v in value.items()	×
55	if k not in subcampaign_workflows}
56
NEW 57	for workflow_name in subcampaign_workflows:	×
NEW 58	if workflow_name in value:	×
59	# Start with the workflow's own config
NEW 60	workflow_config = value[workflow_name].copy()	×
61
62	# Update with common subcampaign config
NEW 63	workflow_config.update(subcampaign_common_config)	×
64
NEW 65	if isinstance(workflow_config, dict):	×
66	# Create combined key: subcampaign.workflow_name
NEW 67	workflows[f"{subcampaign_name}.{workflow_name}"] = workflow_config	×
68	else:
69	# Treat as regular workflow
NEW 70	workflows[key] = value	×
71
NEW 72	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