16725814017

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

Build # 16725814017

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%)

0.86 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	1✔
2	from datetime import datetime, timezone	1✔
3
4	from jsonpath_ng.parser import JsonPathParser	1✔
5
6	from src.model.SchemaConcepts.Schema_Concept import parse_datetime	1✔
7
8
9	class Preprocessor:	1✔
10	"""
11	Use / adapt / extend for final preprocessing steps before converting a dictionary into the according pydantic class instances
12	"""
13
14	parser = JsonPathParser()	1✔
15
16	unit_normalization = {	1✔
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	1✔
28	def normalize_unit(input_value) -> str:	1✔
29	if input_value in Preprocessor.unit_normalization.keys():	×
30	return Preprocessor.unit_normalization[input_value]	×
31	return input_value	×
32
33	@staticmethod	1✔
34	def normalize_all_units(input_dict):	1✔
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"]	1✔
41	for field_name in unit_field_names:	1✔
42	unit_fields = Preprocessor.parser.parse("$.." + field_name)	1✔
43	unit_matches = [m for m in unit_fields.find(input_dict)]	1✔
44	for m in unit_matches:	1✔
45	if type(m.value) != str: continue #TODO: should this be possible?	1✔
46	original_value = m.value	1✔
47	if not Preprocessor.unit_normalization.get(original_value): continue	1✔
48
49	normalized_value = Preprocessor.unit_normalization[original_value]	1✔
50	if normalized_value != original_value:	1✔
51	m.full_path.update(input_dict, normalized_value)	1✔
52
53	@staticmethod	1✔
54	def normalize_datetime(input_value) -> str:	1✔
55	if type(input_value) == dict:	1✔
56	if not input_value.get("Date") and input_value.get("Time"): # Not possible to handle only Time	1✔
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	1✔
60	input_value["Time"] = "00:00:00"	1✔
61	logging.info("Input with date information but no time information found. Setting time to 00:00:00")	1✔
62	input_value = input_value.get("Date") + " " + input_value.get("Time")	1✔
63	output_value = parse_datetime(input_value)	1✔
64	if type(output_value) == datetime:	1✔
65	if output_value.tzinfo:	1✔
66	output_value = output_value.astimezone(timezone.utc) # datetime has timezone info, convert it to UTC	1✔
67	else:
68	output_value = output_value.replace(tzinfo=timezone.utc) # No timezone, assume it's already in UTC	1✔
69	return output_value.isoformat().replace("+00:00", "Z")	1✔
UNCOV 70	return input_value	×
71
72	@staticmethod	1✔
73	def normalize_all_datetimes(input_dict):	1✔
74	fields_for_normalization = ["creationTime", "startTime", "endTime"] #we could do it more generically but may want to limit it to specific fields	1✔
75
76	for f in fields_for_normalization:	1✔
77	date_fields = Preprocessor.parser.parse("$.." + f)	1✔
78	date_matches = [m for m in date_fields.find(input_dict)]	1✔
79	for m in date_matches:	1✔
80	original_value = m.value	1✔
81	normalized_value = Preprocessor.normalize_datetime(original_value)	1✔
82	if normalized_value != original_value:	1✔
83	m.full_path.update(input_dict, normalized_value)	1✔

kit-data-manager / tomo_mapper / 16725814017

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