### Example Client (Python - Asyncio)

Source: https://github.com/als-rsoxs/bcs-install/blob/main/bcs.egg-info/SOURCES.txt

Demonstrates an example asynchronous client using Python's asyncio. This code serves as a template or starting point for building custom clients that interact with external services asynchronously. It requires `aiohttp`.

```python
# src/clients/Example noasyncio.py

import asyncio
import aiohttp

class ExampleClientAsync:
    def __init__(self, base_url):
        self.base_url = base_url
        self.session = aiohttp.ClientSession()

    async def fetch_resource(self, resource_id):
        """Asynchronously fetches a specific resource."""
        url = f"{self.base_url}/resources/{resource_id}"
        try:
            async with self.session.get(url) as response:
                response.raise_for_status()
                return await response.json()
        except aiohttp.ClientError as e:
            print(f"Error fetching resource {resource_id}: {e}")
            return None

    async def create_resource(self, data):
        """Asynchronously creates a new resource."""
        url = f"{self.base_url}/resources"
        try:
            async with self.session.post(url, json=data) as response:
                response.raise_for_status()
                return await response.json()
        except aiohttp.ClientError as e:
            print(f"Error creating resource: {e}")
            return None

    async def close(self):
        await self.session.close()

async def run_example():
    client = ExampleClientAsync("http://example.com/api") # Replace with target API URL
    
    # Fetch a resource
    resource = await client.fetch_resource("123")
    if resource:
        print(f"Fetched resource: {resource}")

    # Create a resource
    new_resource_data = {"name": "test_resource", "value": 99}
    created = await client.create_resource(new_resource_data)
    if created:
        print(f"Created resource: {created}")

    await client.close()

if __name__ == '__main__':
    asyncio.run(run_example())

```

--------------------------------

### Run Files Example with Python

Source: https://github.com/als-rsoxs/bcs-install/blob/main/src/clients/Example.ipynb

Demonstrates how to list, retrieve, and plot files from a BCS server. It uses `server.get_folder_listing` to get all files and folders, then `server.get_text_file` to fetch file content, and finally `plot_bcs_file` to visualize the data. Requires `random` module and `BCSServer` object.

```python
async def test_files(server: BCSz.BCSServer):
    rsp = await server.get_folder_listing(recurse=True)
    files = rsp['files']
    print(f"Found {len(files)} files in {len(rsp['folders'])} folders on the server")

    if files:
        # randomly pick one
        file = random.choice(files)
        print(f"\nImporting and plotting {file} ...")
        plot_bcs_file(file_text=(await server.get_text_file(file))['text'], plot_title=file)
        # # plot the first n files
        # for i in range(min(10, len(files))):
        #     print(f"Working on files[{i}]={files[i]}")
        #     plot_bcs_file(file_text=(await server.get_text_file(files[i]))['text'], plot_title=files[i])
```

```python
await test_files(bl_1300)
```

--------------------------------

### Example Client (Python - Synchronous)

Source: https://github.com/als-rsoxs/bcs-install/blob/main/bcs.egg-info/SOURCES.txt

Provides a synchronous example client using Python's `requests` library. This code illustrates how to make HTTP requests in a blocking manner, suitable for straightforward API interactions. It serves as a basic template for synchronous client development.

```python
# src/clients/Example.py

import requests

class ExampleClient:
    def __init__(self, base_url):
        self.base_url = base_url

    def get_item(self, item_id):
        """Synchronously retrieves an item by its ID."""
        url = f"{self.base_url}/items/{item_id}"
        try:
            response = requests.get(url)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"Error getting item {item_id}: {e}")
            return None

    def create_item(self, data):
        """Synchronously creates a new item."""
        url = f"{self.base_url}/items"
        try:
            response = requests.post(url, json=data)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"Error creating item: {e}")
            return None

if __name__ == '__main__':
    # Example Usage:
    client = ExampleClient("http://localhost:5000/api") # Replace with target API URL

    # Get an item
    item = client.get_item("abc-123")
    if item:
        print(f"Retrieved item: {item}")

    # Create an item
    new_item_data = {"name": "sample_item", "quantity": 5}
    created_item = client.create_item(new_item_data)
    if created_item:
        print(f"Created item: {created_item}")

```

--------------------------------

### Run State Variables Example with Python

Source: https://github.com/als-rsoxs/bcs-install/blob/main/src/clients/Example noasyncio.ipynb

Illustrates how to list, get, and set state variables on the BCS server. It prints the first five state variables and then modifies the 'feedback Gain' variable. Requires the 'BCSz' module.

```python
def test_state_vars(server: BCSz.BCSServer):
    svars = server.list_state_variables()
    sv_names = svars['names']
    print('\nThe first five state variables:')
    for sv_name in sv_names[:5]:
        svd = (server.get_state_variable(sv_name))
        print(f"    {sv_name:30} ({svd['type']:7}) = {svd['value']}")

    sv_name = 'feedback Gain'
    new_value = 2
    server.set_state_variable(sv_name, value=1)
    print()
    print(sv_name, 'is', (server.get_state_variable('feedback Gain'))['value'])
    print(f'Setting {sv_name} to {new_value}')
    server.set_state_variable('feedback Gain', value=new_value)
    print(sv_name, 'is', (server.get_state_variable('feedback Gain'))['value'])
```

