Store

Module for constructing AppletStore.

Dynamic API installation


source

load_oas

 load_oas (oas_url:str='https://tie.digitraffic.fi/swagger/openapi.json',
           destination:str='api/road_digitraffic.json',
           overwrite:bool=False)

Load OpenAPI Specification from URL or file.

Type Default Details
oas_url str https://tie.digitraffic.fi/swagger/openapi.json OpenAPI Specification URL
destination str api/road_digitraffic.json Destination file
overwrite bool False Overwrite existing file
Returns dict OpenAPI Specification

Usage to download three DigiTraffic endpoints

road_digitraffic = load_oas(
    oas_url="https://tie.digitraffic.fi/swagger/openapi.json",
    destination="api/road_digitraffic.json",
    overwrite=False
)
train_digitraffic = load_oas(
    oas_url="https://rata.digitraffic.fi/swagger/openapi.json",
    destination="api/train_digitraffic.json",
    overwrite=False
)
marine_digitraffic = load_oas(
    oas_url="https://meri.digitraffic.fi/swagger/openapi.json",
    destination="api/marine_digitraffic.json",
    overwrite=False
)

Test functions with GPT:

messages = form_msgs([
    ("system", "You are a helpful system administrator. Use the supplied tools to help the user."),
    ("user", "Get me information about the trains departing on 2017-11-09 with train number 1."),
])
complete(messages, toolbox_schema("https://rata.digitraffic.fi", train_digitraffic, fixup=generate_request))
print_msgs(messages)
>> System:
You are a helpful system administrator. Use the supplied tools to help the user.
>> User:
Get me information about the trains departing on 2017-11-09 with train number 1.
>> Assistant:
Here is the information about the train with number 1 departing on 2017-11-09:  - **Operator**: VR
(UIC Code: 10) - **Train Type**: IC (InterCity) - **Category**: Long-distance - **Running Status**:
Not currently running - **Cancelled**: No - **Timetable Type**: Regular - **Acceptance Date for
Timetable**: July 21, 2017  ### Timetable Highlights: 1. **Departure from Helsinki (HKI)**    -
Scheduled: 05:28    - Actual: 05:30    - Platform: 7    - Delay: 3 minutes  2. **Arrival at Pasila
(PSL)**    - Scheduled: 05:33    - Actual: 05:35    - Platform: 4    - Delay: 2 minutes  3.
**Arrival at Turku (TKL)**    - Scheduled: 05:43    - Actual: 05:45    - Platform: 3    - Delay: 2
minutes   4. **Arrival at Lohja (LH)**    - Scheduled: 06:16    - Actual: 06:22    - Platform: 1
- Delay: 6 minutes   5. **Final Arrival at Joensuu (JNS)**    - Scheduled: 09:45    - Actual: 09:52
- Platform: 3    - Delay: 8 minutes   ### Noteworthy stations and delays: - **Helsinki (HKI) to
Pieksämäki (PUS)**: Various stations were passed, with delays ranging from 2 to 8 minutes, with some
stops being commercial. - **Note**: The train was delayed notably due to causes listed on the logs,
such as slight track or platform issues.   **Note**: The train details reflect schedules at planned
and actual times, with slight delays recorded.
messages = form_msgs([
    ("system", "You are a helpful system administrator. Use the supplied tools to help the user."),
    ("user", "Get me information about the trains departing on 2017-11-09 with train number 1.")
])
complete(messages, toolbox_schema("https://rata.digitraffic.fi", train_digitraffic, fixup=generate_request))
print_msgs(messages)
>> System:
You are a helpful system administrator. Use the supplied tools to help the user.
>> User:
Get me information about the trains departing on 2017-11-09 with train number 1.
>> Assistant:
Here is the information about train number 1 that departed on 2017-11-09:  - **Operator:** VR (UIC
Code: 10) - **Train Type:** IC (InterCity) - **Category:** Long-distance - **Cancelled:** No
**Timetable and Stops:**  1. **Departure from Helsinki (HKI):**    - Scheduled Departure: 05:28    -
Actual Departure: 05:30:33    - Delay: 3 minutes  2. **Arrival at Pasila (PSL):**    - Scheduled
Arrival: 05:33    - Actual Arrival: 05:35:26    - Delay: 2 minutes  3. **Departure from Pasila
(PSL):**    - Scheduled Departure: 05:34    - Actual Departure: 05:37:13    - Delay: 3 minutes  4.
**Destination Joensuu (JNS):**    - Scheduled Arrival: 09:45    - Actual Arrival: 09:52:33    -
Delay: 8 minutes    - Causes of delay: Various categorized causes related to logistics and other
factors.  The train had several intermediate stops including Kerava (KE), Lahti (LH), Kouvola (KV),
and several others along its route. The train maintained various delays ranging from 3 to 8 minutes
along its journey with some minor causes contributing to delays.
messages = form_msgs([
    ("system", "You are a helpful system administrator. Use the supplied tools to help the user."),
    ("user", "Are there any active nautical warnings?")
])
complete(messages, toolbox_schema("https://meri.digitraffic.fi", marine_digitraffic, fixup=generate_request))
print_msgs(messages)
>> System:
You are a helpful system administrator. Use the supplied tools to help the user.
>> User:
Are there any active nautical warnings?
>> Assistant:
Yes, there are several active nautical warnings:  1. **Gulf of Finland**:    - **Warning**: Sunken
barge on the north side of Harmaja in approximate position 60-06.9N 024-58.5E. Vessels are advised
to navigate with caution and to avoid using the fairway west of Harmaja.  2. **Inkoo Fairway**:    -
**Warning**: Leading light Skarvörarna Front Nr 11407 at position 59-58.85N 024-08.33E is out of
order.  3. **Kotka Länsisatama 10/9 M Fairway**:    - **Warning**: Leading light Patterinmäki Rear
Nr 140 at position 60-26.94N 026-56.19E is out of order.  4. **Kaskinen Approaches (Sea of
Bothnia)**:    - **Warning**: 30 cubic meters of timber drifting in position 62-09N 021-02E.  5.
**Hessund Fairway**:    - **Warning**: The fairway is closed for traffic from 20.11.2024 to
30.4.2025 due to bridge construction work.  6. **Archipelago Sea, Gulf of Finland, Northern
Baltic**:    - **Warning**: Military exercise "Freezing Winds" taking place from 15.11-29.11.2024.
Vessels are advised to navigate with caution.  7. **Sköldvik 15.3M Fairway (Gulf of Finland)**:    -
**Warning**: Lighthouse Porvoo Nr 294 at position 60-05.59N 025-35.83E has Racon and light out of
order.  8. **Taalintehdas Fairway (Archipelago Sea)**:    - **Warning**: Sector light Padvagrundet
No 20450 at position 59-59.00N 022-46.80E is out of order.  9. **Kokkola Fairway (Bay of Bothnia)**:
- **Warning**: Leading line Outokumpu Rear No 9092 at position 63-51.55N 023-02.76E is out of order.
10. **Nyhamn-Rödkär Fairway (Archipelago Sea)**:     - **Warning**: Sector light Bogskär Nr 6292 at
position 60-04.54N 020-55.50E is out of order.  11. **Vasa Fairway (Sea of Bothnia)**:     -
**Warning**: Leading line Gåsgrund Front No 7400 at position 63-06.53N 021-14.42E Racon out of
order.  12. **Vapparn-Harvaluoto Fairway (Archipelago Sea)**:     - **Warning**: Construction work
ongoing on the Kirjala Bridge. The fairway beneath the bridge (3.6m) is closed until further notice.
13. **Kemi - Ajos Fairway**:     - **Warning**: Construction of radarmark in progress in position
65-25.45N 024-16.95E. Avoid swell when passing.  14. **GNSS and Radar Disturbances (Gulf of
Finland)**:     - **Warning**: GNSS and radar disturbances reported. Caution advised.  Please ensure
to navigate with caution and pay attention to these warnings if you are in the specified areas.

