Skip to content
Snippets Groups Projects
Commit 8a736568 authored by Wernle, Alexandra Nicole's avatar Wernle, Alexandra Nicole
Browse files

Considering the possibility of spikes as a numpy array

parent cff6e572
No related branches found
No related tags found
1 merge request!292New flag metadata class
from typing import Dict, Union, List
from datetime import timedelta
import numpy as np
from obspy import UTCDateTime
from pydantic import BaseModel, Field, validator
from enum import Enum
......@@ -35,7 +36,6 @@ class Flag(BaseModel):
priority=1,
data_valid=False,
metadata= Disturbance{
"channels": "H,E,Z",
"description": "Spike in magnetic field strength",
"field_work": false,
"corrected": false,
......@@ -47,19 +47,16 @@ class Flag(BaseModel):
```
"""
channels: Union[ChannelType, List[ChannelType]]
description: str = Field(..., description="Description of the flag")
field_work: bool = Field(..., description="Flag signaling field work")
corrected: bool = Field(
...,
description="whether or not corrective action has been taken for spikes, noise, etc.",
)
corrected: int = Field(..., description="Corrected ID for processing stage")
# Should there be an attribute for data quality or is data_valid above enough?
class DisturbanceType(str, Enum):
SPIKE = "SPIKE"
SPIKES = "SPIKES"
OFFSET = "OFFSET"
NOISE = "NOISE"
DISTURBANCES = "DISTURBANCES"
class Disturbance(Flag):
......@@ -68,9 +65,9 @@ class Disturbance(Flag):
Artificial disturbances consist of the following types:
SPIKE = Single data point that are outliers in the timeseries.
SPIKES = Single data points that are outliers in the timeseries.
OFFSET = A relatively constant shift or deviation in the baseline magnetic field.
NOISE = A catch-all for any unwanted variations, may include multiple spikes, offsets and/or gaps.
DISTURBANCES = A catch-all for a continuous period of unwanted variations, may include multiple spikes, offsets and/or gaps.
Attributes
----------
......@@ -78,14 +75,17 @@ class Disturbance(Flag):
The type of disturbance(s).
source: str
Source of the disturbance if known or suspected.
magnitude: float
Magnitude of the disturbance in nt if it is a singular spike or offset value.
amplitude: float
Amplitude of the disturbance in nt if it is an offset value or singular spike.
spikes: np.ndarray
NumPy array of timestamps as UTCDateTime. Can be a single spike or many spikes.
"""
disturbance_type: DisturbanceType
source: str = None
magnitude: float
amplitude: float
spikes: np.ndarray
class Gap(Flag):
......@@ -106,42 +106,72 @@ class Gap(Flag):
handling: str = None
class EventType(str, Enum):
GEOMAGNETIC_STORM = "GEOMAGNETIC_STORM"
EARTHQUAKE = "EARTHQUAKE"
class Event(Flag):
"""
This class is used to flag an event of interest such as a geomagnetic storm or earthquake.
Attributes
----------
geomagnetic_storm: bool
Geomagnetic storm period.
event_type : EventType
The type of event.
scale : str
Geomagnetic storm scale denoted by G followed by a number 1 through 5.
Kp : int
Planetary K-index denoted by a number 1 through 10
url : str
A url related to the event. Could be NOAA SWPC or USGS Earthquakes page.
"""
geomagnetic_storm: bool = None # if true, ignore spike detection filter
# hold intensity, K-index, more?
event_type: EventType
scale: str = None
Kp: int = None
url: str = None
# More example usage:
spike_data = {
"channels": ["F"],
"description": "Spike description",
"field_work": True,
"corrected": False,
"disturbance_type": DisturbanceType.SPIKE,
"source": "Lightning",
"magnitude": 5.0,
timestamps_array = np.array(
[
UTCDateTime("2023-11-16T12:00:0"),
UTCDateTime("2023-11-16T12:01:10"),
UTCDateTime("2023-11-16T12:02:30"),
]
)
spikes_data = {
"starttime": "2023-11-16 12:00:00",
"endtime": "2023-11-16 12:02:30",
"description": "Spikes description",
"field_work": False,
"corrected": 32265,
"disturbance_type": DisturbanceType.SPIKES,
"source": "processing",
"spikes": timestamps_array,
}
offset_data = {
"channels": ["X", "Y", "Z"],
"description": "Offset description",
"field_work": False,
"corrected": True,
"corrected": 47999,
"disturbance_type": DisturbanceType.OFFSET,
"source": "Bin change",
"magnitude": 10.0,
"amplitude": 10.0,
}
geomagnetic_storm_data = {
"description": "Geomagnetic storm",
"field_work": False,
"corrected": 36999,
"event_type": EventType.GEOMAGNETIC_STORM,
"scale": "G3",
"Kp": 7,
"url": "https://www.swpc.noaa.gov/products/planetary-k-index",
}
spike_instance = Disturbance(**spike_data)
spike_instance = Disturbance(**spikes_data)
offset_instance = Disturbance(**offset_data)
print(spike_instance.dict())
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment