16725506371

Committed 04 Aug 2025 02:09PM UTC coverage: 86.311% (-3.3%) from 89.629%

Build # 16725506371

Build Type

push

github

Committed by

web-flow

Commit Message

Release v1.1.0

Run Details

350 of 449 new or added lines in 22 files covered. (77.95%)

9 existing lines in 2 files now uncovered.

2396 of 2776 relevant lines covered (86.31%)

1.73 hits per line

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

88.68

/src/Preprocessor.py

import logging
from datetime import datetime, timezone

from jsonpath_ng.parser import JsonPathParser

from src.model.SchemaConcepts.Schema_Concept import parse_datetime


class Preprocessor:
    """
    Use / adapt / extend for final preprocessing steps before converting a dictionary into the according pydantic class instances
    """

    parser = JsonPathParser()

    unit_normalization = {
        'deg': 'degree',
        'degr': 'degree',
        '°': 'degree',
        'um': 'µm',
        'Secs': 's',
        'Mins': 'min',
        'μs': 'us'
    }


    @staticmethod
    def normalize_unit(input_value) -> str:
        if input_value in Preprocessor.unit_normalization.keys():
            return Preprocessor.unit_normalization[input_value]
        return input_value

    @staticmethod
    def normalize_all_units(input_dict):
        """
        Inplace normalization of all values in fields "unit"
        :param input_dict: dictionary to replace units in
        :return: None
        """
        unit_field_names = ["unit", "coordinatesUnit"]
        for field_name in unit_field_names:
            unit_fields = Preprocessor.parser.parse("$.." + field_name)
            unit_matches = [m for m in unit_fields.find(input_dict)]
            for m in unit_matches:
                if type(m.value) != str: continue #TODO: should this be possible?
                original_value = m.value
                if not Preprocessor.unit_normalization.get(original_value): continue

                normalized_value = Preprocessor.unit_normalization[original_value]
                if normalized_value != original_value:
                    m.full_path.update(input_dict, normalized_value)

    @staticmethod
    def normalize_datetime(input_value) -> str:
        if type(input_value) == dict:
            if not input_value.get("Date") and input_value.get("Time"): # Not possible to handle only Time
                logging.warning("Encountered complex date field, but cannot interpret it")
                return input_value
            if input_value.get("Date") and not input_value.get("Time"): # Handle only Date
                input_value["Time"] = "00:00:00"
                logging.info("Input with date information but no time information found. Setting time to 00:00:00")
            input_value = input_value.get("Date") + " " + input_value.get("Time")
        output_value = parse_datetime(input_value)
        if type(output_value) == datetime:
            if output_value.tzinfo:
                output_value = output_value.astimezone(timezone.utc) # datetime has timezone info, convert it to UTC
            else:
                output_value = output_value.replace(tzinfo=timezone.utc) # No timezone, assume it's already in UTC
            return output_value.isoformat().replace("+00:00", "Z")
        return input_value

    @staticmethod
    def normalize_all_datetimes(input_dict):
        fields_for_normalization = ["creationTime", "startTime", "endTime"] #we could do it more generically but may want to limit it to specific fields

        for f in fields_for_normalization:
            date_fields = Preprocessor.parser.parse("$.." + f)
            date_matches = [m for m in date_fields.find(input_dict)]
            for m in date_matches:
                original_value = m.value
                normalized_value = Preprocessor.normalize_datetime(original_value)
                if normalized_value != original_value:
                    m.full_path.update(input_dict, normalized_value)

1	import logging	2✔
2	from datetime import datetime, timezone	2✔
3
4	from jsonpath_ng.parser import JsonPathParser	2✔
5
6	from src.model.SchemaConcepts.Schema_Concept import parse_datetime	2✔
7
8
9	class Preprocessor:	2✔
10	"""
11	Use / adapt / extend for final preprocessing steps before converting a dictionary into the according pydantic class instances
12	"""
13
14	parser = JsonPathParser()	2✔
15
16	unit_normalization = {	2✔
17	'deg': 'degree',
18	'degr': 'degree',
19	'°': 'degree',
20	'um': 'µm',
21	'Secs': 's',
22	'Mins': 'min',
23	'μs': 'us'
24	}
25
26
27	@staticmethod	2✔
28	def normalize_unit(input_value) -> str:	2✔
29	if input_value in Preprocessor.unit_normalization.keys():	×
30	return Preprocessor.unit_normalization[input_value]	×
31	return input_value	×
32
33	@staticmethod	2✔
34	def normalize_all_units(input_dict):	2✔
35	"""
36	Inplace normalization of all values in fields "unit"
37	:param input_dict: dictionary to replace units in
38	:return: None
39	"""
40	unit_field_names = ["unit", "coordinatesUnit"]	2✔
41	for field_name in unit_field_names:	2✔
42	unit_fields = Preprocessor.parser.parse("$.." + field_name)	2✔
43	unit_matches = [m for m in unit_fields.find(input_dict)]	2✔
44	for m in unit_matches:	2✔
45	if type(m.value) != str: continue #TODO: should this be possible?	2✔
46	original_value = m.value	2✔
47	if not Preprocessor.unit_normalization.get(original_value): continue	2✔
48
49	normalized_value = Preprocessor.unit_normalization[original_value]	2✔
50	if normalized_value != original_value:	2✔
51	m.full_path.update(input_dict, normalized_value)	2✔
52
53	@staticmethod	2✔
54	def normalize_datetime(input_value) -> str:	2✔
55	if type(input_value) == dict:	2✔
56	if not input_value.get("Date") and input_value.get("Time"): # Not possible to handle only Time	2✔
57	logging.warning("Encountered complex date field, but cannot interpret it")	×
58	return input_value	×
59	if input_value.get("Date") and not input_value.get("Time"): # Handle only Date	2✔
60	input_value["Time"] = "00:00:00"	2✔
61	logging.info("Input with date information but no time information found. Setting time to 00:00:00")	2✔
62	input_value = input_value.get("Date") + " " + input_value.get("Time")	2✔
63	output_value = parse_datetime(input_value)	2✔
64	if type(output_value) == datetime:	2✔
65	if output_value.tzinfo:	2✔
66	output_value = output_value.astimezone(timezone.utc) # datetime has timezone info, convert it to UTC	2✔
67	else:
68	output_value = output_value.replace(tzinfo=timezone.utc) # No timezone, assume it's already in UTC	2✔
69	return output_value.isoformat().replace("+00:00", "Z")	2✔
UNCOV 70	return input_value	×
71
72	@staticmethod	2✔
73	def normalize_all_datetimes(input_dict):	2✔
74	fields_for_normalization = ["creationTime", "startTime", "endTime"] #we could do it more generically but may want to limit it to specific fields	2✔
75
76	for f in fields_for_normalization:	2✔
77	date_fields = Preprocessor.parser.parse("$.." + f)	2✔
78	date_matches = [m for m in date_fields.find(input_dict)]	2✔
79	for m in date_matches:	2✔
80	original_value = m.value	2✔
81	normalized_value = Preprocessor.normalize_datetime(original_value)	2✔
82	if normalized_value != original_value:	2✔
83	m.full_path.update(input_dict, normalized_value)	2✔

kit-data-manager / tomo_mapper / 16725506371

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