edward/agenda
1
0
Fork 0
Code Issues 10 Pull requests Projects Releases Packages Wiki Activity

Improve space launch email, add tests.

Browse source
This commit is contained in:
Edward Betts 2025-08-12 12:13:27 +01:00
parent 808f5c1d22
commit ebceb4cb51
3 changed files with 667 additions and 333 deletions
Download patch file Download diff file Expand all files Collapse all files

118
agenda/thespacedevs.py
View file

@ -221,12 +221,11 @@ def format_probability_change(old_val: int, new_val: int) -> str:
return f"Launch probability changed from {old_val}% to {new_val}%"
def handle_value_changes(differences: StrDict) -> tuple[list[str], set[str]]:
"""Handle value changes in launch data."""
def format_launch_changes(differences: StrDict) -> str:
"""Convert deepdiff output to human-readable format."""
changes: list[str] = []
processed_paths: set[str] = set()
# Skip fields that aren't useful to users
SKIP_FIELDS = {
"agency_launch_attempt_count",
"agency_launch_attempt_count_year",
@ -238,84 +237,84 @@ def handle_value_changes(differences: StrDict) -> tuple[list[str], set[str]]:
"orbital_launch_attempt_count_year",
}
if "values_changed" not in differences:
return changes, processed_paths
# --- 1. Handle Special Group Value Changes ---
# Process high-level, user-friendly summaries first.
values = differences.get("values_changed", {})
if "root['status']['name']" in values:
old_val = values["root['status']['name']"]["old_value"]
new_val = values["root['status']['name']"]["new_value"]
changes.append(f"Status changed from '{old_val}' to '{new_val}'")
processed_paths.add("root['status']")
for path, change in differences["values_changed"].items():
if any(path.startswith(processed) for processed in processed_paths):
if "root['net_precision']['name']" in values:
old_val = values["root['net_precision']['name']"]["old_value"]
new_val = values["root['net_precision']['name']"]["new_value"]
changes.append(f"Launch precision changed from '{old_val}' to '{new_val}'")
processed_paths.add("root['net_precision']")
# --- 2. Handle Type Changes ---
# This is often more significant than a value change (e.g., probability becoming None).
if "type_changes" in differences:
for path, change in differences["type_changes"].items():
if any(path.startswith(p) for p in processed_paths):
continue
field = path.replace("root['", "").replace("']", "").replace("root.", "")
if field == "probability":
# Use custom formatter only for meaningful None transitions.
if change["old_type"] is type(None) or change["new_type"] is type(None):
changes.append(
format_probability_change(
change["old_value"], change["new_value"]
)
)
else: # For other type changes (e.g., int to str), use the generic message.
changes.append(
f"{field.replace('_', ' ').title()} type changed "
+ f"from {change['old_type'].__name__} to {change['new_type'].__name__}"
)
else:
changes.append(
f"{field.replace('_', ' ').title()} type changed "
+ f"from {change['old_type'].__name__} to {change['new_type'].__name__}"
)
processed_paths.add(path)
# --- 3. Handle Remaining Value Changes ---
for path, change in values.items():
if any(path.startswith(p) for p in processed_paths):
continue
field = path.replace("root['", "").replace("']", "").replace("root.", "")
if field in SKIP_FIELDS:
continue
old_val = change["old_value"]
new_val = change["new_value"]
match field:
case "status['name']":
changes.append(f"Status changed from '{old_val}' to '{new_val}'")
processed_paths.update(
[
"root['status']['id']",
"root['status']['name']",
"root['status']['abbrev']",
"root['status']['description']",
]
)
case x if x.startswith("status["):
continue
case "net_precision['name']":
changes.append(
f"Launch precision changed from '{old_val}' to '{new_val}'"
)
processed_paths.update(
[
"root['net_precision']['id']",
"root['net_precision']['name']",
"root['net_precision']['abbrev']",
"root['net_precision']['description']",
]
)
case x if x.startswith("net_precision["):
continue
case "net":
changes.append(format_datetime_change("Launch time", old_val, new_val))
case "window_start":
changes.append(
format_datetime_change("Launch window start", old_val, new_val)
)
case "window_end":
changes.append(
format_datetime_change("Launch window end", old_val, new_val)
)
case "last_updated":
changes.append(format_datetime_update("Last updated", new_val))
case "name":
changes.append(f"Mission name changed from '{old_val}' to '{new_val}'")
case "probability":
changes.append(format_probability_change(old_val, new_val))
case x if x in SKIP_FIELDS:
continue
case _:
changes.append(f"{field} changed from '{old_val}' to '{new_val}'")
processed_paths.add(path)
return changes, processed_paths
def format_launch_changes(differences: StrDict) -> str:
"""Convert deepdiff output to human-readable format using match/case."""
changes, processed_paths = handle_value_changes(differences)
# Handle other difference types...
# --- 4. Handle Added/Removed Fields ---
if "dictionary_item_added" in differences:
for path in differences["dictionary_item_added"]:
field = path.replace("root['", "").replace("']", "").replace("root.", "")
@ -326,16 +325,9 @@ def format_launch_changes(differences: StrDict) -> str:
field = path.replace("root['", "").replace("']", "").replace("root.", "")
changes.append(f"Field removed: {field.replace('_', ' ').title()}")
if "type_changes" in differences:
for path, change in differences["type_changes"].items():
field = path.replace("root['", "").replace("']", "").replace("root.", "")
changes.append(
f"{field.replace('_', ' ').title()} type changed "
+ f"from {change['old_type'].__name__} to {change['new_type'].__name__}"
)
# Sort changes for deterministic output in tests
return (
"\n".join(f"• {change}" for change in changes)
"\n".join(f"• {change}" for change in sorted(changes))
if changes
else "No specific changes detected"
)