3597069237

Build Type

github

Committed by GitHub

Commit Message

Merge 7b7dc3e1e into 3f47e794f

Pull Request Pull Request #224: Code with Style

Run Details

1490 of 1490 new or added lines in 93 files covered. (100.0%)

4571 of 6109 relevant lines covered (74.82%)

0.75 hits per line

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

95.24

/winterdrp/processors/database/database_modifier.py

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

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

winter-telescope / winterdrp / 3597069237

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