Coverage for src/susi/reduc/pipeline/blocks/block_p.py: 78%

1#!/usr/bin/env python3 

2# -*- coding: utf-8 -*- 

3""" 

4Module that holds the block `P` of the susi pipeline. 

5 

6@author: hoelken 

7""" 

8from ....io import FitsBatch, Fits 

9from ....base import Logging 

10from ..processing_data import ProcessingData 

11from .block import Block, BlockRegistry 

12 

13log = Logging.get_logger() 

14 

15 

16class BlockP(Block): 

17 """ 

18 ## BLOCK P: Polarimetry and Physical units 

19 

20 This block takes care of converting the data useful to science 

21 - TODO dual beem correction 

22 - TODO crosstalk 

23 - TODO Axes calibration (x,y,lambda) 

24 """ 

25 

26 BLOCK_ID = 'P' 

27 

28 @staticmethod 

29 def start(batch: FitsBatch, proc_data: ProcessingData) -> FitsBatch: 

30 """ 

31 Callback-like entry point for the block 

32 """ 

33 return BlockP(batch, proc_data).run().result 

34 

35 def _algorithm(self): 

36 log.debug("'%s' is not implemented and will be ignored...", self.__class__) 

37 for entry in self.batch: 

38 self.result.batch.append(self._process_entry(entry)) 

39 

40 def _process_entry(self, fits: Fits) -> dict: 

41 return { 

42 'file': fits.path, 

43 'data': self._modify_data(fits), 

44 'header': self._modify_header(fits), 

45 } 

46 

47 def _modify_data(self, fits: Fits): 

48 return fits.data 

49 

50 def _modify_header(self, fits: Fits): 

51 Fits.override_header(fits.header, Block.BLOCKS_APPLIED, BlockP.BLOCK_ID, append=True) 

52 return fits.header 

53 

54 

55BlockRegistry().register(BlockP.BLOCK_ID, BlockP)