--------------------------------

### Run AI Example and Display Output

Source: https://github.com/als-rsoxs/bcs-install/blob/main/src/clients/Example.ipynb

Executes an AI test function and displays the resulting output, which includes various channel names and their corresponding data values. This snippet is useful for verifying AI functionality and data retrieval.

```python
await test_ai(bl_1300)
```

--------------------------------

### Initialize and connect to a BCS server

Source: https://github.com/als-rsoxs/bcs-install/blob/main/src/clients/Example.ipynb

This code snippet demonstrates how to create a `BCSServer` object and establish a connection to a BCS server. It shows examples of connecting using a hostname or IP address and specifies the default port. Successful connection is indicated by the server returning its public encryption key.

```python
bl_1300 = BCSz.BCSServer()

# Connect to the server: Customize with addr=netname or addr=ipaddr
# the port should match the server configuration, usually 5577

# await bl_1300.connect(addr='bl601es.dhcp.lbl.gov', port=5577)
# await bl_1300.connect(addr='127.0.0.1', port=5577)

await bl_1300.connect(addr='localhost', port=5577)
```

--------------------------------

### Run Scans Example in Python

Source: https://github.com/als-rsoxs/bcs-install/blob/main/src/clients/Example.ipynb

This Python code snippet demonstrates how to execute scans using the `test_scans` function with a provided `bl_1300` object. It is part of the BCS Install project and is expected to return a dictionary indicating success or failure, along with a log flag and processing time. The output also includes a sequence of actions and a pandas DataFrame representing the scan data.

```python
await test_scans(bl_1300)
```

--------------------------------

### Run Motors Example in Python

Source: https://github.com/als-rsoxs/bcs-install/blob/main/src/clients/Example.ipynb

Executes a test function for motors in the BCS system. It demonstrates how to interact with motor objects, retrieve their status, and check preset configurations. This example is useful for verifying motor control and status reporting.

```python
await test_motor(bl_1300)
```

--------------------------------

### Manage State Variables with Python

Source: https://github.com/als-rsoxs/bcs-install/blob/main/src/clients/Example.ipynb

Illustrates how to list, get, and set state variables on a BCS server. It retrieves a list of all state variable names, prints the first five, and then demonstrates getting and setting a specific variable ('feedback Gain'). Requires `BCSServer` object.

```python
async def test_state_vars(server: BCSz.BCSServer):
    svars = await server.list_state_variables()
    sv_names = svars['names']
    print('\nThe first five state variables:')
    for sv_name in sv_names[:5]:
        svd = (await server.get_state_variable(sv_name))
        print(f"    {sv_name:30} ({svd['type']:7}) = {svd['value']}")

    sv_name = 'feedback Gain'
    new_value = 2
    await server.set_state_variable(sv_name, value=1)
    print()
    print(sv_name, 'is', (await server.get_state_variable('feedback Gain'))['value'])
    print(f'Setting {sv_name} to {new_value}')
    await server.set_state_variable('feedback Gain', value=new_value)
    print(sv_name, 'is', (await server.get_state_variable('feedback Gain'))['value'])
```

--------------------------------

### Run Miscellaneous Example in Python

Source: https://github.com/als-rsoxs/bcs-install/blob/main/src/clients/Example.ipynb

Executes a Python function to test miscellaneous BCS server functionalities, including retrieving subsystem statuses, fetching and displaying panel images, and retrieving video images. It utilizes the 'BCSz' library for image processing and requires a 'BCSServer' instance ('bl_1300'). Dependencies include 'PIL' (Pillow) for image manipulation and 'io' for byte stream handling.

```python
async def test_misc(server: BCSz.BCSServer):
    # subsystems
    r = await server.get_subsystem_status()
    # r['status'] is an array of dictionaries, one entry per subsystem.
    # Convert to a dictionary with the subsystem names as keys
    subsys_dict = dict([(d['name'], d['status']) for d in r['status']])
    for name in subsys_dict.keys():
        print(f"The status of the {name} is {subsys_dict[name]}")

    img_thumbnail_size = 128, 128
    # get_panel_image
    rsp = await server.get_panel_image(name='Motor Display II.vi', quality=90)

    img_decoded = BCSz.bytes_from_blob(rsp['image_blob'])

    img1 = Image.open(io.BytesIO(img_decoded))
    img1.thumbnail(img_thumbnail_size, Image.ANTIALIAS)
    display(img1)
#     img1.show()

    # get_video_image
    rsp = await server.get_video_image(name='Video 1', quality=90)

    img_decoded = BCSz.bytes_from_blob(rsp['image_blob'])

    img2 = Image.open(io.BytesIO(img_decoded))
    img2.thumbnail(img_thumbnail_size, Image.ANTIALIAS)
    display(img2)
#     img2.show()

    return
```

```python
await test_misc(bl_1300)
```

--------------------------------

### Motors and Motion Control Example in Python

Source: https://github.com/als-rsoxs/bcs-install/blob/main/src/clients/Example.ipynb

Demonstrates controlling motors and motion using the BCS server. This Python code lists available motors and presets, checks if a motor is at a specific preset, retrieves detailed motor information including status flags, and initiates a motor movement to a target position, polling for completion. It requires the 'BCSz' library and 'time' for delays.

```python
async def test_motor(server: BCSz.BCSServer):

    #   Display all the motors
    rsp = await server.list_motors()
    motor_names = rsp['names']
    print('\n\nThe BCS server has the following motors:')
    for m in motor_names:
        print(f"    {m}")

    # presets
    rsp = await server.list_presets()
    presets_array = rsp['presets']
    print('\nPresets:')
    for p in presets_array:
        print(p)

    # check if the first preset is in-position
    if len(presets_array) > 0:
        preset_names = [p['Preset Name'] for p in presets_array]
        name = preset_names[0]
        print(f'\nAt preset {name}? ', (await server.at_preset(name=name))['at_preset'])

    # Get motor info, like position, goal, status
    my_motor = 'HY'
    rsp = (await server.get_motor(motors=[my_motor]))
    hy_motor_info = rsp['data'][0]
    print(f"\nget_motor gives {hy_motor_info}")

    # Use MotorStatus helper class to test individual bit-flags of the status word
    status = BCSz.MotorStatus(hy_motor_info['status'])
    print()
    print(f'Motor {my_motor} status: {status}')
    print(f'Move complete = {status.is_set(BCSz.MotorStatus.MOVE_COMPLETE)}')
    print(f'Motor on FWD Limit = {status.is_set(BCSz.MotorStatus.FORWARD_LIMIT)}')

    # get motor FULL: the whole status record
    rd = await server.get_motor_full(motors=[my_motor])
    # print(json.dumps(rd['data'][0], indent=4, sort_keys=True))
    print(rd['data'][0])
    await server.move_motor(motors=['HY'], goals=[75])

    # wait just a bit to let the move begin.
    time.sleep(0.250)   # this sleep call will be replaced by a new endpoint
    print()

    while True:
        rd = (await server.get_motor(motors=[my_motor]))
        hy_motor_info = rd['data'][0]
        status = BCSz.MotorStatus(hy_motor_info['status'])
        print(hy_motor_info['position'], status)
        if status.is_set(BCSz.MotorStatus.MOVE_COMPLETE):
            break
        else:
            time.sleep(0.25)    # don't hit the api server constantly
```

--------------------------------

### Run DIO Example with Python

Source: https://github.com/als-rsoxs/bcs-install/blob/main/src/clients/Example.ipynb

Executes a Digital Input/Output (DIO) test using the provided `bl_1300` object. This function likely interacts with hardware or simulated DIO channels and prints their status. No external dependencies are explicitly mentioned beyond the `bl_1300` object.

```python
await test_dio(bl_1300)
```

--------------------------------

### Run State Variables Example in Python

Source: https://github.com/als-rsoxs/bcs-install/blob/main/src/clients/Example.ipynb

Executes a Python function to retrieve and display the first five state variables from the BCS server. It also demonstrates reading and setting a feedback gain value. This function requires the 'bl_1300' object, which is assumed to be a pre-configured BCS server instance.

```python
await test_state_vars(bl_1300)
```

--------------------------------

### Run Files Example with Python

Source: https://github.com/als-rsoxs/bcs-install/blob/main/src/clients/Example noasyncio.ipynb

Demonstrates how to list files and folders on the BCS server, retrieve file content, and plot the data. It randomly selects a file to process and visualize. Dependencies include the 'random' module and a 'plot_bcs_file' function.

```python
def test_files(server: BCSz.BCSServer):
    rsp = server.get_folder_listing(recurse=True)
    files = rsp['files']
    print(f"Found {len(files)} files in {len(rsp['folders'])} folders on the server")

    if files:
        # randomly pick one
        file = random.choice(files)
        print(f"\nImporting and plotting {file} ...")
        plot_bcs_file(file_text=(server.get_text_file(file))['text'], plot_title=file)
        # # plot the first n files
        # for i in range(min(10, len(files))):
        #     print(f"Working on files[{i}]={files[i]}")
        #     plot_bcs_file(file_text=(server.get_text_file(files[i]))['text'], plot_title=files[i])
```

--------------------------------

### Run DIO Example with Python

Source: https://github.com/als-rsoxs/bcs-install/blob/main/src/clients/Example noasyncio.ipynb

Executes a Digital Input/Output (DIO) test function and displays the output, indicating the status of various channels. This function likely interacts with hardware or simulated DIO components.

```python
def test_dio(server: BCSz.BCSServer):
    rsp = server.get_folder_listing(recurse=True)
    files = rsp['files']
    print(f"Found {len(files)} files in {len(rsp['folders'])} folders on the server")

    if files:
        # randomly pick one
        file = random.choice(files)
        print(f"\nImporting and plotting {file} ...")
        plot_bcs_file(file_text=(server.get_text_file(file))['text'], plot_title=file)
        # # plot the first n files
        # for i in range(min(10, len(files))):
        #     print(f"Working on files[{i}]={files[i]}")
        #     plot_bcs_file(file_text=(server.get_text_file(files[i]))['text'], plot_title=files[i])
```

--------------------------------

### Run Miscellaneous Example with Python

Source: https://github.com/als-rsoxs/bcs-install/blob/main/src/clients/Example noasyncio.ipynb

Demonstrates fetching subsystem status and retrieving and displaying a panel image. It converts the subsystem status array to a dictionary for easier access and displays a thumbnail of the 'Motor Display II.vi' panel image. Requires 'BCSz', 'Image', and 'io' modules.

```python
def test_misc(server: BCSz.BCSServer):
    # subsystems
    r = server.get_subsystem_status()
    # r['status'] is an array of dictionaries, one entry per subsystem.
    # Convert to a dictionary with the subsystem names as keys
    subsys_dict = dict([(d['name'], d['status']) for d in r['status']])
    for name in subsys_dict.keys():
        print(f"The status of the {name} is {subsys_dict[name]}")

    img_thumbnail_size = 128, 128
    #get_panel_image
    rsp = server.get_panel_image(name='Motor Display II.vi', quality=90)

    img_decoded = BCSz.bytes_from_blob(rsp['image_blob'])

    img1 = Image.open(io.BytesIO(img_decoded))
    img1.thumbnail(img_thumbnail_size, Image.ANTIALIAS)
    display(img1)
```

--------------------------------

### Project Dependencies and Metadata (pyproject.toml)

Source: https://github.com/als-rsoxs/bcs-install/blob/main/bcs.egg-info/SOURCES.txt

Defines project metadata, build system requirements, and project dependencies for the BCS install project. This file is crucial for package management and installation using tools like pip.

```toml
[build-system]
requires = ["setuptools>=61.0"]
build-backend = "setuptools.build_meta"

[project]
name = "bcs-install"
version = "0.1.0"
description = "A sample Python project."
readme = "README.md"
requires-python = ">=3.8"
classifiers = [
    "Programming Language :: Python :: 3",
    "License :: OSI Approved :: MIT License",
    "Operating System :: OS Independent",
]

[project.urls "Homepage"] = "https://github.com/pypa/sampleproject"
[project.urls "Bug Tracker"] = "https://github.com/pypa/sampleproject/issues"
```

--------------------------------

### Install Pyright for Stub Generation Troubleshooting

Source: https://github.com/als-rsoxs/bcs-install/blob/main/README.md

Installs or updates Pyright to version 1.1.300 or higher. This is a troubleshooting step if stub generation fails, ensuring the necessary tool is available and up-to-date.

```bash
pip install pyright>=1.1.300
```

--------------------------------

### Run Miscellaneous Example with Python

Source: https://github.com/als-rsoxs/bcs-install/blob/main/src/clients/Example noasyncio.ipynb

Executes a miscellaneous test function, likely for checking the status of various subsystems on the BCS server. The output indicates the status of each subsystem.

```python
test_misc(bl_1300)
```

--------------------------------

### Connect to a BCS server using BCSServer object

Source: https://github.com/als-rsoxs/bcs-install/blob/main/src/clients/Example noasyncio.ipynb

Demonstrates how to establish a connection to a BCS system's API server. It involves creating a `BCSServer` instance and then calling the `connect` method with the server's address and port. Successful connection is indicated by the server responding with its public encryption key.

```python
bl_1300 = BCSz.BCSServer()

# Connect to the server: Customize with addr=netname or addr=ipaddr
# the port should match the server configuration, usually 5577

# bl_1300.connect(addr='bl601es.dhcp.lbl.gov', port=5577)
# bl_1300.connect(addr='127.0.0.1', port=5577)

bl_1300.connect(addr='localhost', port=5577)
```

--------------------------------

### Connect to BCS Server

Source: https://context7.com/als-rsoxs/bcs-install/llms.txt

Demonstrates how to establish a secure ZMQ connection to a BCS server using the BCSServer class. It covers default connection parameters and provides an example of testing the connection.

```python
import asyncio
import BCSz

async def main():
    # Create server instance
    server = BCSz.BCSServer()

    # Connect to BCS server (default localhost:5577)
    await server.connect(addr='localhost', port=5577)

    # Connect to remote beamline
    # await server.connect(addr='bl601es.dhcp.lbl.gov', port=5577)

    # Test connection
    response = await server.test_connection()
    print(f"Connected: {response['success']}")
    # Output: Connected: True

asyncio.run(main())
```

--------------------------------

### Run AI Example and Display Output

Source: https://github.com/als-rsoxs/bcs-install/blob/main/src/clients/Example noasyncio.ipynb

Executes an AI test function and displays the resulting channel data. It retrieves freerun data for multiple channels, calculates averages, and prints both raw and processed data. Dependencies include the 'bl_1300' object and the 'test_ai' function.

```python
test_ai(bl_1300)
```

--------------------------------

### Retrieve and Display Video Image with Python

Source: https://github.com/als-rsoxs/bcs-install/blob/main/src/clients/Example noasyncio.ipynb

Fetches a video image from the server, decodes it from bytes, resizes it to a thumbnail, and displays it. Requires the 'BCSz' library and 'Pillow' for image manipulation.

```python
# get_video_image
    rsp = server.get_video_image(name='Video 1', quality=90)

    img_decoded = BCSz.bytes_from_blob(rsp['image_blob'])

    img2 = Image.open(io.BytesIO(img_decoded))
    img2.thumbnail(img_thumbnail_size, Image.ANTIALIAS)
    display(img2)
```

--------------------------------

### Connect to a BCS server

Source: https://github.com/als-rsoxs/bcs-install/blob/main/src/clients/Example.ipynb

Establishes a connection to the BCS server. You need a running BCS API server and must configure it to accept connections from your local machine. The connection process involves creating a BCSServer object and then calling the connect method with the server's address and port.

```APIDOC
## Connect to a BCS server

### Description
Establishes a connection to the BCS server. Requires a running BCS API server configured for local connections. First, create a `BCSServer` object, then call the `connect` method with the server's address and port.

### Method
`await bl_1300.connect(addr='<server_address>', port=<server_port>)`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```python
bl_1300 = BCSz.BCSServer()
# Connect to the server: Customize with addr=netname or addr=ipaddr
# the port should match the server configuration, usually 5577
await bl_1300.connect(addr='localhost', port=5577)
```

### Response
#### Success Response (200)
Returns the server's public encryption key upon successful connection.

#### Response Example
```
Server Public Key b'q>1<)5vh=$FXAZY&XuMV^K#CY4iL3PPy@AxtUX0='
```
```

--------------------------------

### Install BCS Python Package

Source: https://context7.com/als-rsoxs/bcs-install/llms.txt

Instructions for installing the BCS Python package using uv, the recommended package manager. It supports both editable installations with development dependencies and standard synchronization.

```bash
# Install with uv (recommended)
uv pip install -e ".[dev]"

# Or using uv sync
uv sync
```

--------------------------------

### Install BCS Package with Development Dependencies

Source: https://github.com/als-rsoxs/bcs-install/blob/main/README.md

Installs the BCS Python package, including development dependencies, using uv pip. This command ensures all necessary tools for development and testing are available.

```bash
uv pip install -e ".[dev]"
```

--------------------------------

### Run Scans Example in Python

Source: https://github.com/als-rsoxs/bcs-install/blob/main/src/clients/Example noasyncio.ipynb

This Python code snippet demonstrates how to execute scans using the 'test_scans' function with a provided argument 'bl_1300'. It shows the typical output format, including success status, error descriptions, logging information, API delta time, a sequence of actions, and a pandas DataFrame representing scan data.

```python
test_scans(bl_1300)
```

--------------------------------

### Manage Digital Input/Output (DIO) Channels

Source: https://github.com/als-rsoxs/bcs-install/blob/main/src/clients/Example noasyncio.ipynb

Demonstrates how to interact with Digital Input/Output (DIO) channels using the BCS server. This includes listing available DIOs, retrieving their status, setting output values, and monitoring changes. It requires a 'BCSz.BCSServer' object and the 'time' module for delays.

```python
def test_dio(server: BCSz.BCSServer):
    # Don't really need dio_list, get_di does more
    dio_list = (server.list_dios())['names']
    [print(dio) for dio in dio_list]

    # get all the dio channels, their values and if they are enabled
    r = server.get_di()

    # make a dictionary of DIO's
    dio_dict = dict(zip(r['chans'], r['data']))
    my_dio = 'Shutter'
    if my_dio in dio_dict:
        print(f"The value of {my_dio} is {dio_dict[my_dio]}")

    # make an array of just the enabled DIO's
    dios = []
    for c, u, v in zip(r['chans'], r['enabled'], r['data']):
        if u:
            print(f"Channel: {c} is {v}")
            dios.append(c)

    # watch the DIO monitor on BCS while running the code below
    [server.set_do(chan=c, value=True) for c in dios]  # turn all ON
    time.sleep(1)

    # walk the True through the channels
    for c_on in dios:
        for c in dios:
            r = server.set_do(chan=c, value=bool(c == c_on))
        time.sleep(0.25)

    # flip the first DIO
    dio_name = dios[0]
    dio_value = (server.get_di(dio_name))['data'][0]
    print(f'DIO channel: {dio_name} = {dio_value}')
    server.set_do(chan=dio_name, value=not dio_value)
    time.sleep(0.1)  # bcs subsystem needs time to propagate the change
    dio_value = (server.get_di(dio_name))['data'][0]
    print(f'DIO channel: {dio_name} = {dio_value}')
```

--------------------------------

### Import necessary libraries for BCS API and data handling

Source: https://github.com/als-rsoxs/bcs-install/blob/main/src/clients/Example noasyncio.ipynb

Imports essential Python libraries for interacting with the BCS API, handling data with pandas, image manipulation with PIL, and plotting with matplotlib. These are foundational for all subsequent examples.

```python
# These first two are necessary to use the API
import asyncio
import BCSz

# The rest of the imports are for the Example program
import time
import random

# for working with images
from PIL import Image
import io
from IPython.display import display # to display images

# for plotting and working with files
import pandas as pd
import matplotlib.pyplot as plt
import os
```

--------------------------------

### BCS Client (Python - Asyncio)

Source: https://github.com/als-rsoxs/bcs-install/blob/main/bcs.egg-info/SOURCES.txt

Provides an asynchronous client for interacting with the BCS service using Python's asyncio library. This is suitable for I/O-bound operations where non-blocking behavior is desired. It requires the `aiohttp` library.

