Coverage for C: \ Users \ peaco \ OneDrive \ Documents \ GitHub \ mt_metadata \ mt_metadata \ timeseries \ filtered.py: 100%
23 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-10 00:11 -0800
1# =====================================================
2# Imports
3# =====================================================
4from typing import Annotated
6from pydantic import Field, field_validator, ValidationInfo
8from mt_metadata.base import MetadataBase
9from mt_metadata.common.comment import Comment
12# =====================================================
14# this will be a better way of keeping track of filter names and
15# if they have been applied or not. This can also include the stage
16# number of the filter and could be extended to other attributes.
17# TODO: figure out setting a applied and name.
20class AppliedFilter(MetadataBase):
21 name: Annotated[
22 str,
23 Field(
24 default=None,
25 description="Name of the filter.",
26 json_schema_extra={
27 "units": None,
28 "required": True,
29 "examples": ["low pass"],
30 },
31 ),
32 ]
34 applied: Annotated[
35 bool,
36 Field(
37 default=True,
38 description="Whether the filter has been applied.",
39 json_schema_extra={
40 "units": None,
41 "required": True,
42 "examples": ["True"],
43 },
44 ),
45 ]
47 stage: Annotated[
48 int | None,
49 Field(
50 default=None,
51 description="Stage of the filter in the processing chain.",
52 json_schema_extra={
53 "units": None,
54 "required": False,
55 "examples": [1],
56 },
57 ),
58 ]
60 comments: Annotated[
61 Comment,
62 Field(
63 default_factory=lambda: Comment(), # type: ignore
64 description="Any comments on filters.",
65 alias=None,
66 json_schema_extra={
67 "units": None,
68 "required": False,
69 "examples": ["low pass is not calibrated"],
70 },
71 ),
72 ]
74 @field_validator("comments", mode="before")
75 @classmethod
76 def validate_comments(cls, value, info: ValidationInfo) -> Comment:
77 """
78 Validate that the value is a valid comment.
79 """
80 if isinstance(value, str):
81 return Comment(value=value)
82 return value
84 @field_validator("name", mode="before")
85 @classmethod
86 def validate_name(cls, value, info: ValidationInfo) -> str:
87 """
88 Validate that the name is not empty or whitespace only. Also, replace
89 '/' with ' per ' and convert to lower case for consistency.
90 """
91 if isinstance(value, str):
92 if not value or value.isspace():
93 raise ValueError("Filter name cannot be empty or whitespace only")
94 value = value.replace("/", " per ").lower()
95 return value
98# class Filter(MetadataBase):
99# _objects_included: dict = PrivateAttr(
100# default_factory=lambda: {"applied_filter": AppliedFilter}
101# )
103# filter_list: Annotated[
104# list[AppliedFilter],
105# Field(
106# default_factory=list,
107# description="List of AppliedFilter() objects.",
108# examples=["[AppliedFilter(name='filter_name', applied=True, stage=1)]"],
109# alias=None,
110# json_schema_extra={
111# "units": None,
112# "required": False,
113# },
114# ),
115# ]
117# applied: Annotated[
118# list[bool],
119# Field(
120# default_factory=list,
121# description="List of booleans indicating if the filter has been applied.",
122# examples=[True, False],
123# alias=None,
124# deprecated=deprecated(
125# "'applied' will be deprecated in the future use an AppliedFilter object."
126# ),
127# json_schema_extra={
128# "units": None,
129# "required": False,
130# },
131# ),
132# ]
134# name: Annotated[
135# list[str],
136# Field(
137# default_factory=list,
138# description="List of filter names.",
139# examples=["low pass", "high pass"],
140# alias=None,
141# deprecated=deprecated(
142# "'name' will be deprecated in the future use an AppliedFilter object."
143# ),
144# json_schema_extra={
145# "units": None,
146# "required": False,
147# },
148# ),
149# ]
151# comments: Annotated[
152# Comment,
153# Field(
154# default_factory=Comment,
155# description="Any comments on filters.",
156# examples=["low pass is not calibrated"],
157# alias=None,
158# json_schema_extra={
159# "units": None,
160# "required": False,
161# },
162# ),
163# ]
165# @field_validator("comments", mode="before")
166# @classmethod
167# def validate_comments(cls, value, info: ValidationInfo) -> Comment:
168# """
169# Validate that the value is a valid comment.
170# """
171# if isinstance(value, str):
172# return Comment(value=value)
173# return value
175# @model_validator(mode="after")
176# def validate_applied_and_names(self) -> Self:
177# """
178# Validate the applied_list to ensure it contains only AppliedFilter objects.
179# """
180# if self.name != [] or self.applied != []:
181# logger.warning(
182# "'applied' and 'name' will be deprecated in the future append an "
183# "AppliedFilter(name='name', applied=True) to 'filter_list'."
184# )
186# if len(self.name) != len(self.applied):
187# diff = len(self.name) - len(self.applied)
188# if diff > 0:
189# self.applied.extend([True] * diff)
190# else:
191# self.name.extend(["unknown"] * abs(diff))
193# if len(self.name) != len(self.filter_list):
194# for name, applied, index in zip(
195# self.name, self.applied, range(len(self.name))
196# ):
197# self.filter_list.append(
198# AppliedFilter(name=name, applied=applied, stage=index + 1)
199# )
201# return self
203# def to_dict(
204# self, single: bool = False, nested: bool = False, required: bool = True
205# ) -> dict:
206# """
207# Convert the object to a dictionary. To be compliant with older versions
208# of the metadata, use name and applied as lists
210# Parameters
211# ----------
212# single : bool, optional
213# Whether to return a single dictionary or a list of dictionaries,
214# by default False.
216# Returns
217# -------
218# dict
219# Dictionary representation of the object.
220# """
221# d = OrderedDict()
222# d["name"] = self.name
223# d["applied"] = self.applied
224# d["comments"] = self.comments.to_dict(single=single, nested=nested)
225# return d
227# def add_filter(
228# self,
229# applied_filter: AppliedFilter = None,
230# name: str = None,
231# applied: bool = True,
232# stage: int | None = None,
233# ) -> None:
234# """
235# Add a filter to the filter list.
237# Parameters
238# ----------
239# name : str
240# Name of the filter.
241# applied : bool, optional
242# Whether the filter has been applied, by default True.
243# stage : int | None, optional
244# Stage of the filter in the processing chain, by default None.
245# """
246# if applied_filter is not None:
247# if not isinstance(applied_filter, AppliedFilter):
248# raise TypeError("applied_filter must be an instance of AppliedFilter")
249# if applied_filter.stage is None:
250# applied_filter.stage = len(self.filter_list) + 1
251# self.filter_list.append(applied_filter)
252# else:
253# if name is None:
254# raise ValueError("name must be provided if applied_filter is None")
255# if not isinstance(name, str):
256# raise TypeError("name must be a string")
257# if stage is None:
258# stage = len(self.filter_list) + 1
259# self.filter_list.append(
260# AppliedFilter(name=name, applied=applied, stage=stage)
261# )
263# def remove_filter(self, name: str, reset_stages: bool = True) -> None:
264# """
265# Remove a filter from the filter list.
267# Parameters
268# ----------
269# name : str
270# Name of the filter to remove.
271# reset_stages : bool, optional
272# Whether to reset the stages of the remaining filters, by default True.
273# """
275# new_list = []
276# for f in self.filter_list:
277# if f.name == name:
278# continue
279# if reset_stages:
280# f.stage = len(new_list) + 1
281# new_list.append(f)
282# self.filter_list = new_list