Skip to content

canvod.utils API reference

Shared date/time utilities and processing diagnostics. Configuration management lives in canvod.config instead.

Tools

Utility tools for canVODpy packages.

This module provides common utilities used across all canVODpy packages: - Version management - Date/time utilities for GNSS data - Validation helpers - File hashing - Deprecation warnings

Examples:

>>> from canvod.utils.tools import get_version_from_pyproject
>>> version = get_version_from_pyproject()
>>> from canvod.utils.tools import YYYYDOY
>>> date = YYYYDOY.from_str("2025024")
>>> print(date.to_datetime())
>>> from canvod.utils.tools import gpsweekday
>>> week, day = gpsweekday("15-01-2025")  # dd-mm-yyyy
>>> from canvod.utils.tools import isfloat
>>> isfloat("3.14")  # True
>>> from canvod.utils.tools import sanitize_directory
>>> sanitize_directory(store_path)  # removes stray .DS_Store files
>>> from canvod.utils.tools import deprecated
>>> @deprecated("Use new_function instead.")
... def old_function(): ...

gpsweekday = YYYYDOY.gpsweekday module-attribute

YYYYDOY

Year and day-of-year (DOY) date representation.

Used throughout canvodpy for GNSS data organization and file naming.

Notes

This is a Pydantic dataclass and uses total_ordering for comparisons.

Attributes:

Name Type Description
year int

The year (e.g., 2025).

doy int

Day of year (1-366).

date date

Calculated calendar date.

yydoy str

Short format (YYDDD, e.g., "25001").

Examples:

>>> # From components
>>> d = YYYYDOY(year=2025, doy=1)
>>> d.to_str()
'2025001'
>>> # From date object
>>> import datetime
>>> d = YYYYDOY.from_date(datetime.date(2025, 1, 1))
>>> d.doy
1
>>> # From string
>>> d = YYYYDOY.from_str("2025001")
>>> d.year
2025
>>> # From short string (YYDDD)
>>> d = YYYYDOY.from_yydoy_str("25001")
>>> d.to_str()
'2025001'

gps_week property

GPS week number.

Returns:

Type Description
int

GPS week number since GPS epoch (1980-01-06).

gps_day_of_week property

GPS day of week (0=Sunday, 6=Saturday).

Returns:

Type Description
int

Day of week where 0=Sunday, 6=Saturday.

__post_init__()

Calculate derived fields after initialization.

Returns:

Type Description
None

__repr__()

Return the developer-focused representation.

Returns:

Type Description
str

Detailed representation of the date.

__str__()

Return the user-facing date string.

Returns:

Type Description
str

Date string in YYYYDDD format.

__eq__(other)

Compare equality based on the date string.

Parameters:

Name Type Description Default
other object

Object to compare with.

required

Returns:

Type Description
bool

True if the other object is an equal YYYYDOY instance.

__lt__(other)

Compare ordering for sorting.

Parameters:

Name Type Description Default
other object

Object to compare with.

required

Returns:

Type Description
bool

True if this date is earlier than the other date.

__hash__()

Return the hash of the date string.

Returns:

Type Description
int

Hash value for use in sets and dicts.

from_date(date) classmethod

Create from datetime.date object.

Parameters:

Name Type Description Default
date date

Calendar date

required

Returns:

Type Description
YYYYDOY

from_str(yyyydoy) classmethod

Create from YYYYDDD string.

Parameters:

Name Type Description Default
yyyydoy str or int

Date in YYYYDDD format (e.g., "2025001" or 2025001)

required

Returns:

Type Description
YYYYDOY

from_int(yyyydoy) classmethod

Create from YYYYDDD integer.

Parameters:

Name Type Description Default
yyyydoy int

Date as integer (e.g., 2025001)

required

Returns:

Type Description
YYYYDOY

from_yydoy_str(yydoy) classmethod

Create from YYDDD short string.

Assumes current millennium (20XX).

Parameters:

Name Type Description Default
yydoy str

Short date format (e.g., "25001" for 2025 DOY 001)

required

Returns:

Type Description
YYYYDOY

to_str()

Convert to YYYYDDD string.

Returns:

Type Description
str

Date in YYYYDDD format (e.g., "2025001").

gpsweekday(input_date, is_datetime=False) staticmethod

Calculate GPS week number and day of week from a given date.

GPS time started on January 6, 1980.

Parameters:

Name Type Description Default
input_date datetime.datetime, datetime.date, or str

The date to calculate GPS week for. If string, format: "dd-mm-yyyy"

required
is_datetime bool

Whether input_date is a datetime object (default: False)

False

Returns:

Type Description
tuple[int, int]

(GPS week number, day of week)

YYDOY

Two-digit year + day of year (DOY) date representation.

This is a compact format often used in GNSS filenames.

Examples:

>>> date = YYDOY.from_str("25001")  # 2025, day 1

from_str(yydoy) classmethod

Convert two-digit year DOY string to four-digit YYYYDOY.

Parameters:

Name Type Description Default
yydoy str

Two-digit year + DOY (e.g., "25001" for 2025-01-01)