```python
# src/clients/BCSz noasyncio.py

import asyncio
import aiohttp

class BCSClientAsync:
    def __init__(self, base_url):
        self.base_url = base_url
        self.session = aiohttp.ClientSession()

    async def get_data(self, endpoint):
        """Fetches data from a specified BCS endpoint asynchronously."""
        try:
            async with self.session.get(f"{self.base_url}/{endpoint}") as response:
                response.raise_for_status()
                return await response.json()
        except aiohttp.ClientError as e:
            print(f"Error fetching data from {endpoint}: {e}")
            return None

    async def post_data(self, endpoint, data):
        """Posts data to a specified BCS endpoint asynchronously."""
        try:
            async with self.session.post(f"{self.base_url}/{endpoint}", json=data) as response:
                response.raise_for_status()
                return await response.json()
        except aiohttp.ClientError as e:
            print(f"Error posting data to {endpoint}: {e}")
            return None

    async def close(self):
        """Closes the aiohttp client session."""
        await self.session.close()

async def main():
    client = BCSClientAsync("http://localhost:8000") # Replace with your BCS base URL
    
    # Example: Get some data
    data = await client.get_data("items")
    if data:
        print(f"Received data: {data}")

    # Example: Post some data
    post_result = await client.post_data("items", {"name": "new_item", "value": 100})
    if post_result:
        print(f"Post result: {post_result}")

    await client.close()

if __name__ == '__main__':
    asyncio.run(main())

```

--------------------------------

### AI Channel Operations

Source: https://github.com/als-rsoxs/bcs-install/blob/main/src/clients/Example.ipynb

Demonstrates how to interact with AI (Analog Input) channels on the BCS server, including listing channels, acquiring single-shot data, retrieving acquired data arrays, and accessing freerun data.

```APIDOC
## AI Channel Operations

### Description
Provides examples for interacting with Analog Input (AI) channels. This includes listing available AI channels, acquiring data for specified channels over a duration, retrieving the latest single-shot acquisition results, and fetching freerun data in both raw and array formats.

### Method
- `await server.list_ais()`
- `await server.acquire_data(chans=[channel_name], time=<seconds>, counts=<count>)
- `await server.get_acquired(chans=[channel_name])`
- `await server.get_acquired_array(chans=[channel_name])`
- `await server.get_freerun()`
- `await server.get_freerun_array()`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```python
async def test_ai(server: BCSz.BCSServer):
    # Display all the AI channels
    ai_names = (await server.list_ais())['names']
    [print(c) for c in ai_names]

    ai_name = 'Horiz Centroid - Video'

    if ai_name in ai_names:
        print('\nAcquiring ...')
        r = await server.acquire_data(chans=[ai_name], time=5, counts=0)

        r = await server.get_acquired(chans=[ai_name])
        print(f"Average of last single-shot acquisition for \"{ai_name}\" = {r['data'][0]}")

        r = await server.get_acquired_array(chans=[ai_name])
        print(f"Array data from last single-shot acquisition for \"{ai_name}\" = {r['chans'][0]['data']}")

        r = await server.get_freerun()
        chans = r['chans']
        print(f'\nThere were {len(chans)} channels of freerun-data retrieved.')
        for c, d in zip(chans, r['data']):
            print(f"Channel: {c:25} = {d}")

        r = await server.get_freerun_array()
        chans = r['chans']
        print(f'\nThere were {len(chans)} channels of freerun-data retrieved.')
        for d in chans:
            print(f"Channel: {d['chan']:25} = {d['data']}")
```

### Response
#### Success Response (200)
Responses vary depending on the method called. `list_ais` returns a dictionary with a 'names' key. `acquire_data` returns a status. `get_acquired` and `get_freerun` return dictionaries containing 'data' and 'chans' keys. `get_acquired_array` and `get_freerun_array` return dictionaries with 'chans' containing channel information and data.

#### Response Example
```json
{
  "names": [
    "Sine Function",
    "Horiz Centroid - Video",
    "Some Other Channel"
  ]
}
```

```json
{
  "data": [
    123.45
  ]
}
```

```json
{
  "chans": [
    {
      "chan": "Horiz Centroid - Video",
      "data": [
        120.1, 121.5, 122.3
      ]
    }
  ]
}
```

```json
{
  "chans": [
    "Sine Function",
    "Horiz Centroid - Video"
  ],
  "data": [
    0.5,
    123.45
  ]
}
```

```json
{
  "chans": [
    {
      "chan": "Sine Function",
      "data": [
        0.5, 0.6, 0.7
      ]
    },
    {
      "chan": "Horiz Centroid - Video",
      "data": [
        120.1, 121.5, 122.3
      ]
    }
  ]
}
```
```

--------------------------------

### Manage Digital Input/Output (DIO) Channels

Source: https://github.com/als-rsoxs/bcs-install/blob/main/src/clients/Example.ipynb

Demonstrates how to interact with Digital Input and Output (DIO) channels in the BCS system. It covers listing DIOs, retrieving their status and values, and controlling output channels. This is essential for hardware integration and control.

```python
async def test_dio(server: BCSz.BCSServer):
    # Don't really need dio_list, get_di does more
    dio_list = (await server.list_dios())['names']
    [print(dio) for dio in dio_list]

    # get all the dio channels, their values and if they are enabled
    r = await server.get_di()

    # make a dictionary of DIO's
    dio_dict = dict(zip(r['chans'], r['data']))
    my_dio = 'Shutter'
    if my_dio in dio_dict:
        print(f"The value of {my_dio} is {dio_dict[my_dio]}")

    # make an array of just the enabled DIO's
    dios = []
    for c, u, v in zip(r['chans'], r['enabled'], r['data']):
        if u:
            print(f"Channel: {c} is {v}")
            dios.append(c)

    # watch the DIO monitor on BCS while running the code below
    [await server.set_do(chan=c, value=True) for c in dios]  # turn all ON
    time.sleep(1)

    # walk the True through the channels
    for c_on in dios:
        for c in dios:
            r = await server.set_do(chan=c, value=bool(c == c_on))
        time.sleep(0.25)

    # flip the first DIO
    dio_name = dios[0]
    dio_value = (await server.get_di(dio_name))['data'][0]
    print(f'DIO channel: {dio_name} = {dio_value}')
    await server.set_do(chan=dio_name, value=not dio_value)
    time.sleep(0.1)  # bcs subsystem needs time to propagate the change
    dio_value = (await server.get_di(dio_name))['data'][0]
    print(f'DIO channel: {dio_name} = {dio_value}')
