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
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 from a GitHub link¶
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 |
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 |
required |
Returns:
| Type | Description |
|---|---|
RemoteURL
|
The other file, in the same repository at the same commit, tag or branch. |
Raises:
| Type | Description |
|---|---|
RemoteURLError
|
If |
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
|
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 ( |
'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 |
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 |
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 ( |
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: |
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: |
required |
Returns:
| Type | Description |
|---|---|
Token
|
Token used to restore the previous value via |
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
|
|
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. |
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. |
{}
|
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.