Skip to content

opensampl.vendors.microchip.tp4100

MicrochipTP4100 clock Parser implementation

MicrochipTP4100Probe

Bases: BaseProbe

MicrochipTP4100 Probe Object

Source code in opensampl/vendors/microchip/tp4100.py 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
class MicrochipTP4100Probe(BaseProbe):
    """MicrochipTP4100 Probe Object"""

    vendor = VENDORS.MICROCHIP_TP4100
    MEASUREMENTS: ClassVar = {
        "time-error (ns)": METRICS.PHASE_OFFSET,
    }
    REFERENCES: ClassVar = {"GNSS": REF_TYPES.GNSS}

    class RandomDataConfig(BaseProbe.RandomDataConfig):
        """Model for storing random data generation configurations as provided by CLI or YAML"""

        # Time series parameters
        base_value: Optional[float] = Field(
            default_factory=lambda: random.uniform(-5e-7, 5e-7), description="random.uniform(-5e-7, 5e-7)"
        )
        noise_amplitude: Optional[float] = Field(
            default_factory=lambda: random.uniform(1e-8, 5e-8), description="random.uniform(1e-8, 5e-8)"
        )
        drift_rate: Optional[float] = Field(
            default_factory=lambda: random.uniform(-1e-10, 1e-10), description="random.uniform(-1e-10, 1e-10)"
        )

        metric_type: str = "time-error (ns)"
        reference_type: str = "GNSS"

    @classmethod
    def get_random_data_cli_options(cls) -> list:
        """Return vendor-specific random data generation options."""
        base_options = super().get_random_data_cli_options()
        vendor_options = [
            click.option(
                "--probe-id",
                type=str,
                help=(
                    "The probe_id you want the random data to show up under. "
                    "Randomly generated for each probe if left empty; incremented if multiple probes"
                ),
            ),
        ]
        return base_options + vendor_options

    def __init__(self, input_file: Union[str, Path]):
        """Initialize MicrochipTP4100 object given input_file and determines probe identity from file headers"""
        super().__init__(input_file=input_file)
        self.header = self.get_header()
        self.probe_key = ProbeKey(
            ip_address=self.header.get("host"), probe_id=self.header.get("probe_id", None) or "1-1"
        )

    def get_header(self) -> dict:
        """Retrieve the yaml formatted header information from the input file loaded into a dict"""
        header_lines = []
        with self.input_file.open() as f:
            for line in f:
                if line.startswith("#"):
                    header_lines.append(line[2:])
                else:
                    break

        header_str = "".join(header_lines)
        return {k.strip().lower(): v for k, v in yaml.safe_load(header_str).items()}

    @classmethod
    def filter_files(cls, files: list[Path]) -> list[Path]:
        """Filter the files found in input directory to only take .csv and .txt"""
        return [x for x in files if any(x.name.endswith(ext) for ext in (".csv", ".txt"))]

    def process_time_data(self) -> None:
        """
        Process time series data from the input file.

        Returns:
            pd.DataFrame: DataFrame with columns:
                - time (datetime64[ns]): timestamp for each measurement
                - value (float64): measured value at each timestamp

        """
        collection_method = self.header.get("method", "")
        try:
            df = pd.read_csv(
                self.input_file,
                delimiter=", " if collection_method == "download_file" else ",",
                comment="#",
                engine="python",
            )
        except pd.errors.EmptyDataError as e:
            raise ValueError(f"No data in {self.input_file}") from e

        if len(df) == 0:
            raise ValueError(f"No data in {self.input_file}")

        header_metric = self.header.get("metric").lower()  # We want a value error raised if it's not in there at all
        metric = self.MEASUREMENTS.get(header_metric, None)

        if metric is None:
            logger.warning(f"Metric type {header_metric} not configured for MicrochipTWST; skipping upload")
            return

        if len(df.columns) < 2:
            raise ValueError("Expected at at least 2 columns in the CSV")
        df.columns = ["time", "value", *df.columns[2:]]

        if "(ns)" in header_metric:
            df["value"] = df["value"].apply(lambda x: float(x) / 1e9)

        if collection_method == "download_file":
            df["time"] = pd.to_datetime(df["time"], format="%Y-%m-%d,%H:%M:%S", utc=True)

        header_ref = self.header.get("reference").upper()
        reference = self.REFERENCES.get(header_ref, None)
        if reference is None:
            logger.warning(
                f"Reference type {header_ref} not configured for MicrochipTWST. Setting reference as unknown."
            )
            reference = REF_TYPES.UNKNOWN

        self.send_data(data=df, metric=metric, reference_type=reference)

    def process_metadata(self) -> dict:
        """
        Process metadata from the input file.

        Returns:
            dict: Dictionary mapping table names to ORM objects

        """
        return {"additional_metadata": self.header, "model": "TP 4100"}

    @classmethod
    def generate_random_data(
        cls,
        config: RandomDataConfig,
        probe_key: ProbeKey,
    ) -> ProbeKey:
        """
        Generate random TP4100 test data and send it directly to the database.

        Args:
            probe_key: Probe key to use (generated if None)
            config: RandomDataConfig with parameters specifying how to generate data

        Returns:
            ProbeKey: The probe key used for the generated data

        """
        cls._setup_random_seed(config.seed)

        logger.info(f"Generating random TP4100 data for {probe_key}")

        # Generate metadata header similar to real TP4100 files
        metadata_header = {
            "title": f"Test TP4100 Performance Monitor {probe_key.probe_id}",
            "test_data": True,
            "random_generation_config": config.model_dump(),
        }

        # Generate and send metadata
        metadata = {"additional_metadata": metadata_header, "model": "TP 4100"}

        cls._send_metadata_to_db(probe_key, metadata)

        # Generate time series using base class helper (in nanoseconds, then convert)
        df = config.generate_time_series()

        # Determine metric and reference types
        metric = cls.MEASUREMENTS.get(config.metric_type.lower(), METRICS.PHASE_OFFSET)
        reference = cls.REFERENCES.get(config.reference_type.upper(), REF_TYPES.GNSS)

        # Send data to database
        cls.send_data(
            probe_key=probe_key,
            data=df,
            metric=metric,
            reference_type=reference,
        )

        logger.info(f"Successfully generated {config.duration_hours}h of random TP4100 data for {probe_key}")
        return probe_key

