jepler/wwvbpy
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

switch to ruff as linter

Browse source
This commit is contained in:
Jeff Epler 2023-12-02 11:15:45 -06:00
parent 93e4598da2
commit 323a75a99a
No known key found for this signature in database
GPG key ID: D5BF15AB975AB4DE
15 changed files with 179 additions and 233 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

22
.pre-commit-config.yaml
View file

@ -6,10 +6,6 @@ default_language_version:
python: python3
repos:
- repo: https://github.com/psf/black
rev: 23.11.0
hooks:
- id: black
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v4.5.0
hooks:
@ -22,14 +18,12 @@ repos:
rev: v2.1.0
hooks:
- id: reuse
- repo: https://github.com/pycqa/pylint
rev: v3.0.1
- repo: https://github.com/astral-sh/ruff-pre-commit
# Ruff version.
rev: v0.1.6
hooks:
- id: pylint
additional_dependencies: ["setuptools>=68", beautifulsoup4, requests, adafruit-circuitpython-datetime, click, python-dateutil, leapseconddata]
- repo: https://github.com/pycqa/isort
rev: 5.12.0
hooks:
- id: isort
name: isort (python)
args: ['--profile', 'black']
# Run the linter.
- id: ruff
args: [ --fix ]
# Run the formatter.
- id: ruff-format

5
pyproject.toml
View file

@ -18,3 +18,8 @@ requires = [
build-backend = "setuptools.build_meta"
[tool.setuptools_scm]
write_to = "src/wwvb/__version__.py"
[tool.ruff.lint]
select = ["E", "F"]
ignore = ["E741"]
[tool.ruff]
line-length = 120

8
src/uwwvb.py
View file

@ -16,9 +16,7 @@ always_mark = set((0, 9, 19, 29, 39, 49, 59))
always_zero = set((4, 10, 11, 14, 20, 21, 34, 35, 44, 54))
bcd_weights = (1, 2, 4, 8, 10, 20, 40, 80, 100, 200, 400, 800)
WWVBMinute = namedtuple(
"WWVBMinute", ["year", "days", "hour", "minute", "dst", "ut1", "ls", "ly"]
)
WWVBMinute = namedtuple("WWVBMinute", ["year", "days", "hour", "minute", "dst", "ut1", "ls", "ly"])
class WWVBDecoder:
@ -30,7 +28,9 @@ class WWVBDecoder:
self.state = 1
def update(self, value: int) -> list[int] | None:
"""Update the _state machine when a new symbol is received. If a possible complete _minute is received, return it; otherwise, return None"""
"""Update the _state machine when a new symbol is received.
If a possible complete _minute is received, return it; otherwise, return None"""
result = None
if self.state == 1:
self.minute = []

64
src/wwvb/__init__.py
View file

@ -49,9 +49,7 @@ def _maybe_warn_update(dt: datetime.date) -> None:
# prospective available now.
today = datetime.date.today()
if _date(dt) < today + datetime.timedelta(days=330):
warnings.warn(
"Note: Running `updateiers` may provide better DUT1 and LS information"
)
warnings.warn("Note: Running `updateiers` may provide better DUT1 and LS information")
def get_dut1(dt: DateOrDatetime, *, warn_outdated: bool = True) -> float:
@ -110,9 +108,7 @@ def is_dst_change_day(t: datetime.date, tz: datetime.tzinfo = Mountain) -> bool:
return isdst(t, tz) != isdst(t + datetime.timedelta(1), tz)
def get_dst_change_hour(
t: DateOrDatetime, tz: datetime.tzinfo = Mountain
) -> Optional[int]:
def get_dst_change_hour(t: DateOrDatetime, tz: datetime.tzinfo = Mountain) -> Optional[int]:
"""Return the hour when DST changes"""
lt0 = datetime.datetime(t.year, t.month, t.day, hour=0, tzinfo=tz)
dst0 = lt0.dst()
@ -291,7 +287,9 @@ def extract_bit(v: int, p: int) -> bool:
def hamming_parity(value: int) -> int:
"""Compute the "hamming parity" of a 26-bit number, such as the minute-of-century [See Enhanced WWVB Broadcast Format 4.3]"""
"""Compute the "hamming parity" of a 26-bit number, such as the minute-of-century
For more details, see Enhanced WWVB Broadcast Format 4.3"""
parity = 0
for i in range(4, -1, -1):
bit = 0
@ -324,7 +322,9 @@ _WWVBMinute = collections.namedtuple("_WWVBMinute", "year days hour min dst ut1
class WWVBMinute(_WWVBMinute):
"""Uniquely identifies a minute of time in the WWVB system. To use ut1 and ls information from IERS, create a WWVBMinuteIERS value instead."""
"""Uniquely identifies a minute of time in the WWVB system.
To use ut1 and ls information from IERS, create a WWVBMinuteIERS value instead."""
year: int
hour: int
@ -391,7 +391,11 @@ class WWVBMinute(_WWVBMinute):
def __str__(self) -> str:
"""Implement str()"""
return f"year={self.year:4d} days={self.days:03d} hour={self.hour:02d} min={self.min:02d} dst={self.dst} ut1={self.ut1} ly={int(self.ly)} ls={int(self.ls)}"
return (
f"year={self.year:4d} days={self.days:03d} hour={self.hour:02d} "
f"min={self.min:02d} dst={self.dst} ut1={self.ut1} ly={int(self.ly)} "
f"ls={int(self.ls)}"
)
def as_datetime_utc(self) -> datetime.datetime:
"""Convert to a UTC datetime"""
@ -401,9 +405,7 @@ class WWVBMinute(_WWVBMinute):
as_datetime = as_datetime_utc
def as_datetime_local(
self, standard_time_offset: int = 7 * 3600, dst_observed: bool = True
) -> datetime.datetime:
def as_datetime_local(self, standard_time_offset: int = 7 * 3600, dst_observed: bool = True) -> datetime.datetime:
"""Convert to a local datetime according to the DST bits"""
u = self.as_datetime_utc()
offset = datetime.timedelta(seconds=-standard_time_offset)
@ -471,12 +473,7 @@ class WWVBMinute(_WWVBMinute):
century = (self.year // 100) * 100
# note: This relies on timedelta seconds never including leapseconds!
return (
int(
(
self.as_datetime()
- datetime.datetime(century, 1, 1, tzinfo=datetime.timezone.utc)
).total_seconds()
)
int((self.as_datetime() - datetime.datetime(century, 1, 1, tzinfo=datetime.timezone.utc)).total_seconds())
// 60
)
@ -497,9 +494,7 @@ class WWVBMinute(_WWVBMinute):
t.am[36] = t.am[38] = AmplitudeModulation(ut1_sign)
t.am[37] = AmplitudeModulation(not ut1_sign)
t._put_am_bcd(abs(self.ut1) // 100, 40, 41, 42, 43)
t._put_am_bcd(
self.year, 45, 46, 47, 48, 50, 51, 52, 53
) # Implicitly discards all but lowest 2 digits of year
t._put_am_bcd(self.year, 45, 46, 47, 48, 50, 51, 52, 53) # Implicitly discards all but lowest 2 digits of year
t.am[55] = AmplitudeModulation(self.ly)
t.am[56] = AmplitudeModulation(self.ls)
t._put_am_bcd(self.dst, 57, 58)
@ -604,16 +599,12 @@ class WWVBMinute(_WWVBMinute):
else:
self.fill_pm_timecode_regular(t)
def next_minute(
self, newut1: Optional[int] = None, newls: Optional[bool] = None
) -> "WWVBMinute":
def next_minute(self, newut1: Optional[int] = None, newls: Optional[bool] = None) -> "WWVBMinute":
"""Return an object representing the next minute"""
d = self.as_datetime() + datetime.timedelta(minutes=1)
return self.from_datetime(d, newut1, newls, self)
def previous_minute(
self, newut1: Optional[int] = None, newls: Optional[bool] = None
) -> "WWVBMinute":
def previous_minute(self, newut1: Optional[int] = None, newls: Optional[bool] = None) -> "WWVBMinute":
"""Return an object representing the previous minute"""
d = self.as_datetime() - datetime.timedelta(minutes=1)
return self.from_datetime(d, newut1, newls, self)
@ -717,9 +708,7 @@ class WWVBMinuteIERS(WWVBMinute):
"""A WWVBMinute that uses a database of DUT1 information"""
@classmethod
def _get_dut1_info(
cls, year: int, days: int, old_time: Optional[WWVBMinute] = None
) -> Tuple[int, bool]:
def _get_dut1_info(cls, year: int, days: int, old_time: Optional[WWVBMinute] = None) -> Tuple[int, bool]:
d = datetime.datetime(year, 1, 1) + datetime.timedelta(days - 1)
return int(round(get_dut1(d) * 10)) * 100, isls(d)
@ -763,7 +752,10 @@ class WWVBTimecode:
self.phase = [PhaseModulation.UNSET] * sz
def _get_am_bcd(self, *poslist: int) -> Optional[int]:
"""Convert the bits seq[positions[0]], ... seq[positions[len(positions-1)]] [in MSB order] from BCD to decimal"""
"""Convert AM data to BCD
The the bits ``self.am[poslist[i]]`` in MSB order are converted from
BCD to integer"""
pos = reversed(poslist)
val = [bool(self.am[p]) for p in pos]
result = 0
@ -779,7 +771,11 @@ class WWVBTimecode:
return result
def _put_am_bcd(self, v: int, *poslist: int) -> None:
"""Treating 'poslist' as a sequence of indices, update the AM signal with the value as a BCD number"""
"""Insert BCD coded data into the AM signal
The bits at ``self.am[poslist[i]]`` in MSB order are filled with
the conversion of `v` to BCD
Treating 'poslist' as a sequence of indices, update the AM signal with the value as a BCD number"""
pos = list(poslist)[::-1]
for p, b in zip(pos, bcd_bits(v)):
if b:
@ -798,9 +794,7 @@ class WWVBTimecode:
def __str__(self) -> str:
"""implement str()"""
undefined = [
i for i in range(len(self.am)) if self.am[i] == AmplitudeModulation.UNSET
]
undefined = [i for i in range(len(self.am)) if self.am[i] == AmplitudeModulation.UNSET]
if undefined:
warnings.warn(f"am{undefined} is unset")

9
src/wwvb/decode.py
View file

@ -23,9 +23,7 @@ import wwvb
always_zero = set((4, 10, 11, 14, 20, 21, 34, 35, 44, 54))
def wwvbreceive() -> (
Generator[Optional[wwvb.WWVBTimecode], wwvb.AmplitudeModulation, None]
): # pylint: disable=too-many-branches
def wwvbreceive() -> Generator[Optional[wwvb.WWVBTimecode], wwvb.AmplitudeModulation, None]: # pylint: disable=too-many-branches
"""A stateful decoder of WWVB signals"""
minute: List[wwvb.AmplitudeModulation] = []
state = 1
@ -60,10 +58,7 @@ def wwvbreceive() -> (
elif len(minute) % 10 and value == wwvb.AmplitudeModulation.MARK:
# print("UNEXPECTED MARK")
state = 1
elif (
len(minute) - 1 in always_zero
and value != wwvb.AmplitudeModulation.ZERO
):
elif len(minute) - 1 in always_zero and value != wwvb.AmplitudeModulation.ZERO:
# print("UNEXPECTED NONZERO")
state = 1
elif len(minute) == 60:

12
src/wwvb/gen.py
View file

@ -26,9 +26,7 @@ def parse_timespec( # pylint: disable=unused-argument
return datetime.datetime(year, month, day, hour, minute)
if len(value) == 4:
year, yday, hour, minute = map(int, value)
return datetime.datetime(year, 1, 1, hour, minute) + datetime.timedelta(
days=yday - 1
)
return datetime.datetime(year, 1, 1, hour, minute) + datetime.timedelta(days=yday - 1)
if len(value) == 1:
return dateutil.parser.parse(value[0])
if len(value) == 0:
@ -68,9 +66,7 @@ def parse_timespec( # pylint: disable=unused-argument
help="Force no leap second at the end of the month (Implies --no-iers)",
)
@click.option("--dut1", "-d", type=int, help="Force the DUT1 value (Implies --no-iers)")
@click.option(
"--minutes", "-m", default=10, help="Number of minutes to show (default: 10)"
)
@click.option("--minutes", "-m", default=10, help="Number of minutes to show (default: 10)")
@click.option(
"--style",
default="default",
@ -127,9 +123,7 @@ def main(
if style == "json":
print_timecodes_json(w, minutes, channel, file=sys.stdout)
else:
print_timecodes(
w, minutes, channel, style, all_timecodes=all_timecodes, file=sys.stdout
)
print_timecodes(w, minutes, channel, style, all_timecodes=all_timecodes, file=sys.stdout)
if __name__ == "__main__": # pragma no branch

4
src/wwvb/iersdata.py
View file

@ -24,8 +24,6 @@ for location in [
exec(f.read(), globals(), globals()) # pylint: disable=exec-used
break
start = datetime.datetime.combine(DUT1_DATA_START, datetime.time()).replace(
tzinfo=datetime.timezone.utc
)
start = datetime.datetime.combine(DUT1_DATA_START, datetime.time()).replace(tzinfo=datetime.timezone.utc)
span = datetime.timedelta(days=len(DUT1_OFFSETS))
end = start + span

126
src/wwvb/testcli.py
View file

@ -8,48 +8,57 @@
# pylint: disable=invalid-name
import json
import os
import subprocess
import sys
import unittest
from typing import Any, Sequence
coverage_add = (
("-m", "coverage", "run", "--branch", "-p") if "COVERAGE_RUN" in os.environ else ()
)
coverage_add = ("-m", "coverage", "run", "--branch", "-p") if "COVERAGE_RUN" in os.environ else ()
class CLITestCase(unittest.TestCase):
"""Test various CLI commands within wwvbpy"""
def assertProgramOutput(self, expected: str, *args: str) -> None:
"""Check the output from invoking a program matches the expected"""
def programOutput(self, *args: str) -> str:
env = os.environ.copy()
env["PYTHONIOENCODING"] = "utf-8"
actual = subprocess.check_output(
args, stdin=subprocess.DEVNULL, encoding="utf-8", env=env
)
return subprocess.check_output(args, stdin=subprocess.DEVNULL, encoding="utf-8", env=env)
def moduleArgs(self, *args: str) -> Sequence[str]:
return tuple((sys.executable, *coverage_add, "-m", *args))
def moduleOutput(self, *args: str) -> str:
return self.programOutput(sys.executable, *coverage_add, "-m", *args)
def assertProgramOutput(self, expected: str, *args: str) -> None:
"""Check the output from invoking a program matches the expected"""
actual = self.programOutput(*args)
self.assertMultiLineEqual(expected, actual, f"args={args}")
def assertProgramOutputStarts(self, expected: str, *args: str) -> None:
"""Check the output from invoking a program matches the expected"""
env = os.environ.copy()
env["PYTHONIOENCODING"] = "utf-8"
actual = subprocess.check_output(
args, stdin=subprocess.DEVNULL, encoding="utf-8", env=env
)
actual = self.programOutput(*args)
self.assertMultiLineEqual(expected, actual[: len(expected)], f"args={args}")
def assertModuleOutput(self, expected: str, *args: str) -> None:
"""Check the output from invoking a `python -m modulename` program matches the expected"""
return self.assertProgramOutput(
expected, sys.executable, *coverage_add, "-m", *args
)
actual = self.moduleOutput(*args)
self.assertMultiLineEqual(expected, actual, f"args={args}")
def assertStarts(self, expected: str, actual: str, *args: str) -> None:
self.assertMultiLineEqual(expected, actual[: len(expected)], f"args={args}")
def assertModuleJson(self, expected: Any, *args: str) -> None:
"""Check the output from invoking a `python -m modulename` program matches the expected"""
actual = self.moduleOutput(*args)
self.assertEqual(json.loads(actual), expected)
def assertModuleOutputStarts(self, expected: str, *args: str) -> None:
"""Check the output from invoking a `python -m modulename` program matches the expected"""
return self.assertProgramOutputStarts(
expected, sys.executable, *coverage_add, "-m", *args
)
actual = self.moduleOutput(*args)
self.assertStarts(expected, actual, *args)
def assertProgramError(self, *args: str) -> None:
"""Check the output from invoking a program fails"""
@ -57,16 +66,12 @@ class CLITestCase(unittest.TestCase):
env["PYTHONIOENCODING"] = "utf-8"
with self.assertRaises(subprocess.SubprocessError):
subprocess.check_output(
args,
stdin=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
encoding="utf-8",
env=env,
args, stdin=subprocess.DEVNULL, stderr=subprocess.DEVNULL, encoding="utf-8", env=env
)
def assertModuleError(self, *args: str) -> None:
"""Check the output from invoking a `python -m modulename` program fails"""
return self.assertProgramError(sys.executable, *coverage_add, "-m", *args)
self.assertProgramError(*self.moduleArgs(*args))
def test_gen(self) -> None:
"""test wwvb.gen"""
@ -153,10 +158,25 @@ WWVB timecode: year=2020 days=001 hour=12 min=30 dst=0 ut1=-300 ly=1 ls=0
def test_json(self) -> None:
"""Test the JSON output format"""
self.assertModuleOutput(
"""\
[{"year": 2021, "days": 340, "hour": 3, "minute": 40, "amplitude": "210000000200000001120011001002000000010200010001020001000002", "phase": "111110011011010101000100100110011110001110111010111101001011"}, {"year": 2021, "days": 340, "hour": 3, "minute": 41, "amplitude": "210000001200000001120011001002000000010200010001020001000002", "phase": "001010011100100011000101110000100001101000001111101100000010"}]
""",
self.assertModuleJson(
[
{
"year": 2021,
"days": 340,
"hour": 3,
"minute": 40,
"amplitude": "210000000200000001120011001002000000010200010001020001000002",
"phase": "111110011011010101000100100110011110001110111010111101001011",
},
{
"year": 2021,
"days": 340,
"hour": 3,
"minute": 41,
"amplitude": "210000001200000001120011001002000000010200010001020001000002",
"phase": "001010011100100011000101110000100001101000001111101100000010",
},
],
"wwvb.gen",
"-m",
"2",
@ -166,10 +186,23 @@ WWVB timecode: year=2020 days=001 hour=12 min=30 dst=0 ut1=-300 ly=1 ls=0
"both",
"2021-12-6 3:40",
)
self.assertModuleOutput(
"""\
[{"year": 2021, "days": 340, "hour": 3, "minute": 40, "amplitude": "210000000200000001120011001002000000010200010001020001000002"}, {"year": 2021, "days": 340, "hour": 3, "minute": 41, "amplitude": "210000001200000001120011001002000000010200010001020001000002"}]
""",
self.assertModuleJson(
[
{
"year": 2021,
"days": 340,
"hour": 3,
"minute": 40,
"amplitude": "210000000200000001120011001002000000010200010001020001000002",
},
{
"year": 2021,
"days": 340,
"hour": 3,
"minute": 41,
"amplitude": "210000001200000001120011001002000000010200010001020001000002",
},
],
"wwvb.gen",
"-m",
"2",
@ -179,10 +212,23 @@ WWVB timecode: year=2020 days=001 hour=12 min=30 dst=0 ut1=-300 ly=1 ls=0
"amplitude",
"2021-12-6 3:40",
)
self.assertModuleOutput(
"""\
[{"year": 2021, "days": 340, "hour": 3, "minute": 40, "phase": "111110011011010101000100100110011110001110111010111101001011"}, {"year": 2021, "days": 340, "hour": 3, "minute": 41, "phase": "001010011100100011000101110000100001101000001111101100000010"}]
""",
self.assertModuleJson(
[
{
"year": 2021,
"days": 340,
"hour": 3,
"minute": 40,
"phase": "111110011011010101000100100110011110001110111010111101001011",
},
{
"year": 2021,
"days": 340,
"hour": 3,
"minute": 41,
"phase": "001010011100100011000101110000100001101000001111101100000010",
},
],
"wwvb.gen",
"-m",
"2",
@ -198,9 +244,11 @@ WWVB timecode: year=2020 days=001 hour=12 min=30 dst=0 ut1=-300 ly=1 ls=0
self.assertModuleOutput(
"""\
WWVB timecode: year=2021 days=340 hour=03 min=40 dst=0 ut1=-100 ly=0 ls=0 --style=sextant
2021-340 03:40 🬋🬩🬋🬹🬩🬹🬩🬹🬩🬹🬍🬎🬍🬎🬩🬹🬩🬹🬋🬍🬩🬹🬩🬹🬍🬎🬩🬹🬍🬎🬩🬹🬍🬎🬋🬹🬋🬎🬋🬍🬍🬎🬩🬹🬋🬎🬋🬎🬩🬹🬍🬎🬋🬎🬩🬹🬩🬹🬋🬍🬍🬎🬩🬹🬩🬹🬩🬹🬩🬹🬍🬎🬍🬎🬋🬎🬩🬹🬋🬩🬩🬹🬍🬎🬩🬹🬋🬹🬩🬹🬍🬎🬩🬹🬋🬎🬩🬹🬋🬩🬩🬹🬩🬹🬍🬎🬋🬹🬍🬎🬍🬎🬩🬹🬍🬎🬩🬹🬋🬩
2021-340 03:40 \
🬋🬩🬋🬹🬩🬹🬩🬹🬩🬹🬍🬎🬍🬎🬩🬹🬩🬹🬋🬍🬩🬹🬩🬹🬍🬎🬩🬹🬍🬎🬩🬹🬍🬎🬋🬹🬋🬎🬋🬍🬍🬎🬩🬹🬋🬎🬋🬎🬩🬹🬍🬎🬋🬎🬩🬹🬩🬹🬋🬍🬍🬎🬩🬹🬩🬹🬩🬹🬩🬹🬍🬎🬍🬎🬋🬎🬩🬹🬋🬩🬩🬹🬍🬎🬩🬹🬋🬹🬩🬹🬍🬎🬩🬹🬋🬎🬩🬹🬋🬩🬩🬹🬩🬹🬍🬎🬋🬹🬍🬎🬍🬎🬩🬹🬍🬎🬩🬹🬋🬩
2021-340 03:41 🬋🬍🬋🬎🬩🬹🬍🬎🬩🬹🬍🬎🬍🬎🬩🬹🬋🬹🬋🬩🬍🬎🬍🬎🬩🬹🬍🬎🬍🬎🬍🬎🬩🬹🬋🬹🬋🬎🬋🬍🬍🬎🬩🬹🬋🬎🬋🬹🬩🬹🬩🬹🬋🬎🬍🬎🬍🬎🬋🬍🬩🬹🬍🬎🬍🬎🬍🬎🬍🬎🬩🬹🬩🬹🬋🬎🬩🬹🬋🬍🬍🬎🬍🬎🬍🬎🬋🬎🬩🬹🬩🬹🬩🬹🬋🬹🬩🬹🬋🬍🬩🬹🬩🬹🬍🬎🬋🬎🬍🬎🬍🬎🬍🬎🬍🬎🬩🬹🬋🬍
2021-340 03:41 \
🬋🬍🬋🬎🬩🬹🬍🬎🬩🬹🬍🬎🬍🬎🬩🬹🬋🬹🬋🬩🬍🬎🬍🬎🬩🬹🬍🬎🬍🬎🬍🬎🬩🬹🬋🬹🬋🬎🬋🬍🬍🬎🬩🬹🬋🬎🬋🬹🬩🬹🬩🬹🬋🬎🬍🬎🬍🬎🬋🬍🬩🬹🬍🬎🬍🬎🬍🬎🬍🬎🬩🬹🬩🬹🬋🬎🬩🬹🬋🬍🬍🬎🬍🬎🬍🬎🬋🬎🬩🬹🬩🬹🬩🬹🬋🬹🬩🬹🬋🬍🬩🬹🬩🬹🬍🬎🬋🬎🬍🬎🬍🬎🬍🬎🬍🬎🬩🬹🬋🬍
""",
"wwvb.gen",

16
src/wwvb/testdaylight.py
View file

@ -19,9 +19,7 @@ class TestDaylight(unittest.TestCase):
"""Test that the onset of DST is the same in Mountain and WWVBMinute (which uses ls bits)"""
for h in [8, 9, 10]:
for dm in range(-1441, 1442):
d = datetime.datetime(
2021, 3, 14, h, 0, tzinfo=datetime.timezone.utc
) + datetime.timedelta(minutes=dm)
d = datetime.datetime(2021, 3, 14, h, 0, tzinfo=datetime.timezone.utc) + datetime.timedelta(minutes=dm)
m = wwvb.WWVBMinute.from_datetime(d)
self.assertEqual(
m.as_datetime_local().replace(tzinfo=Mountain),
@ -32,9 +30,7 @@ class TestDaylight(unittest.TestCase):
"""Test that the end of DST is the same in Mountain and WWVBMinute (which uses ls bits)"""
for h in [7, 8, 9]:
for dm in range(-1441, 1442):
d = datetime.datetime(
2021, 11, 7, h, 0, tzinfo=datetime.timezone.utc
) + datetime.timedelta(minutes=dm)
d = datetime.datetime(2021, 11, 7, h, 0, tzinfo=datetime.timezone.utc) + datetime.timedelta(minutes=dm)
m = wwvb.WWVBMinute.from_datetime(d)
self.assertEqual(
m.as_datetime_local().replace(tzinfo=Mountain),
@ -45,9 +41,7 @@ class TestDaylight(unittest.TestCase):
"""Test that middle of DST is the same in Mountain and WWVBMinute (which uses ls bits)"""
for h in [7, 8, 9]:
for dm in (-1, 0, 1):
d = datetime.datetime(
2021, 7, 7, h, 0, tzinfo=datetime.timezone.utc
) + datetime.timedelta(minutes=dm)
d = datetime.datetime(2021, 7, 7, h, 0, tzinfo=datetime.timezone.utc) + datetime.timedelta(minutes=dm)
m = wwvb.WWVBMinute.from_datetime(d)
self.assertEqual(
m.as_datetime_local().replace(tzinfo=Mountain),
@ -58,9 +52,7 @@ class TestDaylight(unittest.TestCase):
"""Test that middle of standard time is the same in Mountain and WWVBMinute (which uses ls bits)"""
for h in [7, 8, 9]:
for dm in (-1, 0, 1):
d = datetime.datetime(
2021, 12, 25, h, 0, tzinfo=datetime.timezone.utc
) + datetime.timedelta(minutes=dm)
d = datetime.datetime(2021, 12, 25, h, 0, tzinfo=datetime.timezone.utc) + datetime.timedelta(minutes=dm)
m = wwvb.WWVBMinute.from_datetime(d)
self.assertEqual(
m.as_datetime_local().replace(tzinfo=Mountain),

4
src/wwvb/testls.py
View file

@ -55,9 +55,7 @@ class TestLeapSecond(unittest.TestCase):
leap.append(nm)
else:
assert not our_is_ls
d = datetime.datetime.combine(nm, datetime.time()).replace(
tzinfo=datetime.timezone.utc
)
d = datetime.datetime.combine(nm, datetime.time()).replace(tzinfo=datetime.timezone.utc)
self.assertEqual(leap, bench)

18
src/wwvb/testpm.py
View file

@ -15,23 +15,9 @@ class TestPhaseModulation(unittest.TestCase):
def test_pm(self) -> None:
"""Compare the generated signal from a reference minute in NIST docs"""
ref_am = (
"2011000002"
"0001001112"
"0001010002"
"0110001012"
"0100000012"
"0010010112"
)
ref_am = "2011000002" "0001001112" "0001010002" "0110001012" "0100000012" "0010010112"
ref_pm = (
"0011101101"
"0001001000"
"0011001000"
"0110001101"
"0011010001"
"0110110110"
)
ref_pm = "0011101101" "0001001000" "0011001000" "0110001101" "0011010001" "0110110110"
ref_minute = wwvb.WWVBMinuteIERS(2012, 186, 17, 30, dst=3)
ref_time = ref_minute.as_timecode()

32
src/wwvb/testuwwvb.py
View file

@ -24,7 +24,9 @@ class WWVBRoundtrip(unittest.TestCase):
def assertDateTimeEqualExceptTzInfo( # pylint: disable=invalid-name
self, a: EitherDatetimeOrNone, b: EitherDatetimeOrNone
) -> None:
"""Test two datetime objects for equality, excluding tzinfo, and allowing adafruit_datetime and core datetime modules to compare equal"""
"""Test two datetime objects for equality
This equality test excludes tzinfo, and allows adafruit_datetime and core datetime modules to compare equal"""
assert a
assert b
self.assertEqual(
@ -35,9 +37,7 @@ class WWVBRoundtrip(unittest.TestCase):
def test_decode(self) -> None:
"""Test decoding of some minutes including a leap second.
Each minute must decode and match the primary decoder."""
minute = wwvb.WWVBMinuteIERS.from_datetime(
datetime.datetime(2012, 6, 30, 23, 50)
)
minute = wwvb.WWVBMinuteIERS.from_datetime(datetime.datetime(2012, 6, 30, 23, 50))
assert minute
decoder = uwwvb.WWVBDecoder()
decoder.update(uwwvb.MARK)
@ -60,17 +60,13 @@ class WWVBRoundtrip(unittest.TestCase):
def test_roundtrip(self) -> None:
"""Test that some big range of times all decode the same as the primary decoder"""
dt = datetime.datetime(2002, 1, 1, 0, 0)
delta = datetime.timedelta(
minutes=7182 if sys.implementation.name == "cpython" else 86400 - 7182
)
delta = datetime.timedelta(minutes=7182 if sys.implementation.name == "cpython" else 86400 - 7182)
while dt.year < 2013:
minute = wwvb.WWVBMinuteIERS.from_datetime(dt)
assert minute
decoded = uwwvb.decode_wwvb([int(i) for i in minute.as_timecode().am])
assert decoded
self.assertDateTimeEqualExceptTzInfo(
minute.as_datetime_utc(), uwwvb.as_datetime_utc(decoded)
)
self.assertDateTimeEqualExceptTzInfo(minute.as_datetime_utc(), uwwvb.as_datetime_utc(decoded))
dt = dt + delta
def test_dst(self) -> None:
@ -93,9 +89,7 @@ class WWVBRoundtrip(unittest.TestCase):
minute = wwvb.WWVBMinuteIERS.from_datetime(dt)
decoded = uwwvb.decode_wwvb([int(i) for i in minute.as_timecode().am])
assert decoded
self.assertDateTimeEqualExceptTzInfo(
minute.as_datetime_local(), uwwvb.as_datetime_local(decoded)
)
self.assertDateTimeEqualExceptTzInfo(minute.as_datetime_local(), uwwvb.as_datetime_local(decoded))
decoded = uwwvb.decode_wwvb([int(i) for i in minute.as_timecode().am])
assert decoded
@ -106,9 +100,7 @@ class WWVBRoundtrip(unittest.TestCase):
def test_noise(self) -> None:
"""Test of the state-machine decoder when faced with pseudorandom noise"""
minute = wwvb.WWVBMinuteIERS.from_datetime(
datetime.datetime(2012, 6, 30, 23, 50)
)
minute = wwvb.WWVBMinuteIERS.from_datetime(datetime.datetime(2012, 6, 30, 23, 50))
r = random.Random(408)
junk = [
r.choice(
@ -141,9 +133,7 @@ class WWVBRoundtrip(unittest.TestCase):
def test_noise2(self) -> None:
"""Test of the full minute decoder with targeted errors to get full coverage"""
minute = wwvb.WWVBMinuteIERS.from_datetime(
datetime.datetime(2012, 6, 30, 23, 50)
)
minute = wwvb.WWVBMinuteIERS.from_datetime(datetime.datetime(2012, 6, 30, 23, 50))
timecode = minute.as_timecode()
decoded = uwwvb.decode_wwvb([int(i) for i in timecode.am])
self.assertIsNotNone(decoded)
@ -178,9 +168,7 @@ class WWVBRoundtrip(unittest.TestCase):
def test_noise3(self) -> None:
"""Test impossible BCD values"""
minute = wwvb.WWVBMinuteIERS.from_datetime(
datetime.datetime(2012, 6, 30, 23, 50)
)
minute = wwvb.WWVBMinuteIERS.from_datetime(datetime.datetime(2012, 6, 30, 23, 50))
timecode = minute.as_timecode()
for poslist in [

68
src/wwvb/testwwvb.py
View file

@ -85,9 +85,7 @@ class WWVBRoundtrip(unittest.TestCase):
def test_decode(self) -> None:
"""Test that a range of minutes including a leap second are correctly decoded by the state-based decoder"""
minute = wwvb.WWVBMinuteIERS.from_datetime(
datetime.datetime(1992, 6, 30, 23, 50)
)
minute = wwvb.WWVBMinuteIERS.from_datetime(datetime.datetime(1992, 6, 30, 23, 50))
decoder = decode.wwvbreceive()
next(decoder)
decoder.send(wwvb.AmplitudeModulation.MARK)
@ -126,17 +124,13 @@ class WWVBRoundtrip(unittest.TestCase):
def test_roundtrip(self) -> None:
"""Test that a wide of minutes are correctly decoded by the state-based decoder"""
dt = datetime.datetime(1992, 1, 1, 0, 0)
delta = datetime.timedelta(
minutes=915 if sys.implementation.name == "cpython" else 86400 - 915
)
delta = datetime.timedelta(minutes=915 if sys.implementation.name == "cpython" else 86400 - 915)
while dt.year < 1993:
minute = wwvb.WWVBMinuteIERS.from_datetime(dt)
assert minute is not None
timecode = minute.as_timecode().am
assert timecode
decoded_minute: Optional[
wwvb.WWVBMinute
] = wwvb.WWVBMinuteIERS.from_timecode_am(minute.as_timecode())
decoded_minute: Optional[wwvb.WWVBMinute] = wwvb.WWVBMinuteIERS.from_timecode_am(minute.as_timecode())
assert decoded_minute
decoded = decoded_minute.as_timecode().am
self.assertEqual(
@ -148,9 +142,7 @@ class WWVBRoundtrip(unittest.TestCase):
def test_noise(self) -> None:
"""Test against pseudorandom noise"""
minute = wwvb.WWVBMinuteIERS.from_datetime(
datetime.datetime(1992, 6, 30, 23, 50)
)
minute = wwvb.WWVBMinuteIERS.from_datetime(datetime.datetime(1992, 6, 30, 23, 50))
r = random.Random(408)
junk = [
r.choice(
@ -180,9 +172,7 @@ class WWVBRoundtrip(unittest.TestCase):
def test_noise2(self) -> None:
"""Test of the full minute decoder with targeted errors to get full coverage"""
minute = wwvb.WWVBMinuteIERS.from_datetime(
datetime.datetime(2012, 6, 30, 23, 50)
)
minute = wwvb.WWVBMinuteIERS.from_datetime(datetime.datetime(2012, 6, 30, 23, 50))
timecode = minute.as_timecode()
decoded = wwvb.WWVBMinute.from_timecode_am(timecode)
self.assertIsNotNone(decoded)
@ -217,9 +207,7 @@ class WWVBRoundtrip(unittest.TestCase):
def test_noise3(self) -> None:
"""Test impossible BCD values"""
minute = wwvb.WWVBMinuteIERS.from_datetime(
datetime.datetime(2012, 6, 30, 23, 50)
)
minute = wwvb.WWVBMinuteIERS.from_datetime(datetime.datetime(2012, 6, 30, 23, 50))
timecode = minute.as_timecode()
for poslist in [
@ -241,16 +229,12 @@ class WWVBRoundtrip(unittest.TestCase):
def test_previous_next_minute(self) -> None:
"""Test that previous minute and next minute are inverses"""
minute = wwvb.WWVBMinuteIERS.from_datetime(
datetime.datetime(1992, 6, 30, 23, 50)
)
minute = wwvb.WWVBMinuteIERS.from_datetime(datetime.datetime(1992, 6, 30, 23, 50))
self.assertEqual(minute, minute.next_minute().previous_minute())
def test_timecode_str(self) -> None:
"""Test the str() and repr() methods"""
minute = wwvb.WWVBMinuteIERS.from_datetime(
datetime.datetime(1992, 6, 30, 23, 50)
)
minute = wwvb.WWVBMinuteIERS.from_datetime(datetime.datetime(1992, 6, 30, 23, 50))
timecode = minute.as_timecode()
self.assertEqual(
str(timecode),
@ -268,9 +252,7 @@ class WWVBRoundtrip(unittest.TestCase):
sm1 = s - datetime.timedelta(days=1)
self.assertEqual(wwvb.get_dut1(s), wwvb.get_dut1(sm1))
e = iersdata.DUT1_DATA_START + datetime.timedelta(
days=len(iersdata.DUT1_OFFSETS) - 1
)
e = iersdata.DUT1_DATA_START + datetime.timedelta(days=len(iersdata.DUT1_OFFSETS) - 1)
ep1 = e + datetime.timedelta(days=1)
self.assertEqual(wwvb.get_dut1(e), wwvb.get_dut1(ep1))
@ -290,17 +272,11 @@ class WWVBRoundtrip(unittest.TestCase):
s = "WWVB timecode: year=1998 days=365 hour=23 min=56 dst=0 ut1=-300 ly=0 ls=1"
t = "year=1998 days=365 hour=23 min=56 dst=0 ut1=-300 ly=0 ls=1"
self.assertEqual(
wwvb.WWVBMinuteIERS.fromstring(s), wwvb.WWVBMinuteIERS.fromstring(t)
)
self.assertEqual(wwvb.WWVBMinuteIERS.fromstring(s), wwvb.WWVBMinuteIERS.fromstring(t))
t = "year=1998 days=365 hour=23 min=56 dst=0 ut1=-300 ls=1"
self.assertEqual(
wwvb.WWVBMinuteIERS.fromstring(s), wwvb.WWVBMinuteIERS.fromstring(t)
)
self.assertEqual(wwvb.WWVBMinuteIERS.fromstring(s), wwvb.WWVBMinuteIERS.fromstring(t))
t = "year=1998 days=365 hour=23 min=56 dst=0"
self.assertEqual(
wwvb.WWVBMinuteIERS.fromstring(s), wwvb.WWVBMinuteIERS.fromstring(t)
)
self.assertEqual(wwvb.WWVBMinuteIERS.fromstring(s), wwvb.WWVBMinuteIERS.fromstring(t))
def test_from_datetime(self) -> None:
"""Test the from_datetime() classmethod"""
@ -322,9 +298,7 @@ class WWVBRoundtrip(unittest.TestCase):
wwvb.WWVBMinute(2021, 1, 1, 1, ls=False)
with self.assertRaises(ValueError):
wwvb.WWVBMinute.fromstring(
"year=1998 days=365 hour=23 min=56 dst=0 ut1=-300 ly=0 ls=1 boo=1"
)
wwvb.WWVBMinute.fromstring("year=1998 days=365 hour=23 min=56 dst=0 ut1=-300 ly=0 ls=1 boo=1")
def test_deprecated(self) -> None:
"""Ensure that the 'maybe_warn_update' function is covered"""
@ -357,25 +331,19 @@ class WWVBRoundtrip(unittest.TestCase):
wwvb.get_dst_next(datetime.datetime(2005, 1, 1), tz=tz.ZoneInfo("Cuba")),
0b101111,
)
date, row = wwvb.get_dst_change_date_and_row(
datetime.datetime(2005, 1, 1), tz=tz.ZoneInfo("Cuba")
)
date, row = wwvb.get_dst_change_date_and_row(datetime.datetime(2005, 1, 1), tz=tz.ZoneInfo("Cuba"))
self.assertIsNone(date)
self.assertIsNone(row)
# California was weird in 1948
self.assertEqual(
wwvb.get_dst_next(
datetime.datetime(1948, 1, 1), tz=tz.ZoneInfo("America/Los_Angeles")
),
wwvb.get_dst_next(datetime.datetime(1948, 1, 1), tz=tz.ZoneInfo("America/Los_Angeles")),
0b100011,
)
# Berlin had DST changes on Monday in 1917
self.assertEqual(
wwvb.get_dst_next(
datetime.datetime(1917, 1, 1), tz=tz.ZoneInfo("Europe/Berlin")
),
wwvb.get_dst_next(datetime.datetime(1917, 1, 1), tz=tz.ZoneInfo("Europe/Berlin")),
0b100011,
)
@ -383,9 +351,7 @@ class WWVBRoundtrip(unittest.TestCase):
# Australia observes DST in the other half of the year compared to the
# Northern hemisphere
self.assertEqual(
wwvb.get_dst_next(
datetime.datetime(2005, 1, 1), tz=tz.ZoneInfo("Australia/Melbourne")
),
wwvb.get_dst_next(datetime.datetime(2005, 1, 1), tz=tz.ZoneInfo("Australia/Melbourne")),
0b100011,
)

20
src/wwvb/updateiers.py
View file

@ -27,10 +27,8 @@ try:
import wwvb.iersdata_dist
OLD_TABLE_START = wwvb.iersdata_dist.DUT1_DATA_START
OLD_TABLE_END = OLD_TABLE_START + datetime.timedelta(
days=len(wwvb.iersdata_dist.DUT1_OFFSETS) - 1
)
except (ImportError, NameError) as e:
OLD_TABLE_END = OLD_TABLE_START + datetime.timedelta(days=len(wwvb.iersdata_dist.DUT1_OFFSETS) - 1)
except (ImportError, NameError):
pass
IERS_URL = "https://datacenter.iers.org/data/csv/finals2000A.all.csv"
if os.path.exists("finals2000A.all.csv"):
@ -93,11 +91,7 @@ def update_iersdata( # pylint: disable=too-many-locals, too-many-branches, too-
assert wwvb_dut1_table
meta = wwvb_data.find("meta", property="article:modified_time")
assert isinstance(meta, bs4.Tag)
wwvb_data_stamp = (
datetime.datetime.fromisoformat(meta.attrs["content"])
.replace(tzinfo=None)
.date()
)
wwvb_data_stamp = datetime.datetime.fromisoformat(meta.attrs["content"]).replace(tzinfo=None).date()
def patch(patch_start: datetime.date, patch_end: datetime.date, val: int) -> None:
off_start = (patch_start - table_start).days
@ -149,9 +143,7 @@ def update_iersdata( # pylint: disable=too-many-locals, too-many-branches, too-
code(f"DUT1_DATA_START = {repr(table_start)}")
c = sorted(chr(ord("a") + ch + 10) for ch in set(offsets))
code(f"{','.join(c)} = tuple({repr(''.join(c))})")
code(
f"DUT1_OFFSETS = str( # {table_start.year:04d}{table_start.month:02d}{table_start.day:02d}"
)
code(f"DUT1_OFFSETS = str( # {table_start.year:04d}{table_start.month:02d}{table_start.day:02d}")
line = ""
j = 0
@ -194,9 +186,7 @@ def iersdata_path(callback: Callable[[str, str], str]) -> str:
default=iersdata_path(platformdirs.user_data_dir),
)
@click.option("--dist", "location", flag_value=DIST_PATH)
@click.option(
"--site", "location", flag_value=iersdata_path(platformdirs.site_data_dir)
)
@click.option("--site", "location", flag_value=iersdata_path(platformdirs.site_data_dir))
def main(location: str) -> None:
"""Update DUT1 data"""
print("will write to", location)

4
src/wwvb/wwvbtk.py
View file

@ -79,9 +79,7 @@ def main(colors: list[str], size: int, min_size: Optional[int]) -> None:
yield timestamp + i, code
timestamp = timestamp + 60
def wwvbsmarttick() -> (
Generator[Tuple[float, wwvb.AmplitudeModulation], None, None]
):
def wwvbsmarttick() -> Generator[Tuple[float, wwvb.AmplitudeModulation], None, None]:
"""Yield consecutive values of the WWVB amplitude signal but deal with time
progressing unexpectedly, such as when the computer is suspended or NTP steps
the clock backwards