Skip to content

Fix comma formatting #12

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 3 additions & 5 deletions kmm/functional_base.py
Original file line number Diff line number Diff line change
@@ -1,15 +1,13 @@
from pydantic import BaseModel, Extra
from pydantic import BaseModel, ConfigDict


class FunctionalBase(BaseModel):
class Config:
allow_mutation = False
extra = Extra.forbid
model_config = ConfigDict(frozen=True, extra="forbid")

def map(self, fn, *args, **kwargs):
return fn(self, *args, **kwargs)

def replace(self, **kwargs):
new_dict = self.dict()
new_dict = self.model_dump()
new_dict.update(**kwargs)
return type(self)(**new_dict)
4 changes: 2 additions & 2 deletions kmm/header/header.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
from pathlib import Path
from xml.etree import ElementTree

from pydantic import validate_arguments
from pydantic import validate_call

import kmm

Expand All @@ -12,7 +12,7 @@ class Header(kmm.FunctionalBase):
sync: int

@staticmethod
@validate_arguments
@validate_call
def from_path(path: Path, raise_on_malformed_data: bool = True):
"""
Loads header data from .hdr file.
Expand Down
77 changes: 71 additions & 6 deletions kmm/positions/positions.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
from pathlib import Path

import pandas as pd
from pydantic import validate_arguments
from pydantic import ConfigDict, validate_call

import kmm
from kmm.header.header import Header
Expand All @@ -10,11 +10,10 @@
class Positions(kmm.FunctionalBase):
dataframe: pd.DataFrame

class Config:
arbitrary_types_allowed = True
model_config = ConfigDict(arbitrary_types_allowed=True)

@staticmethod
@validate_arguments
@validate_call
def from_path(
path: Path,
raise_on_malformed_data: bool = True,
Expand All @@ -37,7 +36,59 @@ def from_path(
return Positions(dataframe=dataframe)

@staticmethod
@validate_arguments
@validate_call
def read_sync_adjust_from_header_path(
header_path: Path,
kmm2=True,
raise_on_malformed_data: bool = True,
replace_commas: bool = True,
):
"""
Convenience method to load positions from a header file, assumes a kmm file in the same directory.
If kmm2 is True, the method will load a kmm2 file, otherwise a kmm file.
"""
kmm_stem = (
header_path.stem.replace("owlsbtlpos", "").split("_2011T")[0] + "_2011T"
)
if kmm2:
kmm_path = header_path.parent / f"{kmm_stem}.kmm2"
else:
kmm_path = header_path.parent / f"{kmm_stem}.kmm"
return Positions.read_sync_adjust(
kmm_path,
header_path,
raise_on_malformed_data=raise_on_malformed_data,
replace_commas=replace_commas,
)

@staticmethod
@validate_call
def read_sync_adjust_from_measurement_name(
measurement_name: str,
input_dir: Path,
kmm2: bool = True,
raise_on_malformed_data: bool = True,
replace_commas: bool = True,
):
"""
Convenience method to load positions from a measurement name, assumes a kmm2 file and a header file in input_dir.
"""
timestamp = "_".join(measurement_name.split("_")[:-1])
part = measurement_name.split("_")[-1]
if kmm2:
kmm_path = input_dir / f"{timestamp}_2011T.kmm2"
else:
kmm_path = input_dir / f"{timestamp}_2011T.kmm"
header_path = input_dir / f"owlsbtlpos{timestamp}_2011T{part}.hdr"
return Positions.read_sync_adjust(
kmm_path,
header_path,
raise_on_malformed_data=raise_on_malformed_data,
replace_commas=replace_commas,
)

@staticmethod
@validate_call
def read_sync_adjust(
kmm_path: Path,
header_path: Path,
Expand All @@ -60,7 +111,7 @@ def read_sync_adjust(
.geodetic()
)

@validate_arguments
@validate_call
def sync_frame_index(
self,
header: Header,
Expand Down Expand Up @@ -97,3 +148,17 @@ def test_empty_kmm():
def test_empty_kmm2():
positions = Positions.from_path("tests/empty.kmm2")
assert len(positions.dataframe) == 0


def test_read_sync_adjust_from_header_path():
positions = Positions.read_sync_adjust_from_header_path(
"tests/owlsbtlpos20210819_165120_2011TA.hdr", kmm2=True
)
assert len(positions.dataframe) > 0


def test_read_sync_adjust_from_measurement_name():
positions = Positions.read_sync_adjust_from_measurement_name(
"20210819_165120_A", "tests"
)
assert len(positions.dataframe) > 0
4 changes: 2 additions & 2 deletions kmm/positions/read_kmm.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,10 +3,10 @@

import numpy as np
import pandas as pd
from pydantic import validate_arguments
from pydantic import validate_call


@validate_arguments
@validate_call
def read_kmm(path: Path, replace_commas: bool = True):
try:
if replace_commas:
Expand Down
4 changes: 2 additions & 2 deletions kmm/positions/read_kmm2.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@

import numpy as np
import pandas as pd
from pydantic import validate_arguments
from pydantic import validate_call

pattern = re.compile(r".+\[.+\]")
pattern2 = re.compile(r"CMAST")
Expand Down Expand Up @@ -46,7 +46,7 @@
)


@validate_arguments
@validate_call
def read_kmm2(
path: Path, raise_on_malformed_data: bool = True, replace_commas: bool = True
):
Expand Down
4 changes: 2 additions & 2 deletions kmm/positions/sync_frame_index.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,12 @@
import numpy as np
from pydantic import validate_arguments
from pydantic import validate_call

from kmm import CarDirection, PositionAdjustment
from kmm.header.header import Header
from kmm.positions.positions import Positions


@validate_arguments(config=dict(arbitrary_types_allowed=True))
@validate_call(config=dict(arbitrary_types_allowed=True))
def sync_frame_index(
positions: Positions,
header: Header,
Expand Down
Loading