source

add_api_tools

 add_api_tools (tools:list, service_name:str, base_url:str,
                oas_url:Optional[str]=None,
                oas_destination:Optional[str]=None)

Add API tools to the toolbox.

Type Default Details
tools list List of existing tools
service_name str Name of the API service
base_url str Base URL of the API service
oas_url Optional None OpenAPI Specification URL
oas_destination Optional None OpenAPI Specification destination file

Example usage:

tools = []
add_api_tools(tools, "road_digitraffic", "https://tie.digitraffic.fi")
assert len(tools) > 0
tools[0]
{'type': 'function',
 'function': {'name': 'tmsStationsDatex2Xml',
  'description': 'The static information of TMS stations in Datex2 format (Traffic Measurement System / LAM)',
  'parameters': {'type': 'object',
   'properties': {'query': {'type': 'object',
     'properties': {'state': {'type': 'string',
       'description': 'Road station state',
       'default': 'ACTIVE',
       'enum': ['ALL', 'REMOVED', 'ACTIVE']}},
     'required': []}},
   'required': []},
  'metadata': {'url': 'https://tie.digitraffic.fi/api/beta/tms-stations-datex2.xml',
   'method': 'get',
   'accepted_queries': ['state'],
   'service': 'road_digitraffic'},
  'fixup': 'llmcam.oas_to_requests.generate_request'}}