```

--------------------------------

### Control and Monitor Motors with Python

Source: https://github.com/als-rsoxs/bcs-install/blob/main/src/clients/Example noasyncio.ipynb

Demonstrates motor control functionalities, including listing motors and presets, checking motor status, moving motors to a goal position, and monitoring move completion. Utilizes the 'BCSz' library for server interaction and motor status interpretation.

```python
def test_motor(server: BCSz.BCSServer):

    #   Display all the motors
    rsp = server.list_motors()
    motor_names = rsp['names']
    print('\n\nThe BCS server has the following motors:')
    for m in motor_names:
        print(f"    {m}")

    # presets
    rsp = server.list_presets()
    presets_array = rsp['presets']
    print('\nPresets:')
    for p in presets_array:
        print(p)

    # check if the first preset is in-position
    if len(presets_array) > 0:
        preset_names = [p['Preset Name'] for p in presets_array]
        name = preset_names[0]
        print(f'\nAt preset {name}? ', (server.at_preset(name=name))['at_preset'])

    # Get motor info, like position, goal, status
    my_motor = 'HY'
    rsp = (server.get_motor(motors=[my_motor]))
    hy_motor_info = rsp['data'][0]
    print(f"\nget_motor gives {hy_motor_info}")

    # Use MotorStatus helper class to test individual bit-flags of the status word
    status = BCSz.MotorStatus(hy_motor_info['status'])
    print()
    print(f'Motor {my_motor} status: {status}')
    print(f'Move complete = {status.is_set(BCSz.MotorStatus.MOVE_COMPLETE)}')
    print(f'Motor on FWD Limit = {status.is_set(BCSz.MotorStatus.FORWARD_LIMIT)}')

    # get motor FULL: the whole status record
    rd = server.get_motor_full(motors=[my_motor])
    # print(json.dumps(rd['data'][0], indent=4, sort_keys=True))
    print(rd['data'][0])
    server.move_motor(motors=['HY'], goals=[75])

    # wait just a bit to let the move begin.
    time.sleep(0.250)   # this sleep call will be replaced by a new endpoint
    print()

    while True:
        rd = (server.get_motor(motors=[my_motor]))
        hy_motor_info = rd['data'][0]
        status = BCSz.MotorStatus(hy_motor_info['status'])
        print(hy_motor_info['position'], status)
        if status.is_set(BCSz.MotorStatus.MOVE_COMPLETE):
            break
        else:
            time.sleep(0.25)    # don't hit the api server constantly
```

--------------------------------

### BCS Client (Python - Synchronous)

Source: https://github.com/als-rsoxs/bcs-install/blob/main/bcs.egg-info/SOURCES.txt

Provides a synchronous client for interacting with the BCS service using standard Python requests. This is suitable for simpler applications or environments where asynchronous programming is not required. It depends on the `requests` library.

```python
# src/clients/BCSz.py

import requests

class BCSClient:
    def __init__(self, base_url):
        self.base_url = base_url

    def get_data(self, endpoint):
        """Fetches data from a specified BCS endpoint synchronously."""
        try:
            response = requests.get(f"{self.base_url}/{endpoint}")
            response.raise_for_status()  # Raise an exception for bad status codes
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"Error fetching data from {endpoint}: {e}")
            return None

    def post_data(self, endpoint, data):
        """Posts data to a specified BCS endpoint synchronously."""
        try:
            response = requests.post(f"{self.base_url}/{endpoint}", json=data)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"Error posting data to {endpoint}: {e}")
            return None

if __name__ == '__main__':
    # Example Usage:
    client = BCSClient("http://localhost:8000") # Replace with your BCS base URL

    # Example: Get some data
    data = client.get_data("items")
    if data:
        print(f"Received data: {data}")

    # Example: Post some data
    post_result = client.post_data("items", {"name": "new_item_sync", "value": 50})
    if post_result:
        print(f"Post result: {post_result}")

```

--------------------------------

### Import necessary libraries for BCS API and data handling

Source: https://github.com/als-rsoxs/bcs-install/blob/main/src/clients/Example.ipynb

Imports essential Python libraries for interacting with the BCS API, handling data with pandas, image manipulation with PIL, plotting with matplotlib, and IPython display.

```python
# These first two are necessary to use the API
import asyncio
import BCSz

