diff --git a/README.md b/README.md index 4635648..2f4d4d5 100644 --- a/README.md +++ b/README.md @@ -100,7 +100,7 @@ import datetime as dt import numpy as np import sigmf from sigmf import SigMFFile -from sigmf.utils import get_data_type_str +from sigmf.utils import get_data_type_str, get_sigmf_iso8601_datetime_now # suppose we have an complex timeseries signal data = np.zeros(1024, dtype=np.complex64) @@ -122,7 +122,7 @@ meta = SigMFFile( # create a capture key at time index 0 meta.add_capture(0, metadata={ SigMFFile.FREQUENCY_KEY: 915000000, - SigMFFile.DATETIME_KEY: dt.datetime.utcnow().isoformat()+'Z', + SigMFFile.DATETIME_KEY: get_sigmf_iso8601_datetime_now(), }) # add an annotation at sample 100 with length 200 & 10 KHz width diff --git a/sigmf/apps/convert_wav.py b/sigmf/apps/convert_wav.py index 9151d2f..c694750 100755 --- a/sigmf/apps/convert_wav.py +++ b/sigmf/apps/convert_wav.py @@ -7,24 +7,29 @@ """converter for wav containers""" import argparse -import datetime import getpass import logging import os import pathlib import tempfile +from typing import Optional from scipy.io import wavfile from .. import SigMFFile, __specification__ from .. import __version__ as toolversion from .. import archive -from ..utils import get_data_type_str +from ..utils import get_data_type_str, get_sigmf_iso8601_datetime_now log = logging.getLogger() -def convert_wav(input_wav_filename, archive_filename=None, start_datetime=None, author=None): +def convert_wav( + input_wav_filename: str, + archive_filename: Optional[str], + start_datetime: Optional[str] = None, + author: Optional[str] = None, +): """ read a .wav and write a .sigmf archive """ @@ -43,12 +48,12 @@ def convert_wav(input_wav_filename, archive_filename=None, start_datetime=None, } if start_datetime is None: - mtime = datetime.datetime.fromtimestamp(input_path.stat().st_mtime) - start_datetime = mtime.isoformat() + "Z" + start_datetime = get_sigmf_iso8601_datetime_now() - capture_info = {SigMFFile.START_INDEX_KEY: 0} - if start_datetime is not None: - capture_info[SigMFFile.DATETIME_KEY] = start_datetime + capture_info = { + SigMFFile.START_INDEX_KEY: 0, + SigMFFile.DATETIME_KEY: start_datetime, + } tmpdir = tempfile.mkdtemp() sigmf_data_filename = input_stem + archive.SIGMF_DATASET_EXT @@ -71,8 +76,8 @@ def main(): parser = argparse.ArgumentParser(description="Convert .wav to .sigmf container.") parser.add_argument("input", type=str, help="Wavfile path") parser.add_argument("--author", type=str, default=None, help=f"set {SigMFFile.AUTHOR_KEY} metadata") - parser.add_argument('-v', '--verbose', action='count', default=0) - parser.add_argument('--version', action='version', version=f'%(prog)s v{toolversion}') + parser.add_argument("-v", "--verbose", action="count", default=0) + parser.add_argument("--version", action="version", version=f"%(prog)s v{toolversion}") args = parser.parse_args() level_lut = { diff --git a/sigmf/utils.py b/sigmf/utils.py index 2e61b42..935d64e 100644 --- a/sigmf/utils.py +++ b/sigmf/utils.py @@ -9,7 +9,7 @@ import re import sys from copy import deepcopy -from datetime import datetime +from datetime import datetime, timezone import numpy as np @@ -19,31 +19,34 @@ def get_sigmf_iso8601_datetime_now() -> str: - """Get current UTC time as iso8601 string""" - return datetime.isoformat(datetime.utcnow()) + "Z" + """Get current UTC time as iso8601 string.""" + return datetime.now(timezone.utc).strftime(SIGMF_DATETIME_ISO8601_FMT) -def parse_iso8601_datetime(datestr: str) -> datetime: +def parse_iso8601_datetime(string: str) -> datetime: """ - Parse an iso8601 string as a datetime + Parse an iso8601 string as a datetime struct. + Input string (indicated by final Z) is in UTC tz. Example ------- >>> parse_iso8601_datetime("1955-11-05T06:15:00Z") - datetime.datetime(1955, 11, 5, 6, 15) + datetime.datetime(1955, 11, 5, 6, 15, tzinfo=datetime.timezone.utc) """ - # provided string exceeds max precision -> truncate to µs - match = re.match(r"^(?P
.*)(?P\.[0-9]{7,})Z$", datestr) + match = re.match(r"^(?P
.*)(?P\.[0-9]{7,})Z$", string) if match: - md = match.groupdict() - length = min(7, len(md["frac"])) - datestr = ''.join([md["dt"], md["frac"][:length], "Z"]) - - try: - timestamp = datetime.strptime(datestr, '%Y-%m-%dT%H:%M:%S.%fZ') - except ValueError: - timestamp = datetime.strptime(datestr, '%Y-%m-%dT%H:%M:%SZ') - return timestamp + # string exceeds max precision allowed by strptime -> truncate to µs + groups = match.groupdict() + length = min(7, len(groups["frac"])) + string = "".join([groups["dt"], groups["frac"][:length], "Z"]) + + if "." in string: + # parse float seconds + format_str = SIGMF_DATETIME_ISO8601_FMT + else: + # parse whole seconds + format_str = SIGMF_DATETIME_ISO8601_FMT.replace(".%f", "") + return datetime.strptime(string, format_str).replace(tzinfo=timezone.utc) def dict_merge(a_dict: dict, b_dict: dict) -> dict: @@ -83,11 +86,10 @@ def get_endian_str(ray: np.ndarray) -> str: if atype.byteorder == "<": return "_le" - elif atype.byteorder == ">": + if atype.byteorder == ">": return "_be" - else: - # endianness is then either '=' (native) or '|' (doesn't matter) - return "_le" if sys.byteorder == "little" else "_be" + # endianness is then either '=' (native) or '|' (doesn't matter) + return "_le" if sys.byteorder == "little" else "_be" def get_data_type_str(ray: np.ndarray) -> str: diff --git a/tests/test_utils.py b/tests/test_utils.py index e3599dc..e4cb9a5 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -6,22 +6,30 @@ """Tests for Utilities""" -from datetime import datetime +from datetime import datetime, timezone import pytest from sigmf import utils -@pytest.mark.parametrize("ts, expected", [ - ("1955-07-04T05:15:00Z", datetime(year=1955, month=7, day=4, hour=5, minute=15, second=00, microsecond=0)), - ("2956-08-05T06:15:12Z", datetime(year=2956, month=8, day=5, hour=6, minute=15, second=12, microsecond=0)), - ("3957-09-06T07:15:12.345Z", datetime(year=3957, month=9, day=6, hour=7, minute=15, second=12, microsecond=345000)), - ("4958-10-07T08:15:12.0345Z", datetime(year=4958, month=10, day=7, hour=8, minute=15, second=12, microsecond=34500)), - ("5959-11-08T09:15:12.000000Z", datetime(year=5959, month=11, day=8, hour=9, minute=15, second=12, microsecond=0)), - ("6960-12-09T10:15:12.123456789123Z", datetime(year=6960, month=12, day=9, hour=10, minute=15, second=12, microsecond=123456)), +@pytest.mark.parametrize("time_str, expected", [ + ("1955-07-04T05:15:00Z", datetime(year=1955, month=7, day=4, hour=5, minute=15, second=00, microsecond=0, tzinfo=timezone.utc)), + ("2956-08-05T06:15:12Z", datetime(year=2956, month=8, day=5, hour=6, minute=15, second=12, microsecond=0, tzinfo=timezone.utc)), + ("3957-09-06T07:15:12.345Z", datetime(year=3957, month=9, day=6, hour=7, minute=15, second=12, microsecond=345000, tzinfo=timezone.utc)), + ("4958-10-07T08:15:12.0345Z", datetime(year=4958, month=10, day=7, hour=8, minute=15, second=12, microsecond=34500, tzinfo=timezone.utc)), + ("5959-11-08T09:15:12.000000Z", datetime(year=5959, month=11, day=8, hour=9, minute=15, second=12, microsecond=0, tzinfo=timezone.utc)), + ("6960-12-09T10:15:12.123456789123Z", datetime(year=6960, month=12, day=9, hour=10, minute=15, second=12, microsecond=123456, tzinfo=timezone.utc)), ]) -def test_parse_simple_iso8601(ts, expected): - dt = utils.parse_iso8601_datetime(ts) - assert dt == expected +def test_parse_simple_iso8601(time_str: str, expected: datetime) -> None: + """Ensure various times are represented as expected""" + date_struct = utils.parse_iso8601_datetime(time_str) + assert date_struct == expected + + +def test_roundtrip_datetime() -> None: + """New string -> struct -> string is ok""" + now_str = utils.get_sigmf_iso8601_datetime_now() + now_struct = utils.parse_iso8601_datetime(now_str) + assert now_str == now_struct.strftime(utils.SIGMF_DATETIME_ISO8601_FMT)