required

Returns:

Type Description
YYYYDOY

Four-digit year DOY object

file_hash(path, chunk_size=8192)

Compute SHA256 hash of a GNSS data file's content.

Uses first 16 characters of the SHA256 hexdigest for a compact hash.

Parameters:

Name Type Description Default
path Path

Path to GNSS data file.

required
chunk_size int

Size of chunks to read (default: 8192 bytes).

8192

Returns:

Type Description
str

First 16 characters of SHA256 hash.

Examples:

>>> from pathlib import Path
>>> hash_val = file_hash(Path("data.rnx"))
>>> len(hash_val)
16

get_gps_week_from_filename(file_name)

Extract GPS week from various GNSS product filenames.

Parameters:

Name Type Description Default
file_name Path

GNSS product filename (e.g., SP3, CLK, TRO, IONEX)

required

Returns:

Type Description
str

GPS week as string

Raises:

Type Description
ValueError

If file type is not recognized

Examples:

>>> from pathlib import Path
>>> week = get_gps_week_from_filename(Path("igs12345.sp3"))

isfloat(value)

Check if a variable can be converted to float.

Parameters:

Name Type Description Default
value Any

Value to check.

required

Returns:

Type Description
bool

True if value can be converted to float, False otherwise.

Examples:

>>> isfloat("3.14")
True
>>> isfloat("hello")
False
>>> isfloat(42)
True

get_version_from_pyproject(pyproject_path=None)

Get the installed canvod-utils package version.

Tries importlib.metadata first, which resolves correctly for any install type (wheel, sdist, or editable). Falls back to walking up from this file's location looking for a pyproject.toml -- this only ever succeeds inside an editable/dev source checkout, since a built wheel's site-packages never ships pyproject.toml. Without the importlib.metadata path first, every store write from a non-editable install (a real PyPI release, or a git-sourced install like the demo notebooks use) raised FileNotFoundError here, since callers pass no explicit pyproject_path and expect this to just work.

Parameters:

Name Type Description Default
pyproject_path Path

Path to pyproject.toml. If None, resolved automatically as described above.

None

Returns:

Type Description
str

Version string.

Examples:

>>> version = get_version_from_pyproject()
>>> print(version)  # e.g., "0.1.0"

Files read from a raw file link in a GitHub repository.

A file is named by its raw file link, at a commit, a tag or a branch::

https://raw.githubusercontent.com/<account>/<repo>/<commit>/<path>
https://raw.githubusercontent.com/<account>/<repo>/refs/tags/<tag>/<path>
https://raw.githubusercontent.com/<account>/<repo>/refs/heads/<branch>/<path>

The user is responsible for what the link points to. Each run fetches the file again; there is no cache. A run fetches a link once, into a folder of the run (:func:run_folder) that only its owner can read; the run's worker processes read the same file, so every part of a run sees the same content. A run folder is removed when the next run of the same process starts (:func:new_run), or when the process ends (the command, the notebook kernel, the Airflow worker). A run is one site; start the next one when the earlier one is done, since its fetched files are gone then.

A file in a private repository needs a GitHub token in the environment variable GH_TOKEN or GITHUB_TOKEN. canvodpy stores no secrets (ADR "Credentials come from the environment"). The token is only sent to raw.githubusercontent.com, never to the target of a redirect, and never appears in messages or logs.

A link is shown in the terminal (canvodpy doctor, error messages) but never recorded: stores and the run history get only its file name (:func:recorded_name, :func:without_urls). A process that runs more than once starts each run with :func:new_run.

canvod-filemap reads naming recipes with this module; canvodpy reads a settings file given as a URL with it.

RemoteURL dataclass

A raw file link in a GitHub repository.

Attributes:

Name Type Description
path str

The URL's path without the leading /: <account>/<repo>/<commit, refs/tags/tag or branch>/<file path>.

url property

The raw file link.

name property

The file name, the only part of the link that is recorded.

sibling(relative)

The file at relative from this file's folder, at the same link.

Parameters:

Name Type Description Default
relative str

A path relative to the folder of this file, with / between folders; .. and absolute paths are refused.

required

Returns:

Type Description
RemoteURL

The other file, in the same repository at the same commit, tag or branch.

Raises:

Type Description
RemoteURLError

If relative is absolute or leaves the folder.

RemoteFileError

Bases: Exception

A file link that cannot be used.

RemoteURLError

Bases: RemoteFileError, ValueError

A URL that is not a raw file link in a GitHub repository.

RemoteFetchError

Bases: RemoteFileError

A file that could not be fetched, or whose content was refused.

fetch_remote(url, *, what='file', check=None, suffixes=('.yaml', '.yml'), max_bytes=MAX_BYTES, timeout=30.0)

The local file of the file at url, fetched once per run.

The URL is checked with :func:parse_remote_url. If the run already fetched it, in this process or another process of the run, that file is returned. Otherwise the file is fetched and given to check; only if check returns is it written into :func:run_folder.

Parameters:

Name Type Description Default
url str

The URL.

required
what str

