Skip to content

Commit

Permalink
update oncat login (#135)
Browse files Browse the repository at this point in the history
  • Loading branch information
@searscr
searscr authored May 22, 2024
1 parent e3b3e56 commit db08bb3
Show file tree
Hide file tree
Showing 4 changed files with 67 additions and 455 deletions.
2 changes: 2 additions & 0 deletions environment.yml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ channels:
- conda-forge
- mantid
- oncat
- neutrons
dependencies:
- anaconda-client
- boa
Expand All @@ -21,5 +22,6 @@ dependencies:
- sphinx-rtd-theme
- python-build
- pyoncat
- pyoncatqt
- gtk3
- libstdcxx-ng
290 changes: 43 additions & 247 deletions src/shiver/views/oncat.py
View file
Original file line number Diff line number Diff line change
@@ -1,220 +1,20 @@
"""PyQt widget for the OnCat widget in General tab."""

import os
import json
import oauthlib
import numpy as np
import pyoncat
from pyoncatqt.login import ONCatLogin
from qtpy.QtWidgets import (
QGridLayout,
QGroupBox,
QPushButton,
QLabel,
QComboBox,
QDialog,
QLineEdit,
QDoubleSpinBox,
)
from qtpy.QtCore import QTimer
from shiver.configuration import get_data


class OncatToken:
"""Class to hold OnCat token"""

def __init__(self, token_path: str):
self.token_path = token_path

def read_token(self):
"""Read token from file"""
# If there is not a token stored, return
# None from the callback.
if not os.path.exists(self.token_path):
return None

with open(self.token_path, encoding="UTF-8") as storage:
return json.load(storage)

def write_token(self, token):
"""Write token to file"""
# check if directory exists
if not os.path.exists(os.path.dirname(self.token_path)):
os.makedirs(os.path.dirname(self.token_path))
# write token to file
with open(self.token_path, "w", encoding="UTF-8") as storage:
json.dump(token, storage)
# change permissions to read-only by user
os.chmod(self.token_path, 0o600)


class OnCatAgent:
"""Agent to interface with OnCat"""

def __init__(self, use_notes=False) -> None:
"""Initialize OnCat agent"""
# get configuration settings
self._use_notes = use_notes
self._oncat_url = get_data("generate_tab.oncat", "oncat_url")
self._client_id = get_data("generate_tab.oncat", "client_id")

user_home_dir = os.path.expanduser("~")
self._token = OncatToken(
os.path.abspath(f"{user_home_dir}/.shiver/oncat_token.json"),
)
self._agent = pyoncat.ONCat(
self._oncat_url,
client_id=self._client_id,
# Pass in token getter/setter callbacks here:
token_getter=self._token.read_token,
token_setter=self._token.write_token,
flow=pyoncat.RESOURCE_OWNER_CREDENTIALS_FLOW,
)

def login(self, username: str, password: str):
"""Connect to OnCat with given username and password.
Parameters
----------
username : str
Username for OnCat
password : str
Password for OnCat
"""
self._agent.login(username, password)

@property
def has_token(self):
"""Check if token exists"""
return self._token.read_token() is not None

@property
def is_connected(self):
"""Check if connected to OnCat"""
try:
self._agent.Facility.list() # pylint: disable=no-member
return True
except pyoncat.InvalidRefreshTokenError:
return False
except pyoncat.LoginRequiredError:
return False
except Exception: # pylint: disable=W0718
return False

def get_ipts(self, facility: str, instrument: str) -> list:
"""Get IPTS numbers for given facility and instrument.
Parameters
----------
facility : str
Facility name
instrument : str
Instrument name
Returns
-------
list
List of IPTS numbers in string format (IPTS-XXXXX)
"""
# return empty list if not connected
if not self.is_connected:
return []
# get list of experiments
experiments = self._agent.Experiment.list( # pylint: disable=no-member
facility=facility,
instrument=instrument,
projection=["facility", "instrument"],
)
return list({exp["id"] for exp in experiments})

def get_datasets(self, facility: str, instrument: str, ipts: int) -> list:
"""Get datasets for given facility, instrument, and IPTS.
Parameters
----------
facility : str
Facility name
instrument : str
Instrument name
ipts : int
IPTS number
Returns
-------
list
List of datasets
"""
# return empty list if not connected
if not self.is_connected:
return []
# get list of datasets
return get_dataset_names(
self._agent,
ipts_number=ipts,
instrument=instrument,
use_notes=self._use_notes,
facility=facility,
)

def get_agent_instance(self):
"""Get OnCat agent instance"""
return self._agent


class OnCatLogin(QDialog):
"""OnCat login dialog"""

def __init__(self, parent=None, error_message_callback=None):
super().__init__(parent)
self.setWindowTitle("Use U/XCAM to connect to OnCat")
self.resize(350, 200)
layout = QGridLayout()
label1 = QLabel("UserId")
self.user_obj = QLineEdit()
layout.addWidget(label1, 0, 0)
layout.addWidget(self.user_obj, 0, 1)
label2 = QLabel("Password")
self.user_pwd = QLineEdit()
layout.addWidget(label2, 1, 0)
layout.addWidget(self.user_pwd, 1, 1)
self.user_pwd.setEchoMode(QLineEdit.Password)
self.button_login = QPushButton("Login")
layout.addWidget(self.button_login, 2, 0, 2, 2)
self.setLayout(layout)

# connect signals and slots
self.button_login.clicked.connect(self.accept)

self.error_message_callback = error_message_callback

def accept(self):
"""Accept"""
# login to OnCat
try:
self.parent().oncat_agent.login(
self.user_obj.text(),
self.user_pwd.text(),
)
except oauthlib.oauth2.rfc6749.errors.InvalidGrantError:
if self.error_message_callback is not None:
self.error_message_callback(
"Invalid username or password. Please try again.",
)
self.user_pwd.setText("")
return
except pyoncat.LoginRequiredError:
if self.error_message_callback is not None:
self.error_message_callback(
"A username and/or password was not provided when logging in.",
)
self.user_pwd.setText("")
return

# ask parent to sync
self.parent().sync_with_remote()
# close dialog
self.close()


class Oncat(QGroupBox):
"""ONCat widget"""

Expand Down Expand Up @@ -258,29 +58,19 @@ def __init__(self, parent=None):
self.oncat_options_layout.addWidget(self.angle_target_label, 3, 0)
self.oncat_options_layout.addWidget(self.angle_target, 3, 1)

# status indicator (disconnected: red, connected: green)
self.status_label = QLabel("")
self.status_label.setToolTip("ONCat connection status.")
self.oncat_options_layout.addWidget(self.status_label, 4, 0)

# connect to OnCat button
self.oncat_button = QPushButton("&Connect to ONCat")
self.oncat_button.setFixedWidth(300)
self.oncat_button.setToolTip("Connect to ONCat (requires login credentials).")
self.oncat_options_layout.addWidget(self.oncat_button, 4, 1)
self.oncat_login = ONCatLogin(key="shiver", parent=self)
self.oncat_login.connection_updated.connect(self.connect_to_oncat)
self.oncat_options_layout.addWidget(self.oncat_login, 4, 0, 1, 2)

# set layout
self.setLayout(self.oncat_options_layout)

# connect signals and slots
self.oncat_button.clicked.connect(self.connect_to_oncat)

# error message callback
self.error_message_callback = None

self.use_notes = get_data("generate_tab.oncat", "use_notes")
# OnCat agent
self.oncat_agent = OnCatAgent(self.use_notes)
self.oncat_agent = self.oncat_login.get_agent_instance()

# Sync with remote
self.sync_with_remote(refresh=True)
Expand All @@ -307,7 +97,7 @@ def __init__(self, parent=None):
@property
def connected_to_oncat(self) -> bool:
"""Check if connected to OnCat"""
return self.oncat_agent.is_connected
return self.oncat_login.is_connected

def get_suggested_path(self) -> str:
"""Return a suggested path based on current selection."""
Expand All @@ -327,7 +117,7 @@ def get_suggested_selected_files(self) -> list:
group_by_angle = self.angle_target.value() > 0

return get_dataset_info(
login=self.oncat_agent.get_agent_instance(),
login=self.oncat_agent,
ipts_number=self.get_ipts_number(),
instrument=self.get_instrument(),
use_notes=self.use_notes,
Expand All @@ -343,39 +133,15 @@ def set_dataset_to_custom(self):

def connect_to_oncat(self):
"""Connect to OnCat"""
# check if connection exists
if self.connected_to_oncat:
return

# check if token exists
if self.oncat_agent.has_token:
if self.error_message_callback:
self.error_message_callback("Session expired. Please login again.")

# prompt for username and password
oncat_login = OnCatLogin(self, self.error_message_callback)
oncat_login.show()

# update connection status
self.sync_with_remote(refresh=True)

def sync_with_remote(self, refresh=False):
"""Update all items within OnCat widget."""
self.update_connection_status()

if self.connected_to_oncat and refresh:
self.update_ipts()
self.update_datasets()

def update_connection_status(self):
"""Update connection status"""
if self.oncat_agent.is_connected:
self.status_label.setText("ONCat: Connected")
self.status_label.setStyleSheet("color: green")
else:
self.status_label.setText("ONCat: Disconnected")
self.status_label.setStyleSheet("color: red")

def connect_error_callback(self, callback):
"""Connect error message callback"""
self.error_message_callback = callback
Expand Down Expand Up @@ -433,7 +199,7 @@ def get_facility(self):
def update_ipts(self):
"""Update IPTS list"""
# get IPTS list from OnCat
ipts_list = self.oncat_agent.get_ipts(
ipts_list = self.get_oncat_ipts(
self.get_facility(),
self.get_instrument(),
)
Expand All @@ -444,15 +210,45 @@ def update_ipts(self):
def update_datasets(self):
"""Update dataset list"""
# get dataset list from OnCat
dataset_list = ["custom"] + self.oncat_agent.get_datasets(
self.get_facility(),
self.get_instrument(),
self.get_ipts_number(),
)
dataset_list = []
if self.connected_to_oncat:
dataset_list = ["custom"] + get_dataset_names(
self.oncat_agent,
facility=self.get_facility(),
instrument=self.get_instrument(),
ipts_number=self.get_ipts_number(),
use_notes=self.use_notes,
)
# update dataset list
self.dataset.clear()
self.dataset.addItems(sorted(dataset_list))

def get_oncat_ipts(self, facility: str, instrument: str) -> list:
"""Get IPTS numbers for given facility and instrument.
Parameters
----------
facility : str
Facility name
instrument : str
Instrument name
Returns
-------
list
List of IPTS numbers in string format (IPTS-XXXXX)
"""
# return empty list if not connected
if not self.connected_to_oncat:
return []
# get list of experiments
experiments = self.oncat_agent.Experiment.list( # pylint: disable=no-member
facility=facility,
instrument=instrument,
projection=["facility", "instrument"],
)
return list({exp["id"] for exp in experiments})


# The following are utility functions migrated from the original
# oncat_util.py
Expand Down
5 changes: 5 additions & 0 deletions tests/conftest.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -33,3 +33,8 @@ def user_conf_file(tmp_path_factory, request):
with open(user_path, "w", encoding="utf8") as config_file:
user_config.write(config_file)
return user_path


@pytest.fixture(autouse=True)
def _get_login(monkeypatch: pytest.fixture) -> None:
monkeypatch.setattr(os, "getlogin", lambda: "test")
Loading

0 comments on commit db08bb3

Please sign in to comment.