Spaces:
Paused
Paused
File size: 4,009 Bytes
3647b02 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 |
# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import asyncio
import json
import random
from datetime import datetime
from glob import glob
from typing import Any, AsyncGenerator, Dict, List
async def do_research(
session_key: str,
query: str,
mock_directory: str = "mock_instances/stocks_24th_3_sections",
) -> AsyncGenerator[Dict[str, Any], None]:
"""
Generator that reads and replays events from mock_instances with random delays.
Does not save any artifacts or events.
Args:
query: The research query to process (ignored in dry mode)
session_key: Unique identifier for this research session (ignored in dry mode)
mock_directory: Directory containing mock instance files to replay
Yields:
Dict containing event data from the mock instance
"""
# Find the mock instance files
mock_files = glob(f"{mock_directory}/*.events.jsonl")
if not mock_files:
yield {
"type": "error",
"description": "No mock instance files found in mock_instances directory",
}
return
# Use the first mock file found
mock_file = mock_files[0]
# Read all events from the mock file
with open(mock_file, "r") as f:
events = [json.loads(line) for line in f if line.strip()]
# Filter for non-reporting events
research_events = [
event
for event in events
if not event.get("type", "").startswith("report_")
and event.get("type", "") != "completed"
]
# Replay each event with a random delay
for event in research_events:
# Update timestamp to current time
event["timestamp"] = datetime.now().isoformat()
# Yield the event
yield event
# Wait for a random time between 0.5 and 2 seconds
await asyncio.sleep(random.uniform(0.5, 2.0))
async def do_reporting(
session_key: str, mock_directory: str = "mock_instances/stocks_24th_3_sections"
) -> AsyncGenerator[Dict[str, Any], None]:
"""
Generator that reads and replays reporting events from mock_instances with random delays.
Does not save any artifacts or events.
Args:
session_key: Unique identifier for this research session (ignored in dry mode)
mock_directory: Directory containing mock instance files to replay
Yields:
Dict containing event data from the mock instance
"""
# Find the mock instance files
mock_files = glob(f"{mock_directory}/*.events.jsonl")
if not mock_files:
yield {
"type": "error",
"description": "No mock instance files found in mock_instances directory",
}
return
# Use the first mock file found
mock_file = mock_files[0]
# Read all events from the mock file
with open(mock_file, "r") as f:
events = [json.loads(line) for line in f if line.strip()]
# Filter for reporting events
reporting_events = [
event for event in events if event.get("type", "").startswith("report_")
]
# Replay each event with a random delay
for event in reporting_events:
# Update timestamp to current time
event["timestamp"] = datetime.now().isoformat()
# Yield the event
yield event
# Wait for a random time between 0.5 and 2 seconds
await asyncio.sleep(random.uniform(0.5, 2.0))
|