### Project Setup and Verification Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/scheduler/drilling-wits-depth-median/README.md Sets up an isolated Python environment, installs dependencies, and runs linters and tests to verify the project setup. Requires 'make' and Python 3.8. ```bash python -m venv env source ./env/bin/activate make install make lint test ``` -------------------------------- ### Corva Task App POST Request Example (Use Case) Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/task/example_task_app/src/post_task.txt A practical example of a POST task request body for a specific use case, demonstrating how to populate the 'properties' object with relevant data. ```json { "task":{ "provider":"big-data-energy", "app_key":"big-data-energy.example_task_app", "asset_id": 67307237, "properties":{ "discounted_revenue":1574832, "discounted_operating_costs":200000, "drilling_and_completions_costs":1000000, "timestamp":1672444526 } } } ``` -------------------------------- ### Corva Task App POST Request Example (Properties) Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/task/example_task_app/src/post_task.txt An example of a POST task request body where task-specific values are provided within the 'properties' object. ```json { "task":{ "provider":"big-data-energy", "app_key":"provider.app_name", "asset_id":5678, "properties":{ "key_1":"string", "key_2":12345, "key_3":1.2345, "key_4":true }, "payload":{ "key_array":[ 123, 45678 ], "key_int":10 } } } ``` -------------------------------- ### Install App Dependencies Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/README.md Installs additional dependencies required by some Corva apps using pip. ```ruby pip install -r requirements.txt ``` -------------------------------- ### Navigate to Example Scheduler App Directory Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/README.md Navigates the user into the directory of a specific example scheduler app. ```shell cd python-apps/corva-example-python-apps/scheduler/example_scheduler_data_time_app/ ``` -------------------------------- ### Corva Task App POST Request cURL Example Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/task/example_task_app/src/post_task.txt A cURL command demonstrating how to send a POST request to the Corva Task App API with a sample task body, including authorization headers. ```c curl -X POST "https://api.corva.ai/v2/tasks" -H "accept: application/json" -H "authorization: Bearer eyJhbGciOiJIUzI1NiJ" -H "Content-Type: application/json" -d "{ \"task\":{ \"provider\":\"big-data-energy\", \"app_key\":\"big-data-energy.example_task_app\", \"asset_id\": 67307237, \"properties\":{ \"discounted_revenue\":1574832, \"discounted_operating_costs\":200000, \"drilling_and_completions_costs\":1000000, \"timestamp\":1672444526 } }}" ``` -------------------------------- ### Install Dependencies Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/task/example_task_app/README.md Installs project dependencies using the make command. Assumes 'make' is installed. ```shell make ``` -------------------------------- ### Install Dependencies Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/stream/example_stream_time_app/README.md Installs project dependencies using the 'make' command. Ensure 'make' is installed on your system. ```shell make ``` -------------------------------- ### Makefile Help Command Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/task/formation-evaluation-importer/README.md Displays available commands for managing the project, including installation, testing, and linting. ```makefile help: @echo "Available commands:" @echo " install - Install dependencies" @echo " lint - Run static code analysis" @echo " test - Run unit tests" @echo " testcov - Run unit tests with coverage report" ``` -------------------------------- ### Install Dependencies Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/scheduler/example_scheduler_depth_app/README.md Installs the necessary dependencies for the Python application. Requires 'make' to be installed on the system. ```sh make ``` -------------------------------- ### Clone Corva Example Python Apps Repository Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/README.md Clones the Corva example Python apps repository from GitHub to your local machine. ```shell git clone https://github.com/corva-ai/corva-example-python-apps.git ``` -------------------------------- ### Install Dependencies Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/scheduler/example_scheduler_data_time_app/README.md Installs the necessary dependencies for the Python application. Requires 'make' to be installed on the system. ```sh make ``` -------------------------------- ### Install Dependencies Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/scheduler/example_scheduler_natural_time_app/README.md Installs the necessary dependencies for the Python application. Requires 'make' to be installed on the system. ```sh make ``` -------------------------------- ### Run Example Scheduler App Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/README.md Executes the Lambda function for the example scheduler data time app. ```ruby python lambda_function.py ``` -------------------------------- ### View Available Commands Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/scheduler/drilling-wits-depth-median/README.md Displays a list of available commands for the project. Requires 'make' to be installed. ```bash make help ``` -------------------------------- ### Makefile Installation Command Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/task/formation-evaluation-importer/README.md Installs project dependencies, typically within a virtual environment. ```makefile install: @echo "Installing dependencies..." @pip install -r requirements.txt ``` -------------------------------- ### Create App Bundle Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/scheduler/example_scheduler_natural_time_app/README.md Creates a zip archive of the application for distribution. Requires 'make' to be installed. ```sh make bundle ``` -------------------------------- ### Create App Bundle Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/scheduler/example_scheduler_depth_app/README.md Creates a zip archive of the application for distribution. Requires 'make' to be installed. ```sh make bundle ``` -------------------------------- ### Corva Example Python App Dependencies Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/task/formation-evaluation-importer/requirements.txt Specifies the Python packages and their versions required for the Corva example applications. It highlights the importance of locking the numpy version to 1.20.3 due to potential compatibility issues with lasio. ```python corva-sdk==1.0.1 lasio==0.29 numpy==1.20.3 pydantic==1.8.2 requests==2.25.1 ``` -------------------------------- ### Run Tests Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/scheduler/example_scheduler_natural_time_app/README.md Executes the test suite for the Python application. Requires 'make' to be installed. ```sh make test ``` -------------------------------- ### Create App Bundle Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/scheduler/example_scheduler_data_time_app/README.md Creates a zip archive of the application for distribution. Requires 'make' to be installed. ```sh make bundle ``` -------------------------------- ### Run Tests Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/scheduler/example_scheduler_data_time_app/README.md Executes the test suite for the Python application. Requires 'make' to be installed. ```sh make test ``` -------------------------------- ### Run Tests Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/scheduler/example_scheduler_depth_app/README.md Executes the test suite for the Python application. Requires 'make' to be installed. ```sh make test ``` -------------------------------- ### Corva Task App POST Request Structure Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/task/example_task_app/src/post_task.txt Defines the required structure for the POST request body sent to the Corva Task App API. It includes the 'task' object with 'provider', 'app_key', 'asset_id', and optional 'properties' and 'payload' objects. ```json { "task":{ "provider":"string", "app_key":"string", "asset_id":integer, "properties": { }, "payload": { } } } ``` -------------------------------- ### Corva Event Structure Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/docs/scheduled.md Defines the structure of events used in Corva's scheduled data apps. Each event contains asset and company identifiers, along with start and end times for the data range. ```python { "asset_id": "Asset ID", "company_id": "Company ID", "start_time": "Left bound of the time range, covered by this event. Use inclusively.", "end_time": "Right bound of the time range, covered by this event. Use inclusively." } ``` -------------------------------- ### Package App for Deployment Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/scheduler/drilling-wits-depth-median/README.md Builds the application package for deployment to Dev Center. Requires 'make', Python 3.8, and an opened terminal in the cloned repository. ```bash PROVIDER=my-provider DEPTH_MILESTONE=10 make build ``` -------------------------------- ### Create App Bundle Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/stream/example_stream_time_app/README.md Bundles the application into a zip archive using the 'make bundle' command, preparing it for deployment. ```shell make bundle ``` -------------------------------- ### Create App Bundle Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/task/example_task_app/README.md Bundles the Python application for deployment using the make bundle command. ```shell make bundle ``` -------------------------------- ### Makefile Unit Tests with Coverage Command Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/task/formation-evaluation-importer/README.md Runs unit tests and generates a code coverage report, often viewable in a browser. ```makefile testcov: @echo "Running tests with coverage..." @pytest --cov=corva_apps --cov-report html ``` -------------------------------- ### Run Tests Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/stream/example_stream_time_app/README.md Executes the project's test suite using the 'make test' command. This verifies the application's functionality. ```shell make test ``` -------------------------------- ### Run Tests Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/task/example_task_app/README.md Executes the project's test suite using the make test command. ```shell make test ``` -------------------------------- ### Makefile Unit Tests Command Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/task/formation-evaluation-importer/README.md Runs unit tests to verify the functionality of the application's components. ```makefile test: @echo "Running tests..." @pytest ``` -------------------------------- ### Makefile Linting Command Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/task/formation-evaluation-importer/README.md Executes static code analysis to ensure code quality and style consistency. ```makefile lint: @echo "Running linter..." @flake8 . ``` -------------------------------- ### Calculate Mean Weight on Bit using Scheduled DataTimeEvent Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/docs/scheduled.md This Python app calculates the mean weight on bit from WITS data for a given asset and time range. It fetches data using the Corva API, processes it, and stores the result. It also handles caching to prevent duplicate data processing. ```Python import statistics from corva import Api, Cache, Logger, ScheduledDataTimeEvent, scheduled @scheduled def lambda_handler(event: ScheduledDataTimeEvent, api: Api, cache: Cache): # You have to fetch the realtime drilling data for the asset based on start and end time of the event. # start_time and end_time are inclusive so the query is structured accordingly to avoid processing duplicate data # We are only querying for weight_on_bit field since that is the only field we need. records = api.get_dataset( provider='corva', dataset='wits', query={ 'asset_id': event.asset_id, 'timestamp': { '$gte': event.start_time, '$lte': event.end_time, } }, sort={'timestamp': 1}, limit=600, fields='data.weight_on_bit' ) # If we did not find any records, we can stop the execution early if not records: Logger.info('Data not found') return None # Getting last processed timestamp from cache last_exported_timestamp = int(cache.load(key='last_processed_timestamp') or 0) # Making sure we are not processing duplicate data if event.end_time <= last_exported_timestamp: Logger.info(f'Already processed data until {last_exported_timestamp}') return None # Computing mean weight on bit from the list of realtime wits records mean_weight_on_bit = statistics.mean(record.get('data', {}).get('weight_on_bit', 0) for record in records) Logger.info(f'Calculated mean WOB: {mean_weight_on_bit}') # Building the output output = { 'timestamp': event.end_time, 'asset_id': event.asset_id, 'company_id': event.event.company_id, 'provider': 'my-company', 'collection': 'my-dataset', 'data': { 'mean_weight_on_bit': mean_weight_on_bit, 'start_time': event.start_time, 'end_time': event.end_time }, 'version': 1 } # If request fails, function will be re-invoked. so no exception handling # This is to prevent data loss in case of the POST to API fails api.post( f'api/v1/data/my-company/my-dataset/', data=[output], ).raise_for_status() # Storing the processed timestamp to cache for the next invoke cache.store(key='last_processed_timestamp', value=event.end_time) return None ``` -------------------------------- ### Track Total WITS Records with Corva API Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/docs/scheduled.md Fetches WITS records, calculates the total count, and updates or creates a record in a custom dataset. It uses Corva's scheduled events, API for data operations, and cache to store the last processed timestamp. ```python from corva import Api, Cache, Logger, ScheduledNaturalTimeEvent, scheduled @scheduled def lambda_handler(event: ScheduledNaturalTimeEvent, api: Api, cache: Cache): # Fetch the latest record from the WITS dataset latest_record = api.get_dataset( provider='corva', dataset='wits', query={'asset_id': event.asset_id}, sort={'timestamp': -1}, limit=1, fields='timestamp,data.weight_on_bit' ) # If we did not find any records, we can stop the execution early if not latest_record: Logger.info('Data not found') return None latest_record = latest_record[0] # Get latest processed timestamp from the cache last_processed_timestamp = int(cache.load(key='last_processed_timestamp') or 0) # Return early if we have already processed this data if latest_record['timestamp'] <= last_processed_timestamp: Logger.info('No new data') return None # Get new records new_records = api.get_dataset( provider='corva', dataset='wits', query={ 'asset_id': event.asset_id, 'timestamp': { '$gt': last_processed_timestamp, } }, sort={'timestamp': 1}, limit=1000, fields='timestamp' ) # Get the record we want to update records = api.get_dataset( provider='my-company', dataset='my-dataset', query={ 'asset_id': event.asset_id, }, sort={'timestamp': -1}, limit=1, fields='timestamp' ) if not records: # Create a new record if we have don't have it yet output = { 'timestamp': event.end_time, 'asset_id': event.asset_id, 'company_id': event.company_id, 'provider': 'my-company', 'collection': 'my-dataset', 'data': { 'total_count': len(new_records) }, 'version': 1 } api.post('/api/v1/data/my-company/my-dataset/', data=output) else: # Update the existing record record = records[0] output = { 'data': { 'total_count': record['data']['total_count'] + len(new_records) } } api.patch(f'/api/v1/data/my-company/my-dataset/{record["id"]}/', data=output) # Storing the processed timestamp to cache for the next invoke cache.store(key='last_processed_timestamp', value=new_records[-1]['timestamp]) return ``` -------------------------------- ### Natural Time Scheduled Events Source: https://github.com/corva-ai/corva-example-python-apps/blob/master/docs/scheduled.md Explanation of natural time scheduling in Corva apps. This type of scheduling is based on clock time and is suitable for applications that do not directly depend on the time ranges passed in data time scheduled events. It also notes that app timeouts should be less than the scheduled interval to prevent overlapping invocations. ```APIDOC Natural Time Scheduled Events: - Based on clock time, not data time ranges. - Suitable for apps not dependent on event time ranges. - App timeouts should be less than scheduled interval to prevent overlaps. Event Properties: - asset_id: Asset ID - company_id: Company ID - schedule_start: Scheduled event trigger time (Unix timestamp) - interval: How often the scheduled event gets triggered (in seconds) Scenario Example: Current time: 2022/04/03 13:00:00 Well stream started: 2022/04/03 13:00:00 App scheduled every: 5 minutes (300 seconds) 1st invoke (Clock time: 2022/04/03 13:00:00): Event: asset_id: 1234 company_id: 1 schedule_start: 1649008800 interval: 300 2nd invoke (Clock time: 2022/04/03 13:05:00): Event: asset_id: 1234 company_id: 1 schedule_start: 1649009100 interval: 300 3rd invoke (Clock time: 2022/04/03 13:10:00): Event: asset_id: 1234 company_id: 1 schedule_start: 1649009400 interval: 300 ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.