What the file is, in lower case, for messages.

'file'
check Callable[[bytes], None] | None

Called with the fetched bytes; raises if the content is refused.

None
suffixes tuple[str, ...]

The file name endings accepted.

('.yaml', '.yml')
max_bytes int

Largest file accepted.

MAX_BYTES
timeout float

Seconds to wait for the server.

30.0

Returns:

Type Description
Path

The local file, with the URL's file name.

Raises:

Type Description
RemoteURLError

If the URL is not a raw file link.

RemoteFetchError

If the file cannot be fetched (not found, no access, a token that is not a token, no network, timeout) or is larger than max_bytes.

parse_remote_url(url, *, what='file', suffixes=('.yaml', '.yml'))

Check that url is a raw file link in a GitHub repository.

Parameters:

Name Type Description Default
url str

The URL.

required
what str

What the file is, in lower case, for messages ("recipe", "settings file").

'file'
suffixes tuple[str, ...]

The file name endings accepted.

('.yaml', '.yml')

Returns:

Type Description
RemoteURL

The URL's path.

Raises:

Type Description
RemoteURLError

If the URL is not https, holds a user name, password, query or fragment, points to a host other than raw.githubusercontent.com (a github.com/.../blob/... link gets the matching raw link), has no <account>/<repo>/<commit, tag or branch>/<file> path, or its file does not end in one of suffixes.

is_url(value)

Whether a setting is a URL rather than a name or a local path.

Parameters:

Name Type Description Default
value str

The setting.

required

Returns:

Type Description
bool

True for https:// and http:// values.

recorded_name(value)

The form of a setting that may be recorded: a URL's file name.

Parameters:

Name Type Description Default
value str

A setting: a URL, a name or a local path.

required

Returns:

Type Description
str

The file name of a raw file link (ROSA_canopy.yaml), the host of any other link (its path can name a private repository), any other value unchanged.

without_urls(text)

text with each http(s) link replaced by what may be recorded.

For text that is recorded, such as an error kept in the run history. Every link is replaced, not only raw file links: a message about a refused link quotes it in other forms too (a GitHub page, another host).

Parameters:

Name Type Description Default
text str

Any text.

required

Returns:

Type Description
str

The text, each link replaced by :func:recorded_name: a raw file link by its file name, any other link by its host.

run_folder()

The folder of this run's fetched files, created if missing.

A new folder in the system's temporary folder that only its owner can read, removed when the process that made it ends. Worker processes started after it was made use the same folder (it is named in the environment variable CANVODPY_RUN_FOLDER), so they read the files the run fetched and leave its removal to the run.

Returns:

Type Description
Path

The folder.

Logging

The run identifier and the stage timing; the log output is configured by canvodpy.setup_logging (see Diagnostics).

Logging helpers shared by the canvod packages.

The packages log through structlog: every module gets its logger with structlog.get_logger(__name__). The output (console, log files) is configured by canvodpy.logging.configure_logging; a package used without canvodpy logs through structlog's defaults.

  • :mod:~canvod.utils.logging.run_context: the run identifier added to every log record and Icechunk commit message of a run.
  • :mod:~canvod.utils.logging.stage_timer: timing of the processing stages and the end-of-run summary.

get_run_id()

Return the run_id bound to the current context, if any.

set_run_id(run_id)

Bind run_id to the current context.

Parameters:

Name Type Description Default
run_id str

Identifier for the current pipeline run, usually from :func:new_run_id.

required

Returns:

Type Description
Token

Token used to restore the previous value via reset_run_id.

reset_run_id(token)

Reset RUN_ID to its previous value.

new_run_id(site_name)

Return a run_id for site_name: the site, the start time in UTC, a random part.

The random part keeps run IDs apart when two machines start a run of one site in the same second; each machine has its own run history, so neither can number its run against the other.

Parameters:

Name Type Description Default
site_name str

Name of the site the run processes.

required

Returns:

Type Description
str

{site_name}-{YYYYMMDD-HHMMSS}-{6 hex digits}, for example ExampleSite-20261010-030000-3f9a1c.

timed_stage(stage, **context)

Decorator form of :func:stage_timer.

record_stage_timing(stage, duration_seconds, status='ok', **context)

Emit one stage_timing event for a duration measured by the caller.

For code that measures several stages from one set of checkpoints instead of one block per stage. The stage counts toward the run summary like a :func:stage_timer block.

Parameters:

Name Type Description Default
stage str

Stage name (e.g. "writing").

required
duration_seconds float

Duration of the stage.

required
status ('ok', 'error')

Whether the stage succeeded.

"ok"
**context Any

Extra fields attached to the event (e.g. receiver=..., date_key=...).

{}

emit_run_summary(**context)

Emit a run_summary event: per-stage breakdown for the current run.

The breakdown covers the stages timed in the main process (in canVODpy: writing the GNSS store, computing and writing VOD); the stages of the worker processes are only in the log. Called at the end of a successful run and also from the crash-handling path in cli/run.py, so a partial summary exists even for a run that later failed -- showing how far it got before dying.