chore: removed redundant /camera/source and /beamline/imafge from server.py and test_server.py
Build and Publish / test (pull_request) Failing after 1m23s
Build and Publish / build (pull_request) Skipped
Build and Publish / Build and Deploy Docs (pull_request) Skipped
Build and Publish / test (pull_request_target) Failing after 1m25s
Build and Publish / build (pull_request_target) Skipped
Build and Publish / Build and Deploy Docs (pull_request_target) Skipped
Build and Publish / test (push) Failing after 1m26s
Build and Publish / build (push) Skipped
Build and Publish / Build and Deploy Docs (push) Skipped
Build and Publish / test (pull_request) Failing after 1m23s
Build and Publish / build (pull_request) Skipped
Build and Publish / Build and Deploy Docs (pull_request) Skipped
Build and Publish / test (pull_request_target) Failing after 1m25s
Build and Publish / build (pull_request_target) Skipped
Build and Publish / Build and Deploy Docs (pull_request_target) Skipped
Build and Publish / test (push) Failing after 1m26s
Build and Publish / build (push) Skipped
Build and Publish / Build and Deploy Docs (push) Skipped
This commit was merged in pull request #98.
This commit is contained in:
@@ -1009,27 +1009,6 @@ async def shutter(val: bool, token: str = Depends(oauth2_scheme)):
|
||||
return "OK"
|
||||
|
||||
|
||||
@app.get("/beamline/image")
|
||||
async def get_image(token: str = Depends(oauth2_scheme)):
|
||||
"""
|
||||
Get the current camera image as a JPEG stream.
|
||||
|
||||
Args:
|
||||
token: OAuth2 access token.
|
||||
|
||||
Returns:
|
||||
StreamingResponse containing the JPEG image.
|
||||
"""
|
||||
auth.check_jwt_ro(cfg, auth.parse_token(token))
|
||||
|
||||
_, encoded_image = cv2.imencode(".jpg", daq.camera_image) # Encodes the image in JPEG format
|
||||
image_bytes = io.BytesIO(
|
||||
encoded_image.tobytes()
|
||||
) # Convert OpenCV byte format to a file-like object
|
||||
|
||||
return StreamingResponse(image_bytes, media_type="image/jpeg")
|
||||
|
||||
|
||||
# TELL procedures
|
||||
@app.get("/sample/curr_sample")
|
||||
async def sample(token: str = Depends(oauth2_scheme)) -> SampleShortInfo:
|
||||
@@ -2546,21 +2525,6 @@ async def send_message_db(
|
||||
return "OK"
|
||||
|
||||
|
||||
@app.get("/camera/source")
|
||||
async def get_camera_source(token: str = Depends(oauth2_scheme)):
|
||||
"""Get the current camera image source."""
|
||||
auth.check_jwt_ro(cfg, auth.parse_token(token))
|
||||
return {"source": daq._AareDAQ__devs.samcam_source}
|
||||
|
||||
|
||||
@app.post("/camera/source")
|
||||
async def set_camera_source(use_zmq: bool, token: str = Depends(oauth2_scheme)):
|
||||
"""Set the camera image source preference."""
|
||||
auth.check_jwt_rw(cfg, auth.parse_token(token))
|
||||
daq._AareDAQ__devs.set_camera_source(use_zmq)
|
||||
return {"source": daq._AareDAQ__devs.samcam_source}
|
||||
|
||||
|
||||
@app.post("/state/maintenance")
|
||||
async def maintenance(token: str = Depends(oauth2_scheme)) -> str:
|
||||
"""
|
||||
|
||||
@@ -74,16 +74,6 @@ def test_login_success(client):
|
||||
assert response.json() == {"access_token": "fake-access-token", "token_type": "bearer"}
|
||||
|
||||
|
||||
def test_get_image(client, mock_backend):
|
||||
mock_daq = mock_backend["daq"]
|
||||
mock_daq.camera_image = np.zeros((100, 100, 3), dtype=np.uint8)
|
||||
|
||||
response = client.get("/beamline/image", headers={"Authorization": "Bearer fake-token"})
|
||||
assert response.status_code == 200
|
||||
assert response.headers["content-type"] == "image/jpeg"
|
||||
assert len(response.content) > 0
|
||||
|
||||
|
||||
def test_mount_returns_tell_exception_when_mount_precheck_fails(client, mock_backend, monkeypatch):
|
||||
from aarecommon.errors.exception_handler import TellCommunicationError
|
||||
|
||||
|
||||
Reference in New Issue
Block a user