3699777599

Build Type

push

github

Committed by GitHub

Commit Message

Add initial mypy integration (#241)

Run Details

223 of 223 new or added lines in 45 files covered. (100.0%)

4575 of 6107 relevant lines covered (74.91%)

0.75 hits per line

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

95.35

/winterdrp/processors/database/database_modifier.py

import logging
from abc import ABC
from collections.abc import Callable
from typing import Optional

import astropy.io.fits
import numpy as np
import pandas as pd
from astropy.io.fits import Header

from winterdrp.data import ImageBatch, SourceBatch
from winterdrp.processors.base_processor import (
    BaseDataframeProcessor,
    BaseImageProcessor,
)
from winterdrp.processors.database.base_database_processor import (
    BaseDatabaseProcessor,
    DataBaseError,
)
from winterdrp.processors.database.database_importer import (
    BaseDatabaseImporter,
    BaseImageDatabaseImporter,
)
from winterdrp.processors.database.postgres import (
    get_sequence_keys_from_table,
    modify_db_entry,
)

logger = logging.getLogger(__name__)


class BaseDatabaseModifier(BaseDatabaseImporter, ABC):
    base_key = "dbmodifier"

    def __init__(self, db_alter_columns: Optional[str] = None, *args, **kwargs):
        super(BaseDatabaseModifier, self).__init__(
            db_output_columns=db_alter_columns, *args, **kwargs
        )
        self.db_alter_columns = db_alter_columns


class ImageDatabaseModifier(BaseDatabaseModifier, BaseImageDatabaseImporter):
    def __init__(self, *args, **kwargs):
        super(ImageDatabaseModifier, self).__init__(*args, **kwargs)

    def _apply_to_images(
        self,
        batch: ImageBatch,
    ) -> ImageBatch:
        for image in batch:
            query_columns, accepted_values, accepted_types = self.get_constraints(image)
            logger.info(f"{query_columns}, {accepted_values}, {accepted_types}")

            modify_db_entry(
                value_dict=image,
                db_query_columns=query_columns,
                db_query_values=accepted_values,
                db_query_comparison_types=accepted_types,
                db_alter_columns=self.db_alter_columns,
                db_table=self.db_table,
                db_name=self.db_name,
                db_user=self.db_user,
                password=self.db_password,
            )

        return batch


class ModifyImageDatabaseSeq(ImageDatabaseModifier):
    def __init__(self, sequence_key: Optional[str | list[str]] = None, *args, **kwargs):
        super(ModifyImageDatabaseSeq, self).__init__(*args, **kwargs)
        self.sequence_key = sequence_key

    def get_constraints(self, image):
        if self.sequence_key is None:
            self.sequence_key = [
                x
                for x in get_sequence_keys_from_table(
                    self.db_table, self.db_name, self.db_user, self.db_password
                )
            ]

        accepted_values = [image[x.upper()] for x in self.sequence_key]
        accepted_types = ["="] * len(accepted_values)
        return self.sequence_key, accepted_values, accepted_types


class DataframeDatabaseModifier(BaseDatabaseModifier, BaseDataframeProcessor):
    def __init__(self, *args, **kwargs):
        super(DataframeDatabaseModifier, self).__init__(*args, **kwargs)

    def _apply_to_candidates(
        self,
        batch: SourceBatch,
    ) -> SourceBatch:
        return SourceBatch

1	import logging	1✔
2	from abc import ABC	1✔
3	from collections.abc import Callable	1✔
4	from typing import Optional	1✔
5
6	import astropy.io.fits	1✔
7	import numpy as np	1✔
8	import pandas as pd	1✔
9	from astropy.io.fits import Header	1✔
10
11	from winterdrp.data import ImageBatch, SourceBatch	1✔
12	from winterdrp.processors.base_processor import (	1✔
13	BaseDataframeProcessor,
14	BaseImageProcessor,
15	)
16	from winterdrp.processors.database.base_database_processor import (	1✔
17	BaseDatabaseProcessor,
18	DataBaseError,
19	)
20	from winterdrp.processors.database.database_importer import (	1✔
21	BaseDatabaseImporter,
22	BaseImageDatabaseImporter,
23	)
24	from winterdrp.processors.database.postgres import (	1✔
25	get_sequence_keys_from_table,
26	modify_db_entry,
27	)
28
29	logger = logging.getLogger(__name__)	1✔
30
31
32	class BaseDatabaseModifier(BaseDatabaseImporter, ABC):	1✔
33	base_key = "dbmodifier"	1✔
34
35	def __init__(self, db_alter_columns: Optional[str] = None, args, *kwargs):	1✔
36	super(BaseDatabaseModifier, self).__init__(	1✔
37	db_output_columns=db_alter_columns, args, *kwargs
38	)
39	self.db_alter_columns = db_alter_columns	1✔
40
41
42	class ImageDatabaseModifier(BaseDatabaseModifier, BaseImageDatabaseImporter):	1✔
43	def __init__(self, args, *kwargs):	1✔
44	super(ImageDatabaseModifier, self).__init__(args, *kwargs)	1✔
45
46	def _apply_to_images(	1✔
47	self,
48	batch: ImageBatch,
49	) -> ImageBatch:
50	for image in batch:	1✔
51	query_columns, accepted_values, accepted_types = self.get_constraints(image)	1✔
52	logger.info(f"{query_columns}, {accepted_values}, {accepted_types}")	1✔
53
54	modify_db_entry(	1✔
55	value_dict=image,
56	db_query_columns=query_columns,
57	db_query_values=accepted_values,
58	db_query_comparison_types=accepted_types,
59	db_alter_columns=self.db_alter_columns,
60	db_table=self.db_table,
61	db_name=self.db_name,
62	db_user=self.db_user,
63	password=self.db_password,
64	)
65
66	return batch	1✔
67
68
69	class ModifyImageDatabaseSeq(ImageDatabaseModifier):	1✔
70	def __init__(self, sequence_key: Optional[str \| list[str]] = None, args, *kwargs):	1✔
71	super(ModifyImageDatabaseSeq, self).__init__(args, *kwargs)	1✔
72	self.sequence_key = sequence_key	1✔
73
74	def get_constraints(self, image):	1✔
75	if self.sequence_key is None:	1✔
76	self.sequence_key = [	1✔
77	x
78	for x in get_sequence_keys_from_table(
79	self.db_table, self.db_name, self.db_user, self.db_password
80	)
81	]
82
83	accepted_values = [image[x.upper()] for x in self.sequence_key]	1✔
84	accepted_types = ["="] * len(accepted_values)	1✔
85	return self.sequence_key, accepted_values, accepted_types	1✔
86
87
88	class DataframeDatabaseModifier(BaseDatabaseModifier, BaseDataframeProcessor):	1✔
89	def __init__(self, args, *kwargs):	1✔
90	super(DataframeDatabaseModifier, self).__init__(args, *kwargs)	×
91
92	def _apply_to_candidates(	1✔
93	self,
94	batch: SourceBatch,
95	) -> SourceBatch:
96	return SourceBatch	×

winter-telescope / winterdrp / 3699777599

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