RandomDataConfig

Bases: RandomDataConfig

Model for storing random data generation configurations as provided by CLI or YAML

Source code in opensampl/vendors/microchip/tp4100.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
class RandomDataConfig(BaseProbe.RandomDataConfig):
    """Model for storing random data generation configurations as provided by CLI or YAML"""

    # Time series parameters
    base_value: Optional[float] = Field(
        default_factory=lambda: random.uniform(-5e-7, 5e-7), description="random.uniform(-5e-7, 5e-7)"
    )
    noise_amplitude: Optional[float] = Field(
        default_factory=lambda: random.uniform(1e-8, 5e-8), description="random.uniform(1e-8, 5e-8)"
    )
    drift_rate: Optional[float] = Field(
        default_factory=lambda: random.uniform(-1e-10, 1e-10), description="random.uniform(-1e-10, 1e-10)"
    )

    metric_type: str = "time-error (ns)"
    reference_type: str = "GNSS"

__init__(input_file)

Initialize MicrochipTP4100 object given input_file and determines probe identity from file headers

Source code in opensampl/vendors/microchip/tp4100.py 
61
62
63
64
65
66
67
def __init__(self, input_file: Union[str, Path]):
    """Initialize MicrochipTP4100 object given input_file and determines probe identity from file headers"""
    super().__init__(input_file=input_file)
    self.header = self.get_header()
    self.probe_key = ProbeKey(
        ip_address=self.header.get("host"), probe_id=self.header.get("probe_id", None) or "1-1"
    )

filter_files(files) classmethod

Filter the files found in input directory to only take .csv and .txt

Source code in opensampl/vendors/microchip/tp4100.py 
82
83
84
85
@classmethod
def filter_files(cls, files: list[Path]) -> list[Path]:
    """Filter the files found in input directory to only take .csv and .txt"""
    return [x for x in files if any(x.name.endswith(ext) for ext in (".csv", ".txt"))]

generate_random_data(config, probe_key) classmethod

Generate random TP4100 test data and send it directly to the database.

Parameters:

Name Type Description Default
probe_key ProbeKey

Probe key to use (generated if None)

 required
config RandomDataConfig

RandomDataConfig with parameters specifying how to generate data

 required

Returns:

Name Type Description
ProbeKey ProbeKey