messages = form_msgs([
    ("system", "You are a helpful system administrator. Use the supplied tools to help the user."),
    ("user", "Get the weather camera information for the stations with ID C01503 and C01504."),
])
complete(messages, tools=tools)
print_msgs(messages)
>> System:
You are a helpful system administrator. Use the supplied tools to help the user.
>> User:
Get the weather camera information for the stations with ID C01503 and C01504.
>> Assistant:
Here is the weather camera information for the stations with IDs C01503 and C01504:  ### Station
C01503 - Inkoo - **Camera Name:** kt51_Inkoo - **Camera Type:** BOSCH - **Location:** [23.99616,
60.05374] - **Municipality:** Inkoo - **Province:** Uusimaa - **Road Address:** Road 51, Section 14,
422 meters from the start - **Presets:**   - **Inkooseen:**
[Image](https://weathercam.digitraffic.fi/C0150301.jpg), Resolution: 1280x720, Direction: Increasing
- **Hankoon:** [Image](https://weathercam.digitraffic.fi/C0150302.jpg), Resolution: 1280x720,
Direction: Decreasing   - **Tienpinta:** [Image](https://weathercam.digitraffic.fi/C0150309.jpg),
Resolution: 1280x720, Direction: Special  ### Station C01504 - Karkkila - **Camera Name:**
vt2_Karkkila_Korpi - **Camera Type:** HIKVISION - **Location:** [24.235601, 60.536727] -
**Municipality:** Karkkila - **Province:** Uusimaa - **Road Address:** Road 2, Section 13, 3818
meters from the start - **Presets:**   - **Poriin:**
[Image](https://weathercam.digitraffic.fi/C0150401.jpg), Resolution: 1280x720, Direction: Increasing
- **Helsinkiin:** [Image](https://weathercam.digitraffic.fi/C0150402.jpg), Resolution: 1280x720,
Direction: Decreasing   - **Tienpinta:** [Image](https://weathercam.digitraffic.fi/C0150409.jpg),
Resolution: 1280x720, Direction: Special  These cameras are used for weather monitoring. Let me know
if you need more information!

Dynamic functions from installed libraries

It is possible to dynamically import a function in installed library and execute it with importlib.

Utility to add functions from installed libraries to ToolBox.


source

add_function_tools

 add_function_tools (tools:list, service_name:str,
                     function_names:list[str])

Add function tools to the toolbox.

Type Details
tools list List of existing tools
service_name str Name of the service
function_names list List of function names (with module prefix)

Example usage:

tools = []
add_function_tools(
    tools, 
    "youtube_live", 
    [
        "llmcam.fn_to_fc.capture_youtube_live_frame_and_save", 
        "llmcam.fn_to_fc.ask_gpt4v_about_image_file"
    ])
assert len(tools) == 2
tools
[{'type': 'function',
  'function': {'name': 'capture_youtube_live_frame_and_save',
   'description': 'Capture a jpeg file from YouTube Live and save in data directory',
   'parameters': {'type': 'object',
    'properties': {'link': {'anyOf': [{'type': 'string',
        'description': 'YouTube Live link'},
       {'type': 'null',
        'description': 'A default value will be automatically used.'}]},
     'place': {'anyOf': [{'type': 'string',
        'description': 'Location of live image'},
       {'type': 'null',
        'description': 'A default value will be automatically used.'}]}},
    'required': []},
   'metadata': {'module': 'llmcam.fn_to_fc', 'service': 'youtube_live'}}},
 {'type': 'function',
  'function': {'name': 'ask_gpt4v_about_image_file',
   'description': 'Tell all about quantitative information from a given image file',
   'parameters': {'type': 'object',
    'properties': {'path': {'type': 'string',
      'description': 'Path to the image file'}},
    'required': ['path']},
   'metadata': {'module': 'llmcam.fn_to_fc', 'service': 'youtube_live'}}}]
messages = form_msgs([
    ("system", "You are a helpful system administrator. Use the supplied tools to assist the user."),
    ("user", "Hi, can you capture and extract information from a YouTube Live? Use the default link.")
])
complete(messages, tools=tools)
print_msgs(messages)
[youtube] Extracting URL: https://www.youtube.com/watch?v=LMZQ7eFhm58
[youtube] LMZQ7eFhm58: Downloading webpage
[youtube] LMZQ7eFhm58: Downloading ios player API JSON
[youtube] LMZQ7eFhm58: Downloading mweb player API JSON
[youtube] LMZQ7eFhm58: Downloading m3u8 information
[youtube] LMZQ7eFhm58: Downloading m3u8 information
cap_2024.11.20_14:15:12_unclear.jpg
>> System:
You are a helpful system administrator. Use the supplied tools to assist the user.
>> User:
Hi, can you capture and extract information from a YouTube Live? Use the default link.
>> Assistant:
I've captured a frame from the YouTube Live and extracted the following information from the image:
- **Timestamp**: November 20, 2024, at 14:07:23 - **Location**: Kauppatori - **Dimensions**: 1280 x
720 pixels  ### Visual Details: - **Buildings**:    - Number of buildings: 8   - Height range: 3-5
stories  - **Vehicles**:   - Number of vehicles: 8   - Available parking spaces: 5   - Vehicle
types: Not specified  - **Water Bodies**:   - Visibility: True   - Type: Harbor   - Number of boats:
2  - **Street Lights**:   - Number of street lights: 15  - **People**:   - Approximate number: 20  -
**Lighting Conditions**:   - Time of day: Afternoon   - Artificial lighting: Minimal  -
**Visibility**:   - Clear conditions: False  - **Sky Conditions**:   - Sky visibility: False   -
Light conditions: Overcast  The image suggests a busy scene near a harbor during the afternoon, with
a somewhat crowded atmosphere and overcast sky conditions.

ToolBox management as FC

While the base functions for adding functions and APIs as tools work for appending new functions to an existing Python list, these functions cannot be easily transformed into tools to be used for function calling because:

  • Changes must be made into the top-level tools and continue to be applied in all subsequent conversations.
  • tools cannot be safely interpreted as a part of message as in some cases new tool changes may exceed token limit.

One solution to this is to introduce a special fix-up function is used for handlers which accesses the correct instance of tools. This function is to be created and attached to the tools instance at the moment of creation. However, this may cause issues on the application side as there might be multiple instances of tools being created. To combat this issue, we can have an alternative workflow involving a mapping of session_tools which maps each instance of tools with a corresponding unique ID. Paired with this mapping will be an fix-up function which extracts the tool instance and pass into core execution function.


source

remove_tools

 remove_tools (tools:list, service_name:str)

Remove tools from the toolbox.

Type Details
tools list List of existing tools
service_name str Name of the service

source

execute_handler_core

 execute_handler_core (tools:list, function_name:str, module:str,
                       **kwargs)

Execute the handler function by retrieving function with session ID.

Type Details
tools list Tools for each session
function_name str Name of the function to execute
module str Module of the function
kwargs

source

handler_schema

 handler_schema (function:Callable, service_name:str='toolbox_handler',
                 fixup:Optional[Callable]=None, **kwargs)

Create a schema for handlers.

Type Default Details
function Callable Handler function
service_name str toolbox_handler Name of the service
fixup Optional None Function to fixup function execution
kwargs

Example usage:

tools = []
def execute_handler(function_name, **kwargs):
    execute_handler_core(tools, function_name, **kwargs)

tools.extend([handler_schema(function, service_name="toolbox_handler", fixup=execute_handler) for function in [
        add_api_tools,
        add_function_tools,
        remove_tools
    ]])
assert len(tools) == 3
tools[0]
{'type': 'function',
 'function': {'name': 'add_api_tools',
  'description': 'Add API tools to the toolbox.',
  'parameters': {'type': 'object',
   'properties': {'service_name': {'type': 'string',
     'description': 'Name of the API service'},
    'base_url': {'type': 'string',
     'description': 'Base URL of the API service'},
    'oas_url': {'anyOf': [{'type': 'string',
       'description': 'OpenAPI Specification URL'},
      {'type': 'null',
       'description': 'A default value will be automatically used.'}]},
    'oas_destination': {'anyOf': [{'type': 'string',
       'description': 'OpenAPI Specification destination file'},
      {'type': 'null',
       'description': 'A default value will be automatically used.'}]}},
   'required': ['service_name', 'base_url']},
  'metadata': {'module': '__main__', 'service': 'toolbox_handler'},
  'fixup': '__main__.execute_handler'}}
execute_handler(
    "add_api_tools",
    module="__main__",
    service_name="road_digitraffic",
    base_url="https://tie.digitraffic.fi"
)
assert len(tools) > 3
tools[4]
{'type': 'function',
 'function': {'name': 'tmsStationsDatex2Json',
  'description': 'The static information of TMS stations in Datex2 format (Traffic Measurement System / LAM)',
  'parameters': {'type': 'object',
   'properties': {'query': {'type': 'object',
     'properties': {'state': {'type': 'string',
       'description': 'Road station state',
       'default': 'ACTIVE',
       'enum': ['ALL', 'REMOVED', 'ACTIVE']}},
     'required': []}},
   'required': []},
  'metadata': {'url': 'https://tie.digitraffic.fi/api/beta/tms-stations-datex2.json',
   'method': 'get',
   'accepted_queries': ['state'],
   'service': 'road_digitraffic'},
  'fixup': 'llmcam.oas_to_requests.generate_request'}}

Simulated GPT workflow:

from llmcam.fn_to_fc import complete, form_msg, form_msgs, print_msgs
tools = []
def execute_handler(function_name, **kwargs):
    execute_handler_core(tools, function_name, **kwargs)

tools.extend([handler_schema(function, service_name="toolbox_handler", fixup=execute_handler) for function in [
        add_api_tools,
        add_function_tools,
        remove_tools
    ]])
assert len(tools) == 3
messages = form_msgs([
    ("system", "You are a helpful system administrator. Use the supplied tools to assist the user."),
    ("user", "Add a new API service called 'road_digitraffic'. Use the base URL 'https://tie.digitraffic.fi', and the OpenAPI Specification URL 'https://tie.digitraffic.fi/swagger/openapi.json'.")
])
complete(messages, tools=tools)
print_msgs(messages)
>> System:
You are a helpful system administrator. Use the supplied tools to assist the user.
>> User:
Add a new API service called 'road_digitraffic'. Use the base URL 'https://tie.digitraffic.fi', and
the OpenAPI Specification URL 'https://tie.digitraffic.fi/swagger/openapi.json'.
>> Assistant:
The new API service 'road_digitraffic' has been added successfully with the base URL
'https://tie.digitraffic.fi' and the OpenAPI Specification URL
'https://tie.digitraffic.fi/swagger/openapi.json'.
len(tools)
64
messages.append(form_msg("user", "Get the weather camera information for the stations with ID C01503 and C01504."))
complete(messages, tools=tools)
print_msgs(messages)
>> System:
You are a helpful system administrator. Use the supplied tools to assist the user.
>> User:
Add a new API service called 'road_digitraffic'. Use the base URL 'https://tie.digitraffic.fi', and
the OpenAPI Specification URL 'https://tie.digitraffic.fi/swagger/openapi.json'.
>> Assistant:
The new API service 'road_digitraffic' has been added successfully with the base URL
'https://tie.digitraffic.fi' and the OpenAPI Specification URL
'https://tie.digitraffic.fi/swagger/openapi.json'.
>> User:
Get the weather camera information for the stations with ID C01503 and C01504.
>> Assistant:
Here is the weather camera information for the stations with ID C01503 and C01504:  ### Station
C01503 (kt51_Inkoo) - **Camera Type:** BOSCH - **Location:** Coordinates [23.99616, 60.05374] -
**Nearest Weather Station ID:** 1013 - **Collection Status:** GATHERING - **Last Updated:**
2024-11-21T03:25:42Z - **Collection Interval:** 600 seconds - **Road Address:** Road 51, Section 14,
Distance from start 422 - **Municipality:** Inkoo (Municipality Code: 149) - **Province:** Uusimaa
(Province Code: 1) - **Purpose:** Weather monitoring - **Presets:**   - **Inkooseen:**
[Image](https://weathercam.digitraffic.fi/C0150301.jpg)   - **Hankoon:**
[Image](https://weathercam.digitraffic.fi/C0150302.jpg)   - **Tienpinta:**
[Image](https://weathercam.digitraffic.fi/C0150309.jpg)  ### Station C01504 (vt2_Karkkila_Korpi) -
**Camera Type:** HIKVISION - **Location:** Coordinates [24.235601, 60.536727] - **Nearest Weather
Station ID:** 1052 - **Collection Status:** GATHERING - **Last Updated:** 2024-11-21T03:27:02Z -
**Collection Interval:** 600 seconds - **Road Address:** Road 2, Section 13, Distance from start
3818 - **Municipality:** Karkkila (Municipality Code: 224) - **Province:** Uusimaa (Province Code:
1) - **Purpose:** Weather monitoring - **Presets:**   - **Poriin:**
[Image](https://weathercam.digitraffic.fi/C0150401.jpg)   - **Helsinkiin:**
[Image](https://weathercam.digitraffic.fi/C0150402.jpg)   - **Tienpinta:**
[Image](https://weathercam.digitraffic.fi/C0150409.jpg)
messages.append(form_msg("user", "Remove the 'road_digitraffic' service."))
complete(messages, tools=tools)
print_msgs(messages)
>> System:
You are a helpful system administrator. Use the supplied tools to assist the user.
>> User:
Add a new API service called 'road_digitraffic'. Use the base URL 'https://tie.digitraffic.fi', and
the OpenAPI Specification URL 'https://tie.digitraffic.fi/swagger/openapi.json'.
>> Assistant:
The new API service 'road_digitraffic' has been added successfully with the base URL
'https://tie.digitraffic.fi' and the OpenAPI Specification URL
'https://tie.digitraffic.fi/swagger/openapi.json'.
>> User:
Get the weather camera information for the stations with ID C01503 and C01504.
>> Assistant:
Here is the weather camera information for the stations with ID C01503 and C01504:  ### Station
C01503 (kt51_Inkoo) - **Camera Type:** BOSCH - **Location:** Coordinates [23.99616, 60.05374] -
**Nearest Weather Station ID:** 1013 - **Collection Status:** GATHERING - **Last Updated:**
2024-11-21T03:25:42Z - **Collection Interval:** 600 seconds - **Road Address:** Road 51, Section 14,
Distance from start 422 - **Municipality:** Inkoo (Municipality Code: 149) - **Province:** Uusimaa
(Province Code: 1) - **Purpose:** Weather monitoring - **Presets:**   - **Inkooseen:**
[Image](https://weathercam.digitraffic.fi/C0150301.jpg)   - **Hankoon:**
[Image](https://weathercam.digitraffic.fi/C0150302.jpg)   - **Tienpinta:**
[Image](https://weathercam.digitraffic.fi/C0150309.jpg)  ### Station C01504 (vt2_Karkkila_Korpi) -
**Camera Type:** HIKVISION - **Location:** Coordinates [24.235601, 60.536727] - **Nearest Weather
Station ID:** 1052 - **Collection Status:** GATHERING - **Last Updated:** 2024-11-21T03:27:02Z -
**Collection Interval:** 600 seconds - **Road Address:** Road 2, Section 13, Distance from start
3818 - **Municipality:** Karkkila (Municipality Code: 224) - **Province:** Uusimaa (Province Code:
1) - **Purpose:** Weather monitoring - **Presets:**   - **Poriin:**
[Image](https://weathercam.digitraffic.fi/C0150401.jpg)   - **Helsinkiin:**
[Image](https://weathercam.digitraffic.fi/C0150402.jpg)   - **Tienpinta:**
[Image](https://weathercam.digitraffic.fi/C0150409.jpg)
>> User:
Remove the 'road_digitraffic' service.
>> Assistant:
The 'road_digitraffic' service has been removed successfully.
# After removing the 'road_digitraffic' service, the tools should only contain the initial handlers
len(tools)
3