From fb29469e49d965297f8b18aa392ae79917e5250e Mon Sep 17 00:00:00 2001 From: Aashna Batabyal Date: Sat, 26 Sep 2026 13:42:11 +0000 Subject: [PATCH 1/2] Add example: export events for external workflow-mining tools Adds examples/export_for_workflow_mining.py, which reuses the existing categorizing + AFK-filtering canonical query (same approach as load_dataframe.py) and writes a flat, sorted JSON list of {timestamp, duration, data} records -- the minimal shape a downstream sequence-clustering or process-mining tool needs, without every raw watcher heartbeat a full bucket dump would include. Passes ruff, mypy, and py_compile with the repo's existing config. README updated to list it alongside the other examples. --- README.md | 1 + examples/export_for_workflow_mining.py | 75 ++++++++++++++++++++++++++ 2 files changed, 76 insertions(+) create mode 100644 examples/export_for_workflow_mining.py diff --git a/README.md b/README.md index 2ca1223..1ccf4d6 100644 --- a/README.md +++ b/README.md @@ -72,3 +72,4 @@ The [`examples/`](examples/) directory contains a couple of example scripts, inc - [`load_dataframe.py`](https://github.com/ActivityWatch/aw-client/blob/master/examples/load_dataframe.py) - loads events from a host using a categorizing & AFK-filtering query, put result in a pandas dataframe, and export as CSV. - [`merge_buckets.py`](examples/merge_buckets.py) - merges two buckets with non-intersecting events by moving all events from one into the other. - [`redact_sensitive.py`](examples/redact_sensitive.py) - redact sensitive events. + - [`export_for_workflow_mining.py`](examples/export_for_workflow_mining.py) - exports canonical, categorized events to a flat JSON file (timestamp/duration/data per event) for use with external process-mining or workflow-discovery tools. diff --git a/examples/export_for_workflow_mining.py b/examples/export_for_workflow_mining.py new file mode 100644 index 0000000..64bdf17 --- /dev/null +++ b/examples/export_for_workflow_mining.py @@ -0,0 +1,75 @@ +""" +Export canonical window events to a flat JSON file shaped for external +process-mining / workflow-discovery tools: a list of +{"timestamp": , "duration": , "data": {...}} records, one +per event, sorted by time. + +This is deliberately *not* a bucket dump -- it reuses the same categorizing ++ AFK-filtering canonical query as load_dataframe.py, so the exported events +are already the "what was actually happening" view rather than every raw +watcher heartbeat, which is what a downstream sequence-clustering or +process-mining tool actually wants as input. +""" + +import argparse +import json +import os +import socket +from datetime import datetime, timedelta, timezone + +from aw_client import ActivityWatchClient +from aw_client.classes import default_classes +from aw_client.queries import DesktopQueryParams, canonicalEvents + + +def build_query(hostname: str) -> str: + canonicalQuery = canonicalEvents( + DesktopQueryParams( + bid_window=f"aw-watcher-window_{hostname}", + bid_afk=f"aw-watcher-afk_{hostname}", + classes=default_classes, + ) + ) + return f""" + {canonicalQuery} + RETURN = {{"events": events}}; + """ + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--days", type=int, default=7, help="How many days back to export (default: 7)" + ) + parser.add_argument( + "--hostname", + default="fakedata" if os.getenv("CI") else socket.gethostname(), + help="Hostname whose window/AFK buckets to query (default: this machine's)", + ) + parser.add_argument( + "--out", default="activitywatch_export.json", help="Output file path" + ) + args = parser.parse_args() + + now = datetime.now(tz=timezone.utc) + since = now - timedelta(days=args.days) + + aw = ActivityWatchClient(client_name="export_for_workflow_mining") + print(f"Querying the last {args.days} day(s) of events for host '{args.hostname}'...") + query = build_query(args.hostname) + data = aw.query(query, [(since, now)]) + + events = [ + {"timestamp": e["timestamp"], "duration": e["duration"], "data": e["data"]} + for e in data[0]["events"] + ] + events.sort(key=lambda e: e["timestamp"]) + + with open(args.out, "w") as f: + json.dump(events, f, indent=2) + + print(f"Wrote {len(events)} events to {args.out}") + + +if __name__ == "__main__": + main() From 6f7fe8b154a9759dbcdbb1b0b9dde20b24e225b2 Mon Sep 17 00:00:00 2001 From: Aashna Batabyal Date: Sat, 26 Sep 2026 14:17:54 +0000 Subject: [PATCH 2/2] Fix ordering, atomic writes, and add test coverage - sort by iso8601.parse_date(), not the raw timestamp string: two valid ISO8601 timestamps using different UTC offsets can sort backwards as strings even though one is unambiguously earlier once parsed - write via a temp file + os.replace() so a failure partway through serialization can't leave a truncated file in place of a prior good export - add test_* functions (matching suggest_categories_gpt.py's convention) covering the query-building, sorting, field-filtering, and atomic-write behavior -- runnable via 'make test-examples' --- examples/export_for_workflow_mining.py | 76 +++++++++++++++++++++++--- 1 file changed, 69 insertions(+), 7 deletions(-) diff --git a/examples/export_for_workflow_mining.py b/examples/export_for_workflow_mining.py index 64bdf17..e175f24 100644 --- a/examples/export_for_workflow_mining.py +++ b/examples/export_for_workflow_mining.py @@ -16,7 +16,9 @@ import os import socket from datetime import datetime, timedelta, timezone +from typing import Any +import iso8601 from aw_client import ActivityWatchClient from aw_client.classes import default_classes from aw_client.queries import DesktopQueryParams, canonicalEvents @@ -36,6 +38,31 @@ def build_query(hostname: str) -> str: """ +def serialize_events(raw_events: list) -> list: + """Keep only the fields downstream tools need, sorted by *parsed* time. + Sorting on the raw timestamp string is not equivalent: two valid + ISO8601 timestamps using different UTC offsets (e.g. "-05:00" vs "Z") + can sort backwards as strings even though one is unambiguously earlier + than the other once parsed.""" + events = [ + {"timestamp": e["timestamp"], "duration": e["duration"], "data": e["data"]} + for e in raw_events + ] + events.sort(key=lambda e: iso8601.parse_date(e["timestamp"])) + return events + + +def write_json_atomic(path: str, data: Any) -> None: + """Write to a temp file in the same directory, then atomically replace + the destination. Without this, a failure partway through serialization + (a very large export, disk full mid-write) would leave a truncated file + in place of whatever good export was already there.""" + tmp_path = f"{path}.tmp" + with open(tmp_path, "w") as f: + json.dump(data, f, indent=2) + os.replace(tmp_path, path) + + def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( @@ -59,16 +86,51 @@ def main() -> None: query = build_query(args.hostname) data = aw.query(query, [(since, now)]) - events = [ - {"timestamp": e["timestamp"], "duration": e["duration"], "data": e["data"]} - for e in data[0]["events"] + events = serialize_events(data[0]["events"]) + write_json_atomic(args.out, events) + + print(f"Wrote {len(events)} events to {args.out}") + + +def test_build_query_embeds_hostname(): + query = build_query("myhost") + assert "aw-watcher-window_myhost" in query + assert "aw-watcher-afk_myhost" in query + + +def test_serialize_events_sorts_by_parsed_time_not_string(): + # Both timestamps are valid ISO8601, but a plain string sort gets this + # backwards: the first is 2024-01-02 04:30 UTC, the second is + # 2024-01-02 01:00 UTC (earlier) -- yet "...01-01T23..." < "...01-02T01..." + # as strings, because the offsets differ. + raw = [ + {"timestamp": "2024-01-01T23:30:00-05:00", "duration": 60, "data": {"app": "later"}}, + {"timestamp": "2024-01-02T01:00:00Z", "duration": 60, "data": {"app": "earlier"}}, ] - events.sort(key=lambda e: e["timestamp"]) + result = serialize_events(raw) + assert [e["data"]["app"] for e in result] == ["earlier", "later"] - with open(args.out, "w") as f: - json.dump(events, f, indent=2) - print(f"Wrote {len(events)} events to {args.out}") +def test_serialize_events_keeps_only_expected_fields(): + raw = [{"timestamp": "2024-01-01T09:00:00Z", "duration": 60, "data": {"app": "a"}, "id": 123}] + result = serialize_events(raw) + assert set(result[0].keys()) == {"timestamp", "duration", "data"} + + +def test_write_json_atomic_preserves_prior_file_on_failed_write(tmp_path): + path = str(tmp_path / "export.json") + write_json_atomic(path, [{"ok": True}]) + + class Unserializable: + pass + + try: + write_json_atomic(path, {"bad": Unserializable()}) + except TypeError: + pass + + with open(path) as f: + assert json.load(f) == [{"ok": True}] if __name__ == "__main__":