The probe key used for the generated data
Source code in opensampl/vendors/microchip/tp4100.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
@classmethod
def generate_random_data(
    cls,
    config: RandomDataConfig,
    probe_key: ProbeKey,
) -> ProbeKey:
    """
    Generate random TP4100 test data and send it directly to the database.

    Args:
        probe_key: Probe key to use (generated if None)
        config: RandomDataConfig with parameters specifying how to generate data

    Returns:
        ProbeKey: The probe key used for the generated data

    """
    cls._setup_random_seed(config.seed)

    logger.info(f"Generating random TP4100 data for {probe_key}")

    # Generate metadata header similar to real TP4100 files
    metadata_header = {
        "title": f"Test TP4100 Performance Monitor {probe_key.probe_id}",
        "test_data": True,
        "random_generation_config": config.model_dump(),
    }

    # Generate and send metadata
    metadata = {"additional_metadata": metadata_header, "model": "TP 4100"}

    cls._send_metadata_to_db(probe_key, metadata)

    # Generate time series using base class helper (in nanoseconds, then convert)
    df = config.generate_time_series()

    # Determine metric and reference types
    metric = cls.MEASUREMENTS.get(config.metric_type.lower(), METRICS.PHASE_OFFSET)
    reference = cls.REFERENCES.get(config.reference_type.upper(), REF_TYPES.GNSS)

    # Send data to database
    cls.send_data(
        probe_key=probe_key,
        data=df,
        metric=metric,
        reference_type=reference,
    )

    logger.info(f"Successfully generated {config.duration_hours}h of random TP4100 data for {probe_key}")
    return probe_key

get_header()

Retrieve the yaml formatted header information from the input file loaded into a dict

Source code in opensampl/vendors/microchip/tp4100.py 
69
70
71
72
73
74
75
76
77
78
79
80
def get_header(self) -> dict:
    """Retrieve the yaml formatted header information from the input file loaded into a dict"""
    header_lines = []
    with self.input_file.open() as f:
        for line in f:
            if line.startswith("#"):
                header_lines.append(line[2:])
            else:
                break

    header_str = "".join(header_lines)
    return {k.strip().lower(): v for k, v in yaml.safe_load(header_str).items()}

get_random_data_cli_options() classmethod

Return vendor-specific random data generation options.

Source code in opensampl/vendors/microchip/tp4100.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
@classmethod
def get_random_data_cli_options(cls) -> list:
    """Return vendor-specific random data generation options."""
    base_options = super().get_random_data_cli_options()
    vendor_options = [
        click.option(
            "--probe-id",
            type=str,
            help=(
                "The probe_id you want the random data to show up under. "
                "Randomly generated for each probe if left empty; incremented if multiple probes"
            ),
        ),
    ]
    return base_options + vendor_options

process_metadata()

Process metadata from the input file.

Returns:

Name Type Description
dict dict

Dictionary mapping table names to ORM objects
Source code in opensampl/vendors/microchip/tp4100.py 
138
139
140
141
142
143
144
145
146
def process_metadata(self) -> dict:
    """
    Process metadata from the input file.

    Returns:
        dict: Dictionary mapping table names to ORM objects

    """
    return {"additional_metadata": self.header, "model": "TP 4100"}

process_time_data()

Process time series data from the input file.

Returns:

Type Description
None

pd.DataFrame: DataFrame with columns: - time (datetime64[ns]): timestamp for each measurement - value (float64): measured value at each timestamp
Source code in opensampl/vendors/microchip/tp4100.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def process_time_data(self) -> None:
    """
    Process time series data from the input file.

    Returns:
        pd.DataFrame: DataFrame with columns:
            - time (datetime64[ns]): timestamp for each measurement
            - value (float64): measured value at each timestamp

    """
    collection_method = self.header.get("method", "")
    try:
        df = pd.read_csv(
            self.input_file,
            delimiter=", " if collection_method == "download_file" else ",",
            comment="#",
            engine="python",
        )
    except pd.errors.EmptyDataError as e:
        raise ValueError(f"No data in {self.input_file}") from e

    if len(df) == 0:
        raise ValueError(f"No data in {self.input_file}")

    header_metric = self.header.get("metric").lower()  # We want a value error raised if it's not in there at all
    metric = self.MEASUREMENTS.get(header_metric, None)

    if metric is None:
        logger.warning(f"Metric type {header_metric} not configured for MicrochipTWST; skipping upload")
        return

    if len(df.columns) < 2:
        raise ValueError("Expected at at least 2 columns in the CSV")
    df.columns = ["time", "value", *df.columns[2:]]

    if "(ns)" in header_metric:
        df["value"] = df["value"].apply(lambda x: float(x) / 1e9)

    if collection_method == "download_file":
        df["time"] = pd.to_datetime(df["time"], format="%Y-%m-%d,%H:%M:%S", utc=True)

    header_ref = self.header.get("reference").upper()
    reference = self.REFERENCES.get(header_ref, None)
    if reference is None:
        logger.warning(
            f"Reference type {header_ref} not configured for MicrochipTWST. Setting reference as unknown."
        )
        reference = REF_TYPES.UNKNOWN

    self.send_data(data=df, metric=metric, reference_type=reference)