# The rest of the imports are for the Example program
import time
import random

# for working with images
from PIL import Image
import io
from IPython.display import display  # to display images

# for plotting and working with files
import pandas as pd
import matplotlib.pyplot as plt
import os
```

--------------------------------

### Digital I/O Operations in Python

Source: https://context7.com/als-rsoxs/bcs-install/llms.txt

Demonstrates how to list, read, and write digital input/output channels using the BCS server. It covers reading DI values, getting specific channel states, and setting DO states. Requires an active server connection.

```python
async def digital_io_operations(server):
    # List all DIO channels
    dio_list = await server.list_dios()
    print(f"DIO channels: {dio_list['names']}")

    # Read DI values
    di_response = await server.get_di()

    for chan, enabled, value in zip(
        di_response['chans'],
        di_response['enabled'],
        di_response['data']
    ):
        if enabled:
            print(f"{chan}: {'ON' if value else 'OFF'}")

    # Read specific channel
    shutter = await server.get_di(chans=['Shutter'])
    print(f"Shutter state: {shutter['data'][0]}")

    # Set digital output
    await server.set_do(chan='Shutter', value=True)  # Open shutter
    time.sleep(0.1)  # Allow state to propagate

    await server.set_do(chan='Shutter', value=False)  # Close shutter

    # Example output:
    # DIO channels: ['Shutter', 'Beam Stop', 'Interlock', 'LED']
    # Shutter: OFF
    # Beam Stop: ON
```

--------------------------------

### Instrument Control Operations in Python

Source: https://context7.com/als-rsoxs/bcs-install/llms.txt

Provides examples for controlling various instruments like detectors and spectrometers. It shows how to list instruments, check driver status, start/stop acquisition, and retrieve acquired data (1D and 2D). Requires an active server connection and instrument availability.

```python
async def instrument_operations(server):
    # List available instruments
    instruments = await server.list_instruments()
    print(f"Instruments: {instruments['names']}")
    print(f"Displayed: {instruments['displayed']}")

    instrument_name = 'CCD Detector'

    # Check driver status
    driver_status = await server.get_instrument_driver_status(name=instrument_name)
    print(f"Driver running: {driver_status['running']}")

    # Start driver if needed
    if not driver_status['running']:
        await server.start_instrument_driver(name=instrument_name)

    # Start acquisition (waits for completion)
    acq_response = await server.start_instrument_acquire(
        name=instrument_name,
        run_type='Exposure',
        acq_time_s=1.0,  # 1 second exposure
        acq_counts=0
    )
    print(f"Acquisition took: {acq_response['elapsed_s']:.3f} seconds")

    # Get acquisition status
    status = await server.get_instrument_acquisition_status(name=instrument_name)
    print(f"Acquiring: {status['acquiring']}")
    print(f"Data available: {status['data_available']}")

    # Retrieve 1D data (spectrum)
    data_1d = await server.get_instrument_acquired1d(name=instrument_name)
    print(f"X range: {data_1d['x'][0]} to {data_1d['x'][-1]}")
    print(f"Y max: {max(data_1d['y'])}")

    # Retrieve 2D data (image)
    data_2d = await server.get_instrument_acquired2d(name=instrument_name)
    print(f"Image shape: {len(data_2d['data'])} x {len(data_2d['data'][0])}")

    # Get acquisition info
    info = await server.get_instrument_acquisition_info(name=instrument_name)
    print(f"File saved to: {info['file_name']}")
    print(f"Live time: {info['live_time']}")
    print(f"Dead time: {info['dead_time']}")

    # Stop acquisition if running
    await server.stop_instrument_acquire(name=instrument_name)
```

--------------------------------

### Acquire and retrieve AI channel data using BCS API

Source: https://github.com/als-rsoxs/bcs-install/blob/main/src/clients/Example noasyncio.ipynb

Shows how to interact with Analog Input (AI) channels on the BCS server. It includes listing available AI channels, acquiring single-shot data for a specified duration, and retrieving both averaged and array data for a given AI channel. It also demonstrates fetching freerun data in both summarized and array formats.

```python
def test_ai(server: BCSz.BCSServer):
    #   Display all the AI channels
    ai_names = (server.list_ais())['names']
    [print(c) for c in ai_names]

    # ai_name = 'Sine Function'
    ai_name = 'Horiz Centroid - Video'

    if ai_name in ai_names:
        print('\nAcquiring ...')
        r = server.acquire_data(chans=[ai_name], time=5, counts=0)
        # print(f'\nAcquire response {r}')

        r = server.get_acquired(chans=[ai_name])
        print(f"Average of last single-shot acquisition for \"{ai_name}\" = {r['data'][0]}")

        r = server.get_acquired_array(chans=[ai_name])
        print(f"Array data from last single-shot acquisition for \"{ai_name}\" = {r['chans'][0]['data']}")

        r = server.get_freerun()
        chans = r['chans']
        print(f'\nThere were {len(chans)} channels of freerun-data retrieved.')
        for c, d in zip(chans, r['data']):
            print(f"Channel: {c:25} = {d}")

        r = server.get_freerun_array()
        chans = r['chans']
        print(f'\nThere were {len(chans)} channels of freerun-data retrieved.')
        for d in chans:
            print(f"Channel: {d['chan']:25} = {d['data']}")
```