Compare commits
3 Commits
mimo/code/
...
feat/mnemo
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
9d11f218ee | ||
|
|
1437613560 | ||
|
|
81b4616c03 |
BIN
nexus/mnemosyne/__pycache__/__init__.cpython-311.pyc
Normal file
BIN
nexus/mnemosyne/__pycache__/__init__.cpython-311.pyc
Normal file
Binary file not shown.
BIN
nexus/mnemosyne/__pycache__/archive.cpython-311.pyc
Normal file
BIN
nexus/mnemosyne/__pycache__/archive.cpython-311.pyc
Normal file
Binary file not shown.
BIN
nexus/mnemosyne/__pycache__/cli.cpython-311.pyc
Normal file
BIN
nexus/mnemosyne/__pycache__/cli.cpython-311.pyc
Normal file
Binary file not shown.
BIN
nexus/mnemosyne/__pycache__/entry.cpython-311.pyc
Normal file
BIN
nexus/mnemosyne/__pycache__/entry.cpython-311.pyc
Normal file
Binary file not shown.
BIN
nexus/mnemosyne/__pycache__/ingest.cpython-311.pyc
Normal file
BIN
nexus/mnemosyne/__pycache__/ingest.cpython-311.pyc
Normal file
Binary file not shown.
BIN
nexus/mnemosyne/__pycache__/linker.cpython-311.pyc
Normal file
BIN
nexus/mnemosyne/__pycache__/linker.cpython-311.pyc
Normal file
Binary file not shown.
@@ -685,3 +685,116 @@ class MnemosyneArchive:
|
||||
|
||||
self._save()
|
||||
return total_links
|
||||
|
||||
def by_date_range(
|
||||
self,
|
||||
start: str,
|
||||
end: str,
|
||||
limit: Optional[int] = None,
|
||||
) -> list[ArchiveEntry]:
|
||||
"""Return entries whose created_at falls within [start, end].
|
||||
|
||||
Args:
|
||||
start: ISO datetime string (inclusive). Can be a date-only string
|
||||
like "2026-03-01" (treated as start of that day UTC).
|
||||
end: ISO datetime string (inclusive). Can be a date-only string
|
||||
like "2026-04-01" (treated as end of that day UTC).
|
||||
limit: Maximum entries to return (None = all).
|
||||
|
||||
Returns:
|
||||
List of ArchiveEntry sorted by created_at ascending.
|
||||
"""
|
||||
start_dt = self._parse_datetime(start, end_of_day=False)
|
||||
end_dt = self._parse_datetime(end, end_of_day=True)
|
||||
|
||||
matched = []
|
||||
for entry in self._entries.values():
|
||||
entry_dt = self._parse_datetime(entry.created_at, end_of_day=False)
|
||||
if entry_dt is None:
|
||||
continue
|
||||
if start_dt <= entry_dt <= end_dt:
|
||||
matched.append(entry)
|
||||
|
||||
matched.sort(key=lambda e: e.created_at)
|
||||
if limit is not None:
|
||||
matched = matched[:limit]
|
||||
return matched
|
||||
|
||||
def temporal_neighbors(
|
||||
self,
|
||||
entry_id: str,
|
||||
window_days: int = 7,
|
||||
) -> list[ArchiveEntry]:
|
||||
"""Return entries created within N days of a given entry.
|
||||
|
||||
Args:
|
||||
entry_id: The reference entry's ID.
|
||||
window_days: Number of days before and after the reference entry's
|
||||
created_at to search (default 7).
|
||||
|
||||
Returns:
|
||||
List of ArchiveEntry sorted by time distance from the reference,
|
||||
closest first. Excludes the reference entry itself.
|
||||
|
||||
Raises:
|
||||
KeyError: If entry_id does not exist.
|
||||
"""
|
||||
from datetime import timedelta
|
||||
|
||||
ref = self._entries.get(entry_id)
|
||||
if ref is None:
|
||||
raise KeyError(entry_id)
|
||||
|
||||
ref_dt = self._parse_datetime(ref.created_at, end_of_day=False)
|
||||
if ref_dt is None:
|
||||
return []
|
||||
|
||||
window = timedelta(days=window_days)
|
||||
start_dt = ref_dt - window
|
||||
end_dt = ref_dt + window
|
||||
|
||||
neighbors = []
|
||||
for entry in self._entries.values():
|
||||
if entry.id == entry_id:
|
||||
continue
|
||||
entry_dt = self._parse_datetime(entry.created_at, end_of_day=False)
|
||||
if entry_dt is None:
|
||||
continue
|
||||
if start_dt <= entry_dt <= end_dt:
|
||||
distance = abs((entry_dt - ref_dt).total_seconds())
|
||||
neighbors.append((distance, entry))
|
||||
|
||||
neighbors.sort(key=lambda x: x[0])
|
||||
return [e for _, e in neighbors]
|
||||
|
||||
@staticmethod
|
||||
def _parse_datetime(value: str, end_of_day: bool = False) -> Optional[datetime]:
|
||||
"""Parse an ISO datetime or date-only string to a timezone-aware datetime.
|
||||
|
||||
Args:
|
||||
value: ISO datetime string, or date-only string (YYYY-MM-DD).
|
||||
end_of_day: If True and value is date-only, set time to 23:59:59 UTC.
|
||||
|
||||
Returns:
|
||||
Timezone-aware datetime (UTC), or None if parsing fails.
|
||||
"""
|
||||
if not value:
|
||||
return None
|
||||
# Date-only strings (no T separator) get special end_of_day handling
|
||||
if "T" not in value and " " not in value:
|
||||
try:
|
||||
from datetime import date
|
||||
d = date.fromisoformat(value)
|
||||
if end_of_day:
|
||||
return datetime(d.year, d.month, d.day, 23, 59, 59, tzinfo=timezone.utc)
|
||||
return datetime(d.year, d.month, d.day, tzinfo=timezone.utc)
|
||||
except ValueError:
|
||||
return None
|
||||
# Full datetime
|
||||
try:
|
||||
dt = datetime.fromisoformat(value)
|
||||
if dt.tzinfo is None:
|
||||
dt = dt.replace(tzinfo=timezone.utc)
|
||||
return dt
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
@@ -3,7 +3,8 @@
|
||||
Provides: mnemosyne ingest, mnemosyne search, mnemosyne link, mnemosyne stats,
|
||||
mnemosyne topics, mnemosyne remove, mnemosyne export,
|
||||
mnemosyne clusters, mnemosyne hubs, mnemosyne bridges, mnemosyne rebuild,
|
||||
mnemosyne tag, mnemosyne untag, mnemosyne retag
|
||||
mnemosyne tag, mnemosyne untag, mnemosyne retag,
|
||||
mnemosyne timeline, mnemosyne neighbors
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -180,8 +181,45 @@ def cmd_retag(args):
|
||||
print(f" Topics: {', '.join(entry.topics) if entry.topics else '(none)'}")
|
||||
|
||||
|
||||
def cmd_timeline(args):
|
||||
archive = MnemosyneArchive()
|
||||
results = archive.by_date_range(args.start, args.end, limit=args.limit)
|
||||
if not results:
|
||||
print(f"No entries between {args.start} and {args.end}.")
|
||||
return
|
||||
print(f"Timeline: {args.start} → {args.end} ({len(results)} entries)")
|
||||
print()
|
||||
for entry in results:
|
||||
print(f" [{entry.created_at[:10]}] {entry.title}")
|
||||
print(f" ID: {entry.id[:8]} | Source: {entry.source} | Topics: {', '.join(entry.topics)}")
|
||||
print()
|
||||
|
||||
|
||||
def cmd_neighbors(args):
|
||||
archive = MnemosyneArchive()
|
||||
# Resolve prefix to full ID
|
||||
matches = [e for e in archive._entries.values() if e.id.startswith(args.entry_id)]
|
||||
if not matches:
|
||||
print(f"No entry matching '{args.entry_id}'.")
|
||||
return
|
||||
if len(matches) > 1:
|
||||
print(f"Ambiguous — {len(matches)} entries match '{args.entry_id}'. Use a longer prefix.")
|
||||
return
|
||||
entry = matches[0]
|
||||
results = archive.temporal_neighbors(entry.id, window_days=args.days)
|
||||
if not results:
|
||||
print(f"No entries within {args.days} days of [{entry.id[:8]}] {entry.title}.")
|
||||
return
|
||||
print(f"Neighbors of [{entry.id[:8]}] {entry.title} (±{args.days} days):")
|
||||
print()
|
||||
for neighbor in results:
|
||||
print(f" [{neighbor.created_at[:10]}] {neighbor.title}")
|
||||
print(f" ID: {neighbor.id[:8]} | Source: {neighbor.source} | Topics: {', '.join(neighbor.topics)}")
|
||||
print()
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(prog="mnemosyne", description="The Living Holographic Archive")
|
||||
parser = argparse.ArgumentParser(prog="mnemosyne", description="Mnemosyne — Living Holographic Archive")
|
||||
sub = parser.add_subparsers(dest="command")
|
||||
|
||||
sub.add_parser("stats", help="Show archive statistics")
|
||||
@@ -233,6 +271,15 @@ def main():
|
||||
rt.add_argument("entry_id", help="Entry ID")
|
||||
rt.add_argument("tags", help="Comma-separated new tag list")
|
||||
|
||||
tl = sub.add_parser("timeline", help="Show entries within a date range")
|
||||
tl.add_argument("start", help="Start date (YYYY-MM-DD or ISO datetime)")
|
||||
tl.add_argument("end", help="End date (YYYY-MM-DD or ISO datetime)")
|
||||
tl.add_argument("-n", "--limit", type=int, default=50, help="Max entries to show")
|
||||
|
||||
nb = sub.add_parser("neighbors", help="Show entries temporally near a reference entry")
|
||||
nb.add_argument("entry_id", help="Reference entry ID (or prefix)")
|
||||
nb.add_argument("-d", "--days", type=int, default=7, help="Window in days (default 7)")
|
||||
|
||||
args = parser.parse_args()
|
||||
if not args.command:
|
||||
parser.print_help()
|
||||
@@ -253,6 +300,8 @@ def main():
|
||||
"tag": cmd_tag,
|
||||
"untag": cmd_untag,
|
||||
"retag": cmd_retag,
|
||||
"timeline": cmd_timeline,
|
||||
"neighbors": cmd_neighbors,
|
||||
}
|
||||
dispatch[args.command](args)
|
||||
|
||||
|
||||
BIN
nexus/mnemosyne/tests/__pycache__/__init__.cpython-311.pyc
Normal file
BIN
nexus/mnemosyne/tests/__pycache__/__init__.cpython-311.pyc
Normal file
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
160
nexus/mnemosyne/tests/test_temporal.py
Normal file
160
nexus/mnemosyne/tests/test_temporal.py
Normal file
@@ -0,0 +1,160 @@
|
||||
"""Tests for Mnemosyne temporal query methods.
|
||||
|
||||
Tests: by_date_range, temporal_neighbors, _parse_datetime.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
|
||||
from nexus.mnemosyne.archive import MnemosyneArchive
|
||||
from nexus.mnemosyne.entry import ArchiveEntry
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def archive():
|
||||
"""Create a fresh archive in a temp directory."""
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
path = Path(tmp) / "test_archive.json"
|
||||
a = MnemosyneArchive(archive_path=path)
|
||||
yield a
|
||||
|
||||
|
||||
class TestParseDatetime:
|
||||
"""Test _parse_datetime static method."""
|
||||
|
||||
def test_iso_datetime(self):
|
||||
dt = MnemosyneArchive._parse_datetime("2026-03-15T10:30:00+00:00")
|
||||
assert dt is not None
|
||||
assert dt.year == 2026
|
||||
assert dt.month == 3
|
||||
assert dt.hour == 10
|
||||
|
||||
def test_date_only_start(self):
|
||||
dt = MnemosyneArchive._parse_datetime("2026-03-15", end_of_day=False)
|
||||
assert dt is not None
|
||||
assert dt.hour == 0
|
||||
assert dt.minute == 0
|
||||
|
||||
def test_date_only_end(self):
|
||||
dt = MnemosyneArchive._parse_datetime("2026-03-15", end_of_day=True)
|
||||
assert dt is not None
|
||||
assert dt.hour == 23
|
||||
assert dt.minute == 59
|
||||
assert dt.second == 59
|
||||
|
||||
def test_naive_datetime_gets_utc(self):
|
||||
dt = MnemosyneArchive._parse_datetime("2026-03-15T10:30:00")
|
||||
assert dt is not None
|
||||
assert dt.tzinfo is not None
|
||||
|
||||
def test_empty_string(self):
|
||||
assert MnemosyneArchive._parse_datetime("") is None
|
||||
|
||||
def test_invalid_string(self):
|
||||
assert MnemosyneArchive._parse_datetime("not-a-date") is None
|
||||
|
||||
|
||||
class TestByDateRange:
|
||||
"""Test by_date_range() method."""
|
||||
|
||||
def _add_dated_entry(self, archive, title, created_at):
|
||||
"""Add an entry with a specific created_at timestamp."""
|
||||
entry = ArchiveEntry(title=title, content=f"Content for {title}")
|
||||
entry.created_at = created_at
|
||||
archive._entries[entry.id] = entry
|
||||
archive._save()
|
||||
return entry
|
||||
|
||||
def test_empty_archive(self, archive):
|
||||
results = archive.by_date_range("2026-01-01", "2026-12-31")
|
||||
assert results == []
|
||||
|
||||
def test_single_entry_in_range(self, archive):
|
||||
self._add_dated_entry(archive, "March entry", "2026-03-15T10:00:00+00:00")
|
||||
results = archive.by_date_range("2026-03-01", "2026-03-31")
|
||||
assert len(results) == 1
|
||||
assert results[0].title == "March entry"
|
||||
|
||||
def test_single_entry_out_of_range(self, archive):
|
||||
self._add_dated_entry(archive, "January entry", "2026-01-10T10:00:00+00:00")
|
||||
results = archive.by_date_range("2026-03-01", "2026-03-31")
|
||||
assert results == []
|
||||
|
||||
def test_boundary_inclusive(self, archive):
|
||||
self._add_dated_entry(archive, "Start", "2026-03-01T00:00:00+00:00")
|
||||
self._add_dated_entry(archive, "End", "2026-03-31T23:59:59+00:00")
|
||||
results = archive.by_date_range("2026-03-01", "2026-03-31")
|
||||
assert len(results) == 2
|
||||
|
||||
def test_sorted_by_date(self, archive):
|
||||
self._add_dated_entry(archive, "Late", "2026-03-20T10:00:00+00:00")
|
||||
self._add_dated_entry(archive, "Early", "2026-03-05T10:00:00+00:00")
|
||||
self._add_dated_entry(archive, "Mid", "2026-03-12T10:00:00+00:00")
|
||||
results = archive.by_date_range("2026-03-01", "2026-03-31")
|
||||
assert [r.title for r in results] == ["Early", "Mid", "Late"]
|
||||
|
||||
def test_limit(self, archive):
|
||||
for i in range(10):
|
||||
self._add_dated_entry(archive, f"Entry {i}", f"2026-03-{i+1:02d}T10:00:00+00:00")
|
||||
results = archive.by_date_range("2026-03-01", "2026-03-31", limit=3)
|
||||
assert len(results) == 3
|
||||
|
||||
def test_date_only_strings(self, archive):
|
||||
self._add_dated_entry(archive, "Mid March", "2026-03-15T10:00:00+00:00")
|
||||
results = archive.by_date_range("2026-03-01", "2026-03-31")
|
||||
assert len(results) == 1
|
||||
|
||||
|
||||
class TestTemporalNeighbors:
|
||||
"""Test temporal_neighbors() method."""
|
||||
|
||||
def _add_dated_entry(self, archive, title, created_at):
|
||||
entry = ArchiveEntry(title=title, content=f"Content for {title}")
|
||||
entry.created_at = created_at
|
||||
archive._entries[entry.id] = entry
|
||||
archive._save()
|
||||
return entry
|
||||
|
||||
def test_no_neighbors(self, archive):
|
||||
ref = self._add_dated_entry(archive, "Reference", "2026-03-15T10:00:00+00:00")
|
||||
self._add_dated_entry(archive, "Far away", "2026-06-01T10:00:00+00:00")
|
||||
results = archive.temporal_neighbors(ref.id, window_days=7)
|
||||
assert results == []
|
||||
|
||||
def test_closest_first(self, archive):
|
||||
ref = self._add_dated_entry(archive, "Reference", "2026-03-15T10:00:00+00:00")
|
||||
far = self._add_dated_entry(archive, "6 days away", "2026-03-21T10:00:00+00:00")
|
||||
close = self._add_dated_entry(archive, "1 day away", "2026-03-16T10:00:00+00:00")
|
||||
results = archive.temporal_neighbors(ref.id, window_days=7)
|
||||
assert len(results) == 2
|
||||
assert results[0].title == "1 day away"
|
||||
assert results[1].title == "6 days away"
|
||||
|
||||
def test_excludes_reference(self, archive):
|
||||
ref = self._add_dated_entry(archive, "Reference", "2026-03-15T10:00:00+00:00")
|
||||
self._add_dated_entry(archive, "Nearby", "2026-03-16T10:00:00+00:00")
|
||||
results = archive.temporal_neighbors(ref.id, window_days=7)
|
||||
ids = [r.id for r in results]
|
||||
assert ref.id not in ids
|
||||
|
||||
def test_default_window(self, archive):
|
||||
ref = self._add_dated_entry(archive, "Reference", "2026-03-15T10:00:00+00:00")
|
||||
self._add_dated_entry(archive, "7 days before", "2026-03-08T10:00:00+00:00")
|
||||
self._add_dated_entry(archive, "8 days before", "2026-03-07T10:00:00+00:00")
|
||||
results = archive.temporal_neighbors(ref.id) # default 7 days
|
||||
titles = [r.title for r in results]
|
||||
assert "7 days before" in titles
|
||||
assert "8 days before" not in titles
|
||||
|
||||
def test_custom_window(self, archive):
|
||||
ref = self._add_dated_entry(archive, "Reference", "2026-03-15T10:00:00+00:00")
|
||||
self._add_dated_entry(archive, "14 days away", "2026-03-29T10:00:00+00:00")
|
||||
results = archive.temporal_neighbors(ref.id, window_days=7)
|
||||
assert len(results) == 0
|
||||
results = archive.temporal_neighbors(ref.id, window_days=15)
|
||||
assert len(results) == 1
|
||||
|
||||
def test_bad_entry_id(self, archive):
|
||||
with pytest.raises(KeyError):
|
||||
archive.temporal_neighbors("nonexistent-id")
|
||||
Reference in New Issue
Block a user