#1523

Committed 12 Dec 2025 06:38PM UTC coverage: 42.55% (+0.1%) from 42.432%

Build # #1523

Build Type

push

coveralls-python

Committed by

gjgilbert

Commit Message

Merge remote-tracking branch 'refs/remotes/origin/spectral-extraction' into spectral-extraction

Run Details

42 of 84 new or added lines in 12 files covered. (50.0%)

1257 existing lines in 12 files now uncovered.

11503 of 27034 relevant lines covered (42.55%)

0.43 hits per line

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

0.0

/modules/Utils/kpf_parse_framework.py

#### WARNING: These classes haven't been been verified to work yet!

from modules.Utils.kpf_parse import HeaderParse
from kpfpipe.primitives.core import KPF_Primitive

# External dependencies
from keckdrpframework.models.action import Action
from keckdrpframework.models.arguments import Arguments
from keckdrpframework.models.processing_context import ProcessingContext


class GetDataProductsFramework(KPF_Primitive):
    """
    This framework primative implements the get_data_products methods 
    from modules/Utils/kpf_parse.py

    Description:
        - `action (keckdrpframework.models.action.Action)`: `action.args` contains 
                   positional arguments and keyword arguments passed by the 
                   `get_data_products` event issued in the recipe:

            - `action.args[0] (kpf_object)`: L0/2D/L1/L2 object to be analyzed
            - `action.args[1] (str)`: data level ('L0', '2D', 'L1', 'L2')
    """

    def __init__(self,
                 action: Action,
                 context: ProcessingContext) -> None:
        KPF_Primitive.__init__(self, action, context)

    def _pre_condition(self) -> bool:
        success = len(self.action.args) >= 2 and isinstance(self.action.args[1], str)
        return success

    def _post_condition(self) -> bool:
        return True

    def _perform(self):
        kpf_object = self.action.args[0]
        data_level_str = self.action.args[1] # L0, D2, L1, or 2D

        if data_level_str == 'L0':
            data_products = get_dp_L0(kpf_object)
        elif data_level_str == '2D':
            data_products = get_dp_2D(kpf_object)
        elif data_level_str == 'L1':
            data_products = get_dp_L1(kpf_object)
        elif data_level_str == 'L2':
            data_products = get_dp_L2(kpf_object)
        else:
            data_prodcts = ['None']

        return Arguments(data_products)


class GetNameFramework(KPF_Primitive):
    """
    This framework primative implements the HeaderParse.get_name() method 
    from modules/Utils/kpf_parse.py

    Description:
        - `action (keckdrpframework.models.action.Action)`: `action.args` contains 
                   positional arguments and keyword arguments passed by the 
                   `get_data_products` event issued in the recipe:

            - `action.args[0] (kpf_object)`: L0/2D/L1/L2 object to be analyzed
    """

    def __init__(self,
                 action: Action,
                 context: ProcessingContext) -> None:
        KPF_Primitive.__init__(self, action, context)

    def _pre_condition(self) -> bool:
        success = len(self.action.args) >= 1
        return success

    def _post_condition(self) -> bool:
        return True

    def _perform(self):
        kpf_object = self.action.args[0]

        # Extract the "name"
        # possible values = 'Bias', 'Dark', 'Flat', 'Wide Flat', 
        #                   'LFC', 'Etalon', 'ThAr', 'UNe', 
        #                   'Sun', 'Star'
        from modules.Utils.kpf_parse import HeaderParse
        hdr = HeaderParse(kpf_object, 'PRIMARY')
        name = hdr.get_name(use_star_names=False)

        return Arguments(name)

1	#### WARNING: These classes haven't been been verified to work yet!
2
NEW 3	from modules.Utils.kpf_parse import HeaderParse	×
4	from kpfpipe.primitives.core import KPF_Primitive	×
5
6	# External dependencies
UNCOV 7	from keckdrpframework.models.action import Action	×
UNCOV 8	from keckdrpframework.models.arguments import Arguments	×
9	from keckdrpframework.models.processing_context import ProcessingContext	×
10
11
UNCOV 12	class GetDataProductsFramework(KPF_Primitive):	×
13	"""
14	This framework primative implements the get_data_products methods
15	from modules/Utils/kpf_parse.py
16
17	Description:
18	- `action (keckdrpframework.models.action.Action)`: `action.args` contains
19	positional arguments and keyword arguments passed by the
20	`get_data_products` event issued in the recipe:
21
22	- `action.args[0] (kpf_object)`: L0/2D/L1/L2 object to be analyzed
23	- `action.args[1] (str)`: data level ('L0', '2D', 'L1', 'L2')
24	"""
25
26	def __init__(self,	×
27	action: Action,
28	context: ProcessingContext) -> None:
29	KPF_Primitive.__init__(self, action, context)	×
30
UNCOV 31	def _pre_condition(self) -> bool:	×
32	success = len(self.action.args) >= 2 and isinstance(self.action.args[1], str)	×
33	return success	×
34
35	def _post_condition(self) -> bool:	×
36	return True	×
37
UNCOV 38	def _perform(self):	×
39	kpf_object = self.action.args[0]	×
40	data_level_str = self.action.args[1] # L0, D2, L1, or 2D	×
41
42	if data_level_str == 'L0':	×
43	data_products = get_dp_L0(kpf_object)	×
44	elif data_level_str == '2D':	×
45	data_products = get_dp_2D(kpf_object)	×
46	elif data_level_str == 'L1':	×
UNCOV 47	data_products = get_dp_L1(kpf_object)	×
48	elif data_level_str == 'L2':	×
UNCOV 49	data_products = get_dp_L2(kpf_object)	×
50	else:
UNCOV 51	data_prodcts = ['None']	×
52
53	return Arguments(data_products)	×
54
55
NEW 56	class GetNameFramework(KPF_Primitive):	×
57	"""
58	This framework primative implements the HeaderParse.get_name() method
59	from modules/Utils/kpf_parse.py
60
61	Description:
62	- `action (keckdrpframework.models.action.Action)`: `action.args` contains
63	positional arguments and keyword arguments passed by the
64	`get_data_products` event issued in the recipe:
65
66	- `action.args[0] (kpf_object)`: L0/2D/L1/L2 object to be analyzed
67	"""
68
NEW 69	def __init__(self,	×
70	action: Action,
71	context: ProcessingContext) -> None:
NEW 72	KPF_Primitive.__init__(self, action, context)	×
73
NEW 74	def _pre_condition(self) -> bool:	×
NEW 75	success = len(self.action.args) >= 1	×
NEW 76	return success	×
77
NEW 78	def _post_condition(self) -> bool:	×
NEW 79	return True	×
80
NEW 81	def _perform(self):	×
NEW 82	kpf_object = self.action.args[0]	×
83
84	# Extract the "name"
85	# possible values = 'Bias', 'Dark', 'Flat', 'Wide Flat',
86	# 'LFC', 'Etalon', 'ThAr', 'UNe',
87	# 'Sun', 'Star'
NEW 88	from modules.Utils.kpf_parse import HeaderParse	×
NEW 89	hdr = HeaderParse(kpf_object, 'PRIMARY')	×
NEW 90	name = hdr.get_name(use_star_names=False)	×
91
NEW 92	return Arguments(name)	×

Keck-DataReductionPipelines / KPF-Pipeline / #1523

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