ITQC/pydase
24
0
Fork 0
mirror of https://github.com/tiqi-group/pydase.git synced 2025-04-23 01:20:03 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #48 from tiqi-group/10-frontend-user-should-be-able-to-add-custom-display-names

Browse Source 
Feat: adds web settings file containing display name configuration
This commit is contained in:
Mose Müller 2024-01-08 17:17:06 +01:00 committed by GitHub
commit 3c0f019af8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
42 changed files with 1136 additions and 490 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

105
README.md
View File

@ -21,12 +21,16 @@
- [`NumberSlider`](#numberslider)
- [`ColouredEnum`](#colouredenum)
- [Extending with New Components](#extending-with-new-components)
- [Customizing Web Interface Style](#customizing-web-interface-style)
- [Understanding Service Persistence](#understanding-service-persistence)
- [Controlling Property State Loading with `@load_state`](#controlling-property-state-loading-with-load_state)
- [Understanding Tasks in pydase](#understanding-tasks-in-pydase)
- [Understanding Units in pydase](#understanding-units-in-pydase)
- [Changing the Log Level](#changing-the-log-level)
- [Configuring pydase via Environment Variables](#configuring-pydase-via-environment-variables)
- [Customizing the Web Interface](#customizing-the-web-interface)
- [Enhancing the Web Interface Style with Custom CSS](#enhancing-the-web-interface-style-with-custom-css)
- [Tailoring Frontend Component Layout](#tailoring-frontend-component-layout)
- [Logging in pydase](#logging-in-pydase)
- [Changing the Log Level](#changing-the-log-level)
- [Documentation](#documentation)
- [Contributing](#contributing)
- [License](#license)
@ -497,31 +501,6 @@ Users can also extend the library by creating custom components. This involves d
<!-- Component User Guide End -->
## Customizing Web Interface Style
`pydase` allows you to enhance the user experience by customizing the web interface's appearance. You can apply your own styles globally across the web interface by passing a custom CSS file to the server during initialization.
Here's how you can use this feature:
1. Prepare your custom CSS file with the desired styles.
2. When initializing your server, use the `css` parameter of the `Server` class to specify the path to your custom CSS file.
```python
from pydase import Server, DataService
class Device(DataService):
# ... your service definition ...
if __name__ == "__main__":
service = MyService()
server = Server(service, css="path/to/your/custom.css").run()
```
This will apply the styles defined in `custom.css` to the web interface, allowing you to maintain branding consistency or improve visual accessibility.
Please ensure that the CSS file path is accessible from the server's running location. Relative or absolute paths can be used depending on your setup.
## Understanding Service Persistence
`pydase` allows you to easily persist the state of your service by saving it to a file. This is especially useful when you want to maintain the service's state across different runs.
@ -676,6 +655,78 @@ if __name__ == "__main__":
For more information about what you can do with the units, please consult the documentation of [`pint`](https://pint.readthedocs.io/en/stable/).
## Configuring pydase via Environment Variables
Configuring `pydase` through environment variables enhances flexibility, security, and reusability. This approach allows for easy adaptation of services across different environments without code changes, promoting scalability and maintainability. With that, it simplifies deployment processes and facilitates centralized configuration management. Moreover, environment variables enable separation of configuration from code, aiding in secure and collaborative development.
`pydase` offers various configurable options:
- **`ENVIRONMENT`**: Sets the operation mode to either "development" or "production". Affects logging behaviour (see [logging section](#logging-in-pydase)).
- **`SERVICE_CONFIG_DIR`**: Specifies the directory for service configuration files, like `web_settings.json`. This directory can also be used to hold user-defined configuration files. Default is the `config` folder in the service root folder. The variable can be accessed through:
```python
import pydase.config
pydase.config.ServiceConfig().config_dir
```
- **`SERVICE_WEB_PORT`**: Defines the port number for the web server. This has to be different for each services running on the same host. Default is 8001.
- **`SERVICE_RPC_PORT`**: Defines the port number for the rpc server. This has to be different for each services running on the same host. Default is 18871.
- **`GENERATE_WEB_SETTINGS`**: When set to true, generates / updates the `web_settings.json` file. If the file already exists, only new entries are appended.
Some of those settings can also be altered directly in code when initializing the server:
```python
import pathlib
from pydase import Server
from your_service_module import YourService
server = Server(
YourService(),
web_port=8080,
rpc_port=18880,
config_dir=pathlib.Path("other_config_dir"), # note that you need to provide an argument of type pathlib.Path
generate_web_settings=True
).run()
```
## Customizing the Web Interface
### Enhancing the Web Interface Style with Custom CSS
`pydase` allows you to enhance the user experience by customizing the web interface's appearance. You can apply your own styles globally across the web interface by passing a custom CSS file to the server during initialization.
Here's how you can use this feature:
1. Prepare your custom CSS file with the desired styles.
2. When initializing your server, use the `css` parameter of the `Server` class to specify the path to your custom CSS file.
```python
from pydase import Server, DataService
class MyService(DataService):
# ... your service definition ...
if __name__ == "__main__":
service = MyService()
server = Server(service, css="path/to/your/custom.css").run()
```
This will apply the styles defined in `custom.css` to the web interface, allowing you to maintain branding consistency or improve visual accessibility.
Please ensure that the CSS file path is accessible from the server's running location. Relative or absolute paths can be used depending on your setup.
### Tailoring Frontend Component Layout
`pydase` enables users to customize the frontend layout via the `web_settings.json` file. Each key in the file corresponds to the full access path of public attributes, properties, and methods of the exposed service, using dot-notation.
- **Custom Display Names**: Modify the `"displayName"` value in the file to change how each component appears in the frontend.
<!-- - **Adjustable Component Order**: The `"index"` values determine the order of components. Alter these values to rearrange the components as desired. -->
The `web_settings.json` file will be stored in the directory specified by `SERVICE_CONFIG_DIR`. You can generate a `web_settings.json` file by setting the `GENERATE_WEB_SETTINGS` to `True`. For more information, see the [configuration section](#configuring-pydase-via-environment-variables).
## Logging in pydase
The `pydase` library organizes its loggers on a per-module basis, mirroring the Python package hierarchy. This structured approach allows for granular control over logging levels and behaviour across different parts of the library.

18
docs/dev-guide/Adding_Components.md
View File

@ -111,6 +111,7 @@ import { setAttribute, runMethod } from '../socket'; // use this when your comp
// or runs a method, respectively
import { DocStringComponent } from './DocStringComponent';
import React, { useEffect, useRef, useState } from 'react';
import { WebSettingsContext } from '../WebSettings';
import { Card, Collapse, Image } from 'react-bootstrap';
import { DocStringComponent } from './DocStringComponent';
import { ChevronDown, ChevronRight } from 'react-bootstrap-icons';
@ -119,7 +120,7 @@ import { LevelName } from './NotificationsComponent';
interface ImageComponentProps {
name: string;
parentPath: string;
parentPath?: string;
readOnly: boolean;
docString: string;
addNotification: (message: string, levelname?: LevelName) => void;
@ -133,9 +134,17 @@ export const ImageComponent = React.memo((props: ImageComponentProps) => {
const renderCount = useRef(0);
const [open, setOpen] = useState(true); // add this if you want to expand/collapse your component
const fullAccessPath = parentPath.concat('.' + name);
const fullAccessPath = [parentPath, name].filter((element) => element).join('.');
const id = getIdFromFullAccessPath(fullAccessPath);
// Web settings contain the user-defined display name of the components (and possibly more later)
const webSettings = useContext(WebSettingsContext);
let displayName = name;
if (webSettings[fullAccessPath] && webSettings[fullAccessPath].displayName) {
displayName = webSettings[fullAccessPath].displayName;
}
useEffect(() => {
renderCount.current++;
});
@ -156,7 +165,7 @@ export const ImageComponent = React.memo((props: ImageComponentProps) => {
onClick={() => setOpen(!open)}
style={{ cursor: 'pointer' }} // Change cursor style on hover
>
{name} {open ? <ChevronDown /> : <ChevronRight />}
{displayName} {open ? <ChevronDown /> : <ChevronRight />}
</Card.Header>
<Collapse in={open}>
<Card.Body>
@ -206,6 +215,7 @@ React components in the frontend often need to send updates to the backend, part
export const ButtonComponent = React.memo((props: ButtonComponentProps) => {
// ...
const { name, parentPath, value } = props;
let displayName = ... // to access the user-defined display name
const setChecked = (checked: boolean) => {
setAttribute(name, parentPath, checked);
@ -217,7 +227,7 @@ React components in the frontend often need to send updates to the backend, part
value={parentPath}
// ... other props
onChange={(e) => setChecked(e.currentTarget.checked)}>
<p>{name}</p>
{displayName}
</ToggleButton>
);
});

27
frontend/src/App.tsx
View File

@ -1,4 +1,4 @@
import { useCallback, useEffect, useReducer, useRef, useState } from 'react';
import { useCallback, useEffect, useReducer, useState } from 'react';
import { Navbar, Form, Offcanvas, Container } from 'react-bootstrap';
import { hostname, port, socket } from './socket';
import {
@ -13,6 +13,7 @@ import {
} from './components/NotificationsComponent';
import { ConnectionToast } from './components/ConnectionToast';
import { SerializedValue, setNestedValueByPath, State } from './utils/stateUtils';
import { WebSettingsContext, WebSetting } from './WebSettings';
type Action =
| { type: 'SET_DATA'; data: State }
@ -42,18 +43,13 @@ const reducer = (state: State, action: Action): State => {
};
const App = () => {
const [state, dispatch] = useReducer(reducer, null);
const stateRef = useRef(state); // Declare a reference to hold the current state
const [webSettings, setWebSettings] = useState<Record<string, WebSetting>>({});
const [isInstantUpdate, setIsInstantUpdate] = useState(false);
const [showSettings, setShowSettings] = useState(false);
const [showNotification, setShowNotification] = useState(false);
const [notifications, setNotifications] = useState<Notification[]>([]);
const [connectionStatus, setConnectionStatus] = useState('connecting');
// Keep the state reference up to date
useEffect(() => {
stateRef.current = state;
}, [state]);
useEffect(() => {
// Allow the user to add a custom css file
fetch(`http://${hostname}:${port}/custom.css`)
@ -74,6 +70,9 @@ const App = () => {
fetch(`http://${hostname}:${port}/service-properties`)
.then((response) => response.json())
.then((data: State) => dispatch({ type: 'SET_DATA', data }));
fetch(`http://${hostname}:${port}/web-settings`)
.then((response) => response.json())
.then((data: Record<string, WebSetting>) => setWebSettings(data));
setConnectionStatus('connected');
});
socket.on('disconnect', () => {
@ -184,12 +183,14 @@ const App = () => {
</Offcanvas>
<div className="App navbarOffset">
<DataServiceComponent
name={''}
props={state as DataServiceJSON}
isInstantUpdate={isInstantUpdate}
addNotification={addNotification}
/>
<WebSettingsContext.Provider value={webSettings}>
<DataServiceComponent
name={''}
props={state as DataServiceJSON}
isInstantUpdate={isInstantUpdate}
addNotification={addNotification}
/>
</WebSettingsContext.Provider>
</div>
<ConnectionToast connectionStatus={connectionStatus} />
</>

8
frontend/src/WebSettings.tsx Normal file
View File

@ -0,0 +1,8 @@
import { createContext } from 'react';
export const WebSettingsContext = createContext<Record<string, WebSetting>>({});
export type WebSetting = {
displayName: string;
index: number;
};

14
frontend/src/components/AsyncMethodComponent.tsx
View File

@ -1,9 +1,10 @@
import React, { useEffect, useRef } from 'react';
import React, { useContext, useEffect, useRef } from 'react';
import { runMethod } from '../socket';
import { InputGroup, Form, Button } from 'react-bootstrap';
import { DocStringComponent } from './DocStringComponent';
import { getIdFromFullAccessPath } from '../utils/stringUtils';
import { LevelName } from './NotificationsComponent';
import { WebSettingsContext } from '../WebSettings';
interface AsyncMethodProps {
name: string;
@ -19,7 +20,14 @@ export const AsyncMethodComponent = React.memo((props: AsyncMethodProps) => {
const { name, parentPath, docString, value: runningTask, addNotification } = props;
const renderCount = useRef(0);
const formRef = useRef(null);
const id = getIdFromFullAccessPath(parentPath.concat('.' + name));
const fullAccessPath = [parentPath, name].filter((element) => element).join('.');
const id = getIdFromFullAccessPath(fullAccessPath);
const webSettings = useContext(WebSettingsContext);
let displayName = name;
if (webSettings[fullAccessPath] && webSettings[fullAccessPath].displayName) {
displayName = webSettings[fullAccessPath].displayName;
}
useEffect(() => {
renderCount.current++;
@ -95,7 +103,7 @@ export const AsyncMethodComponent = React.memo((props: AsyncMethodProps) => {
<div>Render count: {renderCount.current}</div>
)}
<h5>
Function: {name}
Function: {displayName}
<DocStringComponent docString={docString} />
</h5>
<Form onSubmit={execute} ref={formRef}>

19
frontend/src/components/ButtonComponent.tsx
View File

@ -1,4 +1,5 @@
import React, { useEffect, useRef } from 'react';
import React, { useContext, useEffect, useRef } from 'react';
import { WebSettingsContext } from '../WebSettings';
import { ToggleButton } from 'react-bootstrap';
import { setAttribute } from '../socket';
import { DocStringComponent } from './DocStringComponent';
@ -16,10 +17,16 @@ interface ButtonComponentProps {
}
export const ButtonComponent = React.memo((props: ButtonComponentProps) => {
const { name, parentPath, value, readOnly, docString, mapping, addNotification } =
props;
const buttonName = mapping ? (value ? mapping[0] : mapping[1]) : name;
const id = getIdFromFullAccessPath(parentPath.concat('.' + name));
const { name, parentPath, value, readOnly, docString, addNotification } = props;
// const buttonName = props.mapping ? (value ? props.mapping[0] : props.mapping[1]) : name;
const fullAccessPath = [parentPath, name].filter((element) => element).join('.');
const id = getIdFromFullAccessPath(fullAccessPath);
const webSettings = useContext(WebSettingsContext);
let displayName = name;
if (webSettings[fullAccessPath] && webSettings[fullAccessPath].displayName) {
displayName = webSettings[fullAccessPath].displayName;
}
const renderCount = useRef(0);
@ -50,7 +57,7 @@ export const ButtonComponent = React.memo((props: ButtonComponentProps) => {
value={parentPath}
disabled={readOnly}
onChange={(e) => setChecked(e.currentTarget.checked)}>
{buttonName}
{displayName}
</ToggleButton>
</div>
);

14
frontend/src/components/ColouredEnumComponent.tsx
View File

@ -1,4 +1,5 @@
import React, { useEffect, useRef } from 'react';
import React, { useContext, useEffect, useRef } from 'react';
import { WebSettingsContext } from '../WebSettings';
import { InputGroup, Form, Row, Col } from 'react-bootstrap';
import { setAttribute } from '../socket';
import { DocStringComponent } from './DocStringComponent';
@ -26,7 +27,14 @@ export const ColouredEnumComponent = React.memo((props: ColouredEnumComponentPro
addNotification
} = props;
const renderCount = useRef(0);
const id = getIdFromFullAccessPath(parentPath.concat('.' + name));
const fullAccessPath = [parentPath, name].filter((element) => element).join('.');
const id = getIdFromFullAccessPath(fullAccessPath);
const webSettings = useContext(WebSettingsContext);
let displayName = name;
if (webSettings[fullAccessPath] && webSettings[fullAccessPath].displayName) {
displayName = webSettings[fullAccessPath].displayName;
}
useEffect(() => {
renderCount.current++;
@ -48,7 +56,7 @@ export const ColouredEnumComponent = React.memo((props: ColouredEnumComponentPro
<DocStringComponent docString={docString} />
<Row>
<Col className="d-flex align-items-center">
<InputGroup.Text>{name}</InputGroup.Text>
<InputGroup.Text>{displayName}</InputGroup.Text>
{readOnly ? (
// Display the Form.Control when readOnly is true
<Form.Control

16
frontend/src/components/DataServiceComponent.tsx
View File

@ -1,10 +1,11 @@
import { useState } from 'react';
import { useContext, useState } from 'react';
import React from 'react';
import { Card, Collapse } from 'react-bootstrap';
import { ChevronDown, ChevronRight } from 'react-bootstrap-icons';
import { Attribute, GenericComponent } from './GenericComponent';
import { getIdFromFullAccessPath } from '../utils/stringUtils';
import { LevelName } from './NotificationsComponent';
import { WebSettingsContext } from '../WebSettings';
type DataServiceProps = {
name: string;
@ -20,17 +21,24 @@ export const DataServiceComponent = React.memo(
({
name,
props,
parentPath = 'DataService',
parentPath = '',
isInstantUpdate,
addNotification
}: DataServiceProps) => {
const [open, setOpen] = useState(true);
let fullAccessPath = parentPath;
if (name) {
fullAccessPath = parentPath.concat('.' + name);
fullAccessPath = [parentPath, name].filter((element) => element).join('.');
}
const id = getIdFromFullAccessPath(fullAccessPath);
const webSettings = useContext(WebSettingsContext);
let displayName = fullAccessPath;
if (webSettings[fullAccessPath] && webSettings[fullAccessPath].displayName) {
displayName = webSettings[fullAccessPath].displayName;
}
return (
<div className="dataServiceComponent" id={id}>
<Card className="mb-3">
@ -38,7 +46,7 @@ export const DataServiceComponent = React.memo(
onClick={() => setOpen(!open)}
style={{ cursor: 'pointer' }} // Change cursor style on hover
>
{fullAccessPath} {open ? <ChevronDown /> : <ChevronRight />}
{displayName} {open ? <ChevronDown /> : <ChevronRight />}
</Card.Header>
<Collapse in={open}>
<Card.Body>

16
frontend/src/components/EnumComponent.tsx
View File

@ -1,6 +1,8 @@
import React, { useEffect, useRef } from 'react';
import React, { useContext, useEffect, useRef } from 'react';
import { WebSettingsContext } from '../WebSettings';
import { InputGroup, Form, Row, Col } from 'react-bootstrap';
import { setAttribute } from '../socket';
import { getIdFromFullAccessPath } from '../utils/stringUtils';
import { DocStringComponent } from './DocStringComponent';
import { LevelName } from './NotificationsComponent';
@ -24,6 +26,14 @@ export const EnumComponent = React.memo((props: EnumComponentProps) => {
} = props;
const renderCount = useRef(0);
const fullAccessPath = [parentPath, name].filter((element) => element).join('.');
const id = getIdFromFullAccessPath(fullAccessPath);
const webSettings = useContext(WebSettingsContext);
let displayName = name;
if (webSettings[fullAccessPath] && webSettings[fullAccessPath].displayName) {
displayName = webSettings[fullAccessPath].displayName;
}
useEffect(() => {
renderCount.current++;
@ -38,14 +48,14 @@ export const EnumComponent = React.memo((props: EnumComponentProps) => {
};
return (
<div className={'enumComponent'} id={parentPath.concat('.' + name)}>
<div className={'enumComponent'} id={id}>
{process.env.NODE_ENV === 'development' && (
<div>Render count: {renderCount.current}</div>
)}
<DocStringComponent docString={docString} />
<Row>
<Col className="d-flex align-items-center">
<InputGroup.Text>{name}</InputGroup.Text>
<InputGroup.Text>{displayName}</InputGroup.Text>
<Form.Select
aria-label="Default select example"
value={value}

1
frontend/src/components/GenericComponent.tsx
View File

@ -186,7 +186,6 @@ export const GenericComponent = React.memo(
/>
);
} else if (attribute.type === 'ColouredEnum') {
console.log(attribute);
return (
<ColouredEnumComponent
name={name}

14
frontend/src/components/ImageComponent.tsx
View File

@ -1,4 +1,5 @@
import React, { useEffect, useRef, useState } from 'react';
import React, { useContext, useEffect, useRef, useState } from 'react';
import { WebSettingsContext } from '../WebSettings';
import { Card, Collapse, Image } from 'react-bootstrap';
import { DocStringComponent } from './DocStringComponent';
import { ChevronDown, ChevronRight } from 'react-bootstrap-icons';
@ -20,7 +21,14 @@ export const ImageComponent = React.memo((props: ImageComponentProps) => {
const renderCount = useRef(0);
const [open, setOpen] = useState(true);
const id = getIdFromFullAccessPath(parentPath.concat('.' + name));
const fullAccessPath = [parentPath, name].filter((element) => element).join('.');
const id = getIdFromFullAccessPath(fullAccessPath);
const webSettings = useContext(WebSettingsContext);
let displayName = name;
if (webSettings[fullAccessPath] && webSettings[fullAccessPath].displayName) {
displayName = webSettings[fullAccessPath].displayName;
}
useEffect(() => {
renderCount.current++;
@ -40,7 +48,7 @@ export const ImageComponent = React.memo((props: ImageComponentProps) => {
onClick={() => setOpen(!open)}
style={{ cursor: 'pointer' }} // Change cursor style on hover
>
{name} {open ? <ChevronDown /> : <ChevronRight />}
{displayName} {open ? <ChevronDown /> : <ChevronRight />}
</Card.Header>
<Collapse in={open}>
<Card.Body>

3
frontend/src/components/ListComponent.tsx
View File

@ -18,7 +18,8 @@ export const ListComponent = React.memo((props: ListComponentProps) => {
props;
const renderCount = useRef(0);
const id = getIdFromFullAccessPath(parentPath.concat('.' + name));
const fullAccessPath = [parentPath, name].filter((element) => element).join('.');
const id = getIdFromFullAccessPath(fullAccessPath);
useEffect(() => {
renderCount.current++;

14
frontend/src/components/MethodComponent.tsx
View File

@ -1,4 +1,5 @@
import React, { useState, useEffect, useRef } from 'react';
import React, { useContext, useEffect, useRef, useState } from 'react';
import { WebSettingsContext } from '../WebSettings';
import { runMethod } from '../socket';
import { Button, InputGroup, Form, Collapse } from 'react-bootstrap';
import { DocStringComponent } from './DocStringComponent';
@ -21,7 +22,14 @@ export const MethodComponent = React.memo((props: MethodProps) => {
const [hideOutput, setHideOutput] = useState(false);
// Add a new state variable to hold the list of function calls
const [functionCalls, setFunctionCalls] = useState([]);
const id = getIdFromFullAccessPath(parentPath.concat('.' + name));
const fullAccessPath = [parentPath, name].filter((element) => element).join('.');
const id = getIdFromFullAccessPath(fullAccessPath);
const webSettings = useContext(WebSettingsContext);
let displayName = name;
if (webSettings[fullAccessPath] && webSettings[fullAccessPath].displayName) {
displayName = webSettings[fullAccessPath].displayName;
}
useEffect(() => {
renderCount.current++;
@ -80,7 +88,7 @@ export const MethodComponent = React.memo((props: MethodProps) => {
<div>Render count: {renderCount.current}</div>
)}
<h5 onClick={() => setHideOutput(!hideOutput)} style={{ cursor: 'pointer' }}>
Function: {name}
Function: {displayName}
<DocStringComponent docString={docString} />
</h5>
<Form onSubmit={execute}>

13
frontend/src/components/NumberComponent.tsx
View File

@ -1,4 +1,5 @@
import React, { useEffect, useRef, useState } from 'react';
import React, { useContext, useEffect, useRef, useState } from 'react';
import { WebSettingsContext } from '../WebSettings';
import { Form, InputGroup } from 'react-bootstrap';
import { setAttribute } from '../socket';
import { DocStringComponent } from './DocStringComponent';
@ -146,8 +147,14 @@ export const NumberComponent = React.memo((props: NumberComponentProps) => {
const [cursorPosition, setCursorPosition] = useState(null);
// Create a state for the input string
const [inputString, setInputString] = useState(props.value.toString());
const fullAccessPath = parentPath.concat('.' + name);
const fullAccessPath = [parentPath, name].filter((element) => element).join('.');
const id = getIdFromFullAccessPath(fullAccessPath);
const webSettings = useContext(WebSettingsContext);
let displayName = name;
if (webSettings[fullAccessPath] && webSettings[fullAccessPath].displayName) {
displayName = webSettings[fullAccessPath].displayName;
}
useEffect(() => {
renderCount.current++;
@ -309,7 +316,7 @@ export const NumberComponent = React.memo((props: NumberComponentProps) => {
<DocStringComponent docString={docString} />
<div className="d-flex">
<InputGroup>
{showName && <InputGroup.Text>{name}</InputGroup.Text>}
{showName && <InputGroup.Text>{displayName}</InputGroup.Text>}
<Form.Control
type="text"
value={inputString}

13
frontend/src/components/SliderComponent.tsx
View File

@ -1,4 +1,5 @@
import React, { useEffect, useRef, useState } from 'react';
import React, { useContext, useEffect, useRef, useState } from 'react';
import { WebSettingsContext } from '../WebSettings';
import { InputGroup, Form, Row, Col, Collapse, ToggleButton } from 'react-bootstrap';
import { setAttribute } from '../socket';
import { DocStringComponent } from './DocStringComponent';
@ -34,8 +35,14 @@ export const SliderComponent = React.memo((props: SliderComponentProps) => {
isInstantUpdate,
addNotification
} = props;
const fullAccessPath = parentPath.concat('.' + name);
const fullAccessPath = [parentPath, name].filter((element) => element).join('.');
const id = getIdFromFullAccessPath(fullAccessPath);
const webSettings = useContext(WebSettingsContext);
let displayName = name;
if (webSettings[fullAccessPath] && webSettings[fullAccessPath].displayName) {
displayName = webSettings[fullAccessPath].displayName;
}
useEffect(() => {
renderCount.current++;
@ -101,7 +108,7 @@ export const SliderComponent = React.memo((props: SliderComponentProps) => {
<DocStringComponent docString={docString} />
<Row>
<Col xs="auto" xl="auto">
<InputGroup.Text>{name}</InputGroup.Text>
<InputGroup.Text>{displayName}</InputGroup.Text>
</Col>
<Col xs="5" xl>
<Slider

13
frontend/src/components/StringComponent.tsx
View File

@ -1,10 +1,11 @@
import React, { useEffect, useRef, useState } from 'react';
import React, { useContext, useEffect, useRef, useState } from 'react';
import { Form, InputGroup } from 'react-bootstrap';
import { setAttribute } from '../socket';
import { DocStringComponent } from './DocStringComponent';
import '../App.css';
import { getIdFromFullAccessPath } from '../utils/stringUtils';
import { LevelName } from './NotificationsComponent';
import { WebSettingsContext } from '../WebSettings';
// TODO: add button functionality
@ -24,8 +25,14 @@ export const StringComponent = React.memo((props: StringComponentProps) => {
const renderCount = useRef(0);
const [inputString, setInputString] = useState(props.value);
const fullAccessPath = parentPath.concat('.' + name);
const fullAccessPath = [parentPath, name].filter((element) => element).join('.');
const id = getIdFromFullAccessPath(fullAccessPath);
const webSettings = useContext(WebSettingsContext);
let displayName = name;
if (webSettings[fullAccessPath] && webSettings[fullAccessPath].displayName) {
displayName = webSettings[fullAccessPath].displayName;
}
useEffect(() => {
renderCount.current++;
@ -65,7 +72,7 @@ export const StringComponent = React.memo((props: StringComponentProps) => {
)}
<DocStringComponent docString={docString} />
<InputGroup>
<InputGroup.Text>{name}</InputGroup.Text>
<InputGroup.Text>{displayName}</InputGroup.Text>
<Form.Control
type="text"
value={inputString}

18
frontend/src/utils/stringUtils.ts
View File

@ -1,12 +1,16 @@
export function getIdFromFullAccessPath(fullAccessPath: string) {
// Replace '].' with a single dash
let id = fullAccessPath.replace(/\]\./g, '-');
if (fullAccessPath) {
// Replace '].' with a single dash
let id = fullAccessPath.replace(/\]\./g, '-');
// Replace any character that is not a word character or underscore with a dash
id = id.replace(/[^\w_]+/g, '-');
// Replace any character that is not a word character or underscore with a dash
id = id.replace(/[^\w_]+/g, '-');
// Remove any trailing dashes
id = id.replace(/-+$/, '');
// Remove any trailing dashes
id = id.replace(/-+$/, '');
return id;
return id;
} else {
return 'main';
}
}

20
poetry.lock generated
View File

@ -1217,6 +1217,24 @@ tomli = {version = ">=1.0.0", markers = "python_version < \"3.11\""}
[package.extras]
testing = ["argcomplete", "attrs (>=19.2.0)", "hypothesis (>=3.56)", "mock", "nose", "pygments (>=2.7.2)", "requests", "setuptools", "xmlschema"]
[[package]]
name = "pytest-asyncio"
version = "0.23.2"
description = "Pytest support for asyncio"
optional = false
python-versions = ">=3.8"
files = [
{file = "pytest-asyncio-0.23.2.tar.gz", hash = "sha256:c16052382554c7b22d48782ab3438d5b10f8cf7a4bdcae7f0f67f097d95beecc"},
{file = "pytest_asyncio-0.23.2-py3-none-any.whl", hash = "sha256:ea9021364e32d58f0be43b91c6233fb8d2224ccef2398d6837559e587682808f"},
]
[package.dependencies]
pytest = ">=7.0.0"
[package.extras]
docs = ["sphinx (>=5.3)", "sphinx-rtd-theme (>=1.0)"]
testing = ["coverage (>=6.2)", "hypothesis (>=5.7.1)"]
[[package]]
name = "pytest-cov"
version = "4.1.0"
@ -1714,4 +1732,4 @@ h11 = ">=0.9.0,<1"
[metadata]
lock-version = "2.0"
python-versions = "^3.10"
content-hash = "817578b5a679df165d8a01cc67aab7aecf1bd90bdb2181b946cdd4f123328bdc"
content-hash = "5517c75ad8c7968145c6883d7a0035e5b3b6c839fc8ac69e7082a96c9d646b67"

1
pyproject.toml
View File

@ -31,6 +31,7 @@ matplotlib = "^3.7.2"
pyright = "^1.1.323"
pytest-mock = "^3.11.1"
ruff = "^0.1.5"
pytest-asyncio = "^0.23.2"
[tool.poetry.group.docs]
optional = true

15
src/pydase/config.py
View File

@ -1,3 +1,4 @@
from pathlib import Path
from typing import Literal
from confz import BaseConfig, EnvSource
@ -7,3 +8,17 @@ class OperationMode(BaseConfig): # type: ignore[misc]
environment: Literal["development", "production"] = "development"
CONFIG_SOURCES = EnvSource(allow=["ENVIRONMENT"])
class ServiceConfig(BaseConfig): # type: ignore[misc]
config_dir: Path = Path("config")
web_port: int = 8001
rpc_port: int = 18871
CONFIG_SOURCES = EnvSource(prefix="SERVICE_")
class WebServerConfig(BaseConfig): # type: ignore[misc]
generate_web_settings: bool = False
CONFIG_SOURCES = EnvSource(allow=["GENERATE_WEB_SETTINGS"])

24
src/pydase/data_service/state_manager.py
View File

@ -9,12 +9,14 @@ import pydase.units as u
from pydase.data_service.data_service_cache import DataServiceCache
from pydase.utils.helpers import (
get_object_attr_from_path_list,
is_property_attribute,
parse_list_attr_and_index,
)
from pydase.utils.serializer import (
dump,
generate_serialized_data_paths,
get_nested_dict_by_path,
serialized_dict_is_nested_object,
)
if TYPE_CHECKING:
@ -246,7 +248,7 @@ class StateManager:
else:
setattr(target_obj, attr_name, value)
def __is_loadable_state_attribute(self, property_path: str) -> bool:
def __is_loadable_state_attribute(self, full_access_path: str) -> bool:
"""Checks if an attribute defined by a dot-separated path should be loaded from
storage.
@ -255,13 +257,12 @@ class StateManager:
"""
parent_object = get_object_attr_from_path_list(
self.service, property_path.split(".")[:-1]
self.service, full_access_path.split(".")[:-1]
)
attr_name = property_path.split(".")[-1]
attr_name = full_access_path.split(".")[-1]
prop = getattr(type(parent_object), attr_name, None)
if isinstance(prop, property):
if is_property_attribute(parent_object, attr_name):
prop = getattr(type(parent_object), attr_name)
has_decorator = has_load_state_decorator(prop)
if not has_decorator:
logger.debug(
@ -270,4 +271,13 @@ class StateManager:
attr_name,
)
return has_decorator
return True
cached_serialization_dict = get_nested_dict_by_path(
self.cache, full_access_path
)
if cached_serialization_dict["value"] == "method":
return False
# nested objects cannot be loaded
return not serialized_dict_is_nested_object(cached_serialization_dict)

7
src/pydase/data_service/task_manager.py
View File

@ -3,7 +3,6 @@ from __future__ import annotations
import asyncio
import inspect
import logging
from functools import wraps
from typing import TYPE_CHECKING, Any, TypedDict
from pydase.data_service.abstract_data_service import AbstractDataService
@ -78,7 +77,6 @@ class TaskManager:
def __init__(self, service: DataService) -> None:
self.service = service
self._loop = asyncio.get_event_loop()
self.tasks: dict[str, TaskDict] = {}
"""A dictionary to keep track of running tasks. The keys are the names of the
@ -88,6 +86,10 @@ class TaskManager:
self._set_start_and_stop_for_async_methods()
@property
def _loop(self) -> asyncio.AbstractEventLoop:
return asyncio.get_running_loop()
def _set_start_and_stop_for_async_methods(self) -> None:
# inspect the methods of the class
for name, method in inspect.getmembers(
@ -154,7 +156,6 @@ class TaskManager:
method (callable): The coroutine to be turned into an asyncio task.
"""
@wraps(method)
def start_task(*args: Any, **kwargs: Any) -> None:
def task_done_callback(task: asyncio.Task[None], name: str) -> None:
"""Handles tasks that have finished.

6
src/pydase/frontend/asset-manifest.json
View File

@ -1,13 +1,13 @@
{
"files": {
"main.css": "/static/css/main.2d8458eb.css",
"main.js": "/static/js/main.da9f921a.js",
"main.js": "/static/js/main.ea55bba6.js",
"index.html": "/index.html",
"main.2d8458eb.css.map": "/static/css/main.2d8458eb.css.map",
"main.da9f921a.js.map": "/static/js/main.da9f921a.js.map"
"main.ea55bba6.js.map": "/static/js/main.ea55bba6.js.map"
},
"entrypoints": [
"static/css/main.2d8458eb.css",
"static/js/main.da9f921a.js"
"static/js/main.ea55bba6.js"
]
}

2
src/pydase/frontend/index.html
View File

@ -1 +1 @@
<!doctype html><html lang="en"><head><meta charset="utf-8"/><link rel="icon" href="/favicon.ico"/><meta name="viewport" content="width=device-width,initial-scale=1"/><meta name="theme-color" content="#000000"/><meta name="description" content="Web site displaying a pydase UI."/><link rel="apple-touch-icon" href="/logo192.png"/><link rel="manifest" href="/manifest.json"/><title>pydase App</title><script defer="defer" src="/static/js/main.da9f921a.js"></script><link href="/static/css/main.2d8458eb.css" rel="stylesheet"></head><body><noscript>You need to enable JavaScript to run this app.</noscript><div id="root"></div></body></html>
<!doctype html><html lang="en"><head><meta charset="utf-8"/><link rel="icon" href="/favicon.ico"/><meta name="viewport" content="width=device-width,initial-scale=1"/><meta name="theme-color" content="#000000"/><meta name="description" content="Web site displaying a pydase UI."/><link rel="apple-touch-icon" href="/logo192.png"/><link rel="manifest" href="/manifest.json"/><title>pydase App</title><script defer="defer" src="/static/js/main.ea55bba6.js"></script><link href="/static/css/main.2d8458eb.css" rel="stylesheet"></head><body><noscript>You need to enable JavaScript to run this app.</noscript><div id="root"></div></body></html>

1
src/pydase/frontend/static/js/main.da9f921a.js.map
View File

File diff suppressed because one or more lines are too long

6
src/pydase/frontend/static/js/main.da9f921a.js → src/pydase/frontend/static/js/main.ea55bba6.js
View File

File diff suppressed because one or more lines are too long

0
src/pydase/frontend/static/js/main.da9f921a.js.LICENSE.txt → src/pydase/frontend/static/js/main.ea55bba6.js.LICENSE.txt
View File

1
src/pydase/frontend/static/js/main.ea55bba6.js.map Normal file
View File

File diff suppressed because one or more lines are too long

6
src/pydase/server/__init__.py
View File

@ -1,3 +1,7 @@
from pydase.server.server import Server
from pydase.server.web_server.web_server import WebServer
__all__ = ["Server"]
__all__ = [
"Server",
"WebServer",
]

257
src/pydase/server/server.py
View File

@ -8,16 +8,14 @@ from pathlib import Path
from types import FrameType
from typing import Any, Protocol, TypedDict
import uvicorn
from rpyc import ForkingServer, ThreadedServer # type: ignore[import-untyped]
from rpyc import ThreadedServer # type: ignore[import-untyped]
from uvicorn.server import HANDLED_SIGNALS
from pydase import DataService
from pydase.config import ServiceConfig
from pydase.data_service.data_service_observer import DataServiceObserver
from pydase.data_service.state_manager import StateManager
from pydase.utils.serializer import dump
from .web_server import WebAPI
from pydase.server.web_server import WebServer
logger = logging.getLogger(__name__)
@ -31,35 +29,27 @@ class AdditionalServerProtocol(Protocol):
any server implementing it should have an __init__ method for initialization and a
serve method for starting the server.
Parameters:
-----------
service: DataService
The instance of DataService that the server will use. This could be the main
application or a specific service that the server will provide.
port: int
The port number at which the server will be accessible. This should be a valid
port number, typically in the range 1024-65535.
host: str
The hostname or IP address at which the server will be hosted. This could be a
local address (like '127.0.0.1' for localhost) or a public IP address.
state_manager: StateManager
The state manager managing the state cache and persistence of the exposed
service.
**kwargs: Any
Any additional parameters required for initializing the server. These parameters
are specific to the server's implementation.
Args:
data_service_observer:
Observer for the DataService, handling state updates and communication to
connected clients through injected callbacks. Can be utilized to access the
service and state manager, and to add custom state-update callbacks.
host:
Hostname or IP address where the server is accessible. Commonly '0.0.0.0' to
bind to all network interfaces.
port:
Port number on which the server listens. Typically in the range 1024-65535
(non-standard ports).
**kwargs:
Any additional parameters required for initializing the server. These
parameters are specific to the server's implementation.
"""
def __init__(
self,
service: DataService,
port: int,
data_service_observer: DataServiceObserver,
host: str,
state_manager: StateManager,
port: int,
**kwargs: Any,
) -> None:
...
@ -89,87 +79,86 @@ class Server:
"""
The `Server` class provides a flexible server implementation for the `DataService`.
Parameters:
-----------
service: DataService
The DataService instance that this server will manage.
host: str
The host address for the server. Default is '0.0.0.0', which means all available
network interfaces.
rpc_port: int
The port number for the RPC server. Default is 18871.
web_port: int
The port number for the web server. Default is 8001.
enable_rpc: bool
Whether to enable the RPC server. Default is True.
enable_web: bool
Whether to enable the web server. Default is True.
filename: str | Path | None
Filename of the file managing the service state persistence. Defaults to None.
use_forking_server: bool
Whether to use ForkingServer for multiprocessing. Default is False.
additional_servers : list[AdditionalServer]
A list of additional servers to run alongside the main server. Each entry in the
list should be a dictionary with the following structure:
Args:
service: DataService
The DataService instance that this server will manage.
host: str
The host address for the server. Default is '0.0.0.0', which means all
available network interfaces.
rpc_port: int
The port number for the RPC server. Default is
`pydase.config.ServiceConfig().rpc_port`.
web_port: int
The port number for the web server. Default is
`pydase.config.ServiceConfig().web_port`.
enable_rpc: bool
Whether to enable the RPC server. Default is True.
enable_web: bool
Whether to enable the web server. Default is True.
filename: str | Path | None
Filename of the file managing the service state persistence. Defaults to None.
use_forking_server: bool
Whether to use ForkingServer for multiprocessing. Default is False.
additional_servers : list[AdditionalServer]
A list of additional servers to run alongside the main server. Each entry in
the list should be a dictionary with the following structure:
- server: A class that adheres to the AdditionalServerProtocol. This class
should have an `__init__` method that accepts the DataService instance,
port, host, and optional keyword arguments, and a `serve` method that is
a coroutine responsible for starting the server.
- port: The port on which the additional server will be running.
- kwargs: A dictionary containing additional keyword arguments that will be
passed to the server's `__init__` method.
- server: A class that adheres to the AdditionalServerProtocol. This class
should have an `__init__` method that accepts the DataService instance,
port, host, and optional keyword arguments, and a `serve` method that is a
coroutine responsible for starting the server.
- port: The port on which the additional server will be running.
- kwargs: A dictionary containing additional keyword arguments that will be
passed to the server's `__init__` method.
Here's an example of how you might define an additional server:
Here's an example of how you might define an additional server:
>>> class MyCustomServer:
... def __init__(
... self,
... service: DataService,
... port: int,
... host: str,
... state_manager: StateManager,
... **kwargs: Any
... ):
... self.service = service
... self.state_manager = state_manager
... self.port = port
... self.host = host
... # handle any additional arguments...
...
... async def serve(self):
... # code to start the server...
>>> class MyCustomServer:
... def __init__(
... self,
... data_service_observer: DataServiceObserver,
... host: str,
... port: int,
... **kwargs: Any,
... ) -> None:
... self.observer = data_service_observer
... self.state_manager = self.observer.state_manager
... self.service = self.state_manager.service
... self.port = port
... self.host = host
... # handle any additional arguments...
...
... async def serve(self):
... # code to start the server...
And here's how you might add it to the `additional_servers` list when creating a
`Server` instance:
And here's how you might add it to the `additional_servers` list when creating
a `Server` instance:
>>> server = Server(
... service=my_data_service,
... additional_servers=[
... {
... "server": MyCustomServer,
... "port": 12345,
... "kwargs": {"some_arg": "some_value"}
... }
... ],
... )
... server.run()
>>> server = Server(
... service=my_data_service,
... additional_servers=[
... {
... "server": MyCustomServer,
... "port": 12345,
... "kwargs": {"some_arg": "some_value"}
... }
... ],
... )
... server.run()
**kwargs: Any
Additional keyword arguments.
**kwargs: Any
Additional keyword arguments.
"""
def __init__( # noqa: PLR0913
self,
service: DataService,
host: str = "0.0.0.0",
rpc_port: int = 18871,
web_port: int = 8001,
rpc_port: int = ServiceConfig().rpc_port,
web_port: int = ServiceConfig().web_port,
enable_rpc: bool = True,
enable_web: bool = True,
filename: str | Path | None = None,
use_forking_server: bool = False,
additional_servers: list[AdditionalServer] | None = None,
**kwargs: Any,
) -> None:
@ -183,7 +172,6 @@ class Server:
self._enable_web = enable_web
self._kwargs = kwargs
self._loop: asyncio.AbstractEventLoop
self._rpc_server_type = ForkingServer if use_forking_server else ThreadedServer
self._additional_servers = additional_servers
self.should_exit = False
self.servers: dict[str, asyncio.Future[Any]] = {}
@ -199,23 +187,8 @@ class Server:
Initializes the asyncio event loop and starts the server.
This method should be called to start the server after it's been instantiated.
Raises
------
Exception
If there's an error while running the server, the error will be propagated
after the server is shut down.
"""
try:
self._loop = asyncio.get_event_loop()
except RuntimeError:
self._loop = asyncio.new_event_loop()
asyncio.set_event_loop(self._loop)
try:
self._loop.run_until_complete(self.serve())
except Exception:
self._loop.run_until_complete(self.shutdown())
raise
asyncio.run(self.serve())
async def serve(self) -> None:
process_id = os.getpid()
@ -230,7 +203,7 @@ class Server:
logger.info("Finished server process [%s]", process_id)
async def startup(self) -> None: # noqa: C901
async def startup(self) -> None:
self._loop = asyncio.get_running_loop()
self._loop.set_exception_handler(self.custom_exception_handler)
self.install_signal_handlers()
@ -238,7 +211,7 @@ class Server:
if self._enable_rpc:
self.executor = ThreadPoolExecutor()
self._rpc_server = self._rpc_server_type(
self._rpc_server = ThreadedServer(
self._service,
port=self._rpc_port,
protocol_config={
@ -252,10 +225,9 @@ class Server:
self.servers["rpyc"] = future_or_task
for server in self._additional_servers:
addin_server = server["server"](
self._service,
port=server["port"],
data_service_observer=self._observer,
host=self._host,
state_manager=self._state_manager,
port=server["port"],
**server["kwargs"],
)
@ -266,49 +238,13 @@ class Server:
future_or_task = self._loop.create_task(addin_server.serve())
self.servers[server_name] = future_or_task
if self._enable_web:
self._wapi = WebAPI(
service=self._service,
state_manager=self._state_manager,
self._web_server = WebServer(
data_service_observer=self._observer,
host=self._host,
port=self._web_port,
**self._kwargs,
)
web_server = uvicorn.Server(
uvicorn.Config(
self._wapi.fastapi_app, host=self._host, port=self._web_port
)
)
def sio_callback(
full_access_path: str, value: Any, cached_value_dict: dict[str, Any]
) -> None:
if cached_value_dict != {}:
serialized_value = dump(value)
if cached_value_dict["type"] != "method":
cached_value_dict["type"] = serialized_value["type"]
cached_value_dict["value"] = serialized_value["value"]
async def notify() -> None:
try:
await self._wapi.sio.emit(
"notify",
{
"data": {
"full_access_path": full_access_path,
"value": cached_value_dict,
}
},
)
except Exception as e:
logger.warning("Failed to send notification: %s", e)
self._loop.create_task(notify())
self._observer.add_notification_callback(sio_callback)
# overwrite uvicorn's signal handlers, otherwise it will bogart SIGINT and
# SIGTERM, which makes it impossible to escape out of
web_server.install_signal_handlers = lambda: None # type: ignore[method-assign]
future_or_task = self._loop.create_task(web_server.serve())
future_or_task = self._loop.create_task(self._web_server.serve())
self.servers["web"] = future_or_task
async def main_loop(self) -> None:
@ -319,8 +255,7 @@ class Server:
logger.info("Shutting down")
logger.info("Saving data to %s.", self._state_manager.filename)
if self._state_manager is not None:
self._state_manager.save_state()
self._state_manager.save_state()
await self.__cancel_servers()
await self.__cancel_tasks()
@ -382,7 +317,7 @@ class Server:
async def emit_exception() -> None:
try:
await self._wapi.sio.emit(
await self._web_server._sio.emit(
"exception",
{
"data": {

175
src/pydase/server/web_server.py
View File

@ -1,175 +0,0 @@
import logging
from pathlib import Path
from typing import Any, TypedDict
import socketio # type: ignore[import-untyped]
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
from pydase import DataService
from pydase.data_service.data_service import process_callable_attribute
from pydase.data_service.state_manager import StateManager
from pydase.utils.helpers import get_object_attr_from_path_list
from pydase.utils.logging import SocketIOHandler
from pydase.version import __version__
logger = logging.getLogger(__name__)
class UpdateDict(TypedDict):
"""
A TypedDict subclass representing a dictionary used for updating attributes in a
DataService.
Attributes:
----------
name : str
The name of the attribute to be updated in the DataService instance.
If the attribute is part of a nested structure, this would be the name of the
attribute in the last nested object. For example, for an attribute access path
'attr1.list_attr[0].attr2', 'attr2' would be the name.
parent_path : str
The access path for the parent object of the attribute to be updated. This is
used to construct the full access path for the attribute. For example, for an
attribute access path 'attr1.list_attr[0].attr2', 'attr1.list_attr[0]' would be
the parent_path.
value : Any
The new value to be assigned to the attribute. The type of this value should
match the type of the attribute to be updated.
"""
name: str
parent_path: str
value: Any
class RunMethodDict(TypedDict):
"""
A TypedDict subclass representing a dictionary used for running methods from the
exposed DataService.
Attributes:
name (str): The name of the method to be run.
parent_path (str): The access path for the parent object of the method to be
run. This is used to construct the full access path for the method. For
example, for an method with access path 'attr1.list_attr[0].method_name',
'attr1.list_attr[0]' would be the parent_path.
kwargs (dict[str, Any]): The arguments passed to the method.
"""
name: str
parent_path: str
kwargs: dict[str, Any]
class WebAPI:
__sio_app: socketio.ASGIApp
__fastapi_app: FastAPI
def __init__( # noqa: PLR0913
self,
service: DataService,
state_manager: StateManager,
frontend: str | Path | None = None,
css: str | Path | None = None,
enable_cors: bool = True,
*args: Any,
**kwargs: Any,
) -> None:
self.service = service
self.state_manager = state_manager
self.frontend = frontend
self.css = css
self.enable_cors = enable_cors
self.args = args
self.kwargs = kwargs
self.setup_socketio()
self.setup_fastapi_app()
self.setup_logging_handler()
def setup_logging_handler(self) -> None:
logger = logging.getLogger()
logger.addHandler(SocketIOHandler(self.__sio))
def setup_socketio(self) -> None:
# the socketio ASGI app, to notify clients when params update
if self.enable_cors:
sio = socketio.AsyncServer(async_mode="asgi", cors_allowed_origins="*")
else:
sio = socketio.AsyncServer(async_mode="asgi")
@sio.event
def set_attribute(sid: str, data: UpdateDict) -> Any:
logger.debug("Received frontend update: %s", data)
path_list = [*data["parent_path"].split("."), data["name"]]
path_list.remove("DataService") # always at the start, does not do anything
path = ".".join(path_list)
return self.state_manager.set_service_attribute_value_by_path(
path=path, value=data["value"]
)
@sio.event
def run_method(sid: str, data: RunMethodDict) -> Any:
logger.debug("Running method: %s", data)
path_list = [*data["parent_path"].split("."), data["name"]]
path_list.remove("DataService") # always at the start, does not do anything
method = get_object_attr_from_path_list(self.service, path_list)
return process_callable_attribute(method, data["kwargs"])
self.__sio = sio
self.__sio_app = socketio.ASGIApp(self.__sio)
def setup_fastapi_app(self) -> None:
app = FastAPI()
if self.enable_cors:
app.add_middleware(
CORSMiddleware,
allow_credentials=True,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
app.mount("/ws", self.__sio_app)
@app.get("/version")
def version() -> str:
return __version__
@app.get("/name")
def name() -> str:
return self.service.get_service_name()
@app.get("/service-properties")
def service_properties() -> dict[str, Any]:
return self.state_manager.cache
# exposing custom.css file provided by user
if self.css is not None:
@app.get("/custom.css")
async def styles() -> FileResponse:
return FileResponse(str(self.css))
app.mount(
"/",
StaticFiles(
directory=Path(__file__).parent.parent / "frontend",
html=True,
),
)
self.__fastapi_app = app
@property
def sio(self) -> socketio.AsyncServer:
return self.__sio
@property
def fastapi_app(self) -> FastAPI:
return self.__fastapi_app

3
src/pydase/server/web_server/__init__.py Normal file
View File

@ -0,0 +1,3 @@
from pydase.server.web_server.web_server import WebServer
__all__ = ["WebServer"]

149
src/pydase/server/web_server/sio_setup.py Normal file
View File

@ -0,0 +1,149 @@
import asyncio
import logging
from typing import Any, TypedDict
import socketio # type: ignore[import-untyped]
from pydase.data_service.data_service import process_callable_attribute
from pydase.data_service.data_service_observer import DataServiceObserver
from pydase.data_service.state_manager import StateManager
from pydase.utils.helpers import get_object_attr_from_path_list
from pydase.utils.logging import SocketIOHandler
from pydase.utils.serializer import dump
logger = logging.getLogger(__name__)
class UpdateDict(TypedDict):
"""
A TypedDict subclass representing a dictionary used for updating attributes in a
DataService.
Attributes:
----------
name : str
The name of the attribute to be updated in the DataService instance.
If the attribute is part of a nested structure, this would be the name of the
attribute in the last nested object. For example, for an attribute access path
'attr1.list_attr[0].attr2', 'attr2' would be the name.
parent_path : str
The access path for the parent object of the attribute to be updated. This is
used to construct the full access path for the attribute. For example, for an
attribute access path 'attr1.list_attr[0].attr2', 'attr1.list_attr[0]' would be
the parent_path.
value : Any
The new value to be assigned to the attribute. The type of this value should
match the type of the attribute to be updated.
"""
name: str
parent_path: str
value: Any
class RunMethodDict(TypedDict):
"""
A TypedDict subclass representing a dictionary used for running methods from the
exposed DataService.
Attributes:
name (str): The name of the method to be run.
parent_path (str): The access path for the parent object of the method to be
run. This is used to construct the full access path for the method. For
example, for an method with access path 'attr1.list_attr[0].method_name',
'attr1.list_attr[0]' would be the parent_path.
kwargs (dict[str, Any]): The arguments passed to the method.
"""
name: str
parent_path: str
kwargs: dict[str, Any]
def setup_sio_server(
observer: DataServiceObserver,
enable_cors: bool,
loop: asyncio.AbstractEventLoop,
) -> socketio.AsyncServer:
"""
Sets up and configures a Socket.IO asynchronous server.
Args:
observer (DataServiceObserver):
The observer managing state updates and communication.
enable_cors (bool):
Flag indicating whether CORS should be enabled for the server.
loop (asyncio.AbstractEventLoop):
The event loop in which the server will run.
Returns:
socketio.AsyncServer: The configured Socket.IO asynchronous server.
"""
state_manager = observer.state_manager
if enable_cors:
sio = socketio.AsyncServer(async_mode="asgi", cors_allowed_origins="*")
else:
sio = socketio.AsyncServer(async_mode="asgi")
setup_sio_events(sio, state_manager)
setup_logging_handler(sio)
# Add notification callback to observer
def sio_callback(
full_access_path: str, value: Any, cached_value_dict: dict[str, Any]
) -> None:
if cached_value_dict != {}:
serialized_value = dump(value)
if cached_value_dict["type"] != "method":
cached_value_dict["type"] = serialized_value["type"]
cached_value_dict["value"] = serialized_value["value"]
async def notify() -> None:
try:
await sio.emit(
"notify",
{
"data": {
"full_access_path": full_access_path,
"value": cached_value_dict,
}
},
)
except Exception as e:
logger.warning("Failed to send notification: %s", e)
loop.create_task(notify())
observer.add_notification_callback(sio_callback)
return sio
def setup_sio_events(sio: socketio.AsyncServer, state_manager: StateManager) -> None:
@sio.event
def set_attribute(sid: str, data: UpdateDict) -> Any:
logger.debug("Received frontend update: %s", data)
parent_path = data["parent_path"].split(".")
path_list = [element for element in parent_path if element] + [data["name"]]
path = ".".join(path_list)
return state_manager.set_service_attribute_value_by_path(
path=path, value=data["value"]
)
@sio.event
def run_method(sid: str, data: RunMethodDict) -> Any:
logger.debug("Running method: %s", data)
parent_path = data["parent_path"].split(".")
path_list = [element for element in parent_path if element] + [data["name"]]
method = get_object_attr_from_path_list(state_manager.service, path_list)
return process_callable_attribute(method, data["kwargs"])
def setup_logging_handler(sio: socketio.AsyncServer) -> None:
logger = logging.getLogger()
logger.addHandler(SocketIOHandler(sio))

185
src/pydase/server/web_server/web_server.py Normal file
View File

@ -0,0 +1,185 @@
import asyncio
import json
import logging
from pathlib import Path
from typing import Any
import socketio # type: ignore[import-untyped]
import uvicorn
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
from pydase.config import ServiceConfig, WebServerConfig
from pydase.data_service.data_service_observer import DataServiceObserver
from pydase.server.web_server.sio_setup import (
setup_sio_server,
)
from pydase.utils.serializer import generate_serialized_data_paths
from pydase.version import __version__
logger = logging.getLogger(__name__)
class WebServer:
"""
Represents a web server that adheres to the AdditionalServerProtocol, designed to
work with a DataService instance. This server facilitates client-server
communication and state management through web protocols and socket connections.
The WebServer class initializes and manages a web server environment using FastAPI
and Socket.IO, allowing for HTTP and WebSocket communications. It incorporates CORS
(Cross-Origin Resource Sharing) support, custom CSS, and serves a frontend static
files directory. It also initializes web server settings based on configuration
files or generates default settings if necessary.
Configuration for the web server (like service configuration directory and whether
to generate new web settings) is determined in the following order of precedence:
1. Values provided directly to the constructor.
2. Environment variable settings (via configuration classes like
`pydase.config.ServiceConfig` and `pydase.config.WebServerConfig`).
3. Default values defined in the configuration classes.
Args:
data_service_observer (DataServiceObserver): Observer for the DataService,
handling state updates and communication to connected clients.
host (str): Hostname or IP address where the server is accessible. Commonly
'0.0.0.0' to bind to all network interfaces.
port (int): Port number on which the server listens. Typically in the range
1024-65535 (non-standard ports).
css (str | Path | None, optional): Path to a custom CSS file for styling the
frontend. If None, no custom styles are applied. Defaults to None.
enable_cors (bool, optional): Flag to enable or disable CORS policy. When True,
CORS is enabled, allowing cross-origin requests. Defaults to True.
config_dir (Path | None, optional): Path to the configuration
directory where the web settings will be stored. Defaults to
`pydase.config.ServiceConfig().config_dir`.
generate_new_web_settings (bool | None, optional): Flag to enable or disable
generation of new web settings if the configuration file is missing. Defaults
to `pydase.config.WebServerConfig().generate_new_web_settings`.
**kwargs (Any): Additional unused keyword arguments.
"""
def __init__( # noqa: PLR0913
self,
data_service_observer: DataServiceObserver,
host: str,
port: int,
css: str | Path | None = None,
enable_cors: bool = True,
config_dir: Path = ServiceConfig().config_dir,
generate_web_settings: bool = WebServerConfig().generate_web_settings,
**kwargs: Any,
) -> None:
self.observer = data_service_observer
self.state_manager = self.observer.state_manager
self.service = self.state_manager.service
self.port = port
self.host = host
self.css = css
self.enable_cors = enable_cors
self._service_config_dir = config_dir
self._generate_web_settings = generate_web_settings
self._loop: asyncio.AbstractEventLoop
self._initialise_configuration()
async def serve(self) -> None:
self._loop = asyncio.get_running_loop()
self._setup_socketio()
self._setup_fastapi_app()
self.web_server = uvicorn.Server(
uvicorn.Config(self.__fastapi_app, host=self.host, port=self.port)
)
# overwrite uvicorn's signal handlers, otherwise it will bogart SIGINT and
# SIGTERM, which makes it impossible to escape out of
self.web_server.install_signal_handlers = lambda: None # type: ignore[method-assign]
await self.web_server.serve()
def _initialise_configuration(self) -> None:
logger.debug("Initialising web server configuration...")
file_path = self._service_config_dir / "web_settings.json"
if self._generate_web_settings:
# File does not exist, create it with default content
logger.debug("Generating web settings file...")
file_path.parent.mkdir(
parents=True, exist_ok=True
) # Ensure directory exists
file_path.write_text(json.dumps(self.web_settings, indent=4))
def _get_web_settings_from_file(self) -> dict[str, dict[str, Any]]:
file_path = self._service_config_dir / "web_settings.json"
web_settings = {}
# File exists, read its content
if file_path.exists():
logger.debug(
"Reading configuration from file '%s' ...", file_path.absolute()
)
web_settings = json.loads(file_path.read_text())
return web_settings
@property
def web_settings(self) -> dict[str, dict[str, Any]]:
current_web_settings = self._get_web_settings_from_file()
for path in generate_serialized_data_paths(self.state_manager.cache):
if path in current_web_settings:
continue
current_web_settings[path] = {"displayName": path.split(".")[-1]}
return current_web_settings
def _setup_socketio(self) -> None:
self._sio = setup_sio_server(self.observer, self.enable_cors, self._loop)
self.__sio_app = socketio.ASGIApp(self._sio)
def _setup_fastapi_app(self) -> None: # noqa: C901
app = FastAPI()
if self.enable_cors:
app.add_middleware(
CORSMiddleware,
allow_credentials=True,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
app.mount("/ws", self.__sio_app)
@app.get("/version")
def version() -> str:
return __version__
@app.get("/name")
def name() -> str:
return type(self.service).__name__
@app.get("/service-properties")
def service_properties() -> dict[str, Any]:
return self.state_manager.cache
@app.get("/web-settings")
def web_settings() -> dict[str, Any]:
return self.web_settings
# exposing custom.css file provided by user
if self.css is not None:
@app.get("/custom.css")
async def styles() -> FileResponse:
return FileResponse(str(self.css))
app.mount(
"/",
StaticFiles(
directory=Path(__file__).parent.parent.parent / "frontend",
html=True,
),
)
self.__fastapi_app = app

53
src/pydase/utils/serializer.py
View File

@ -341,41 +341,48 @@ def get_next_level_dict_by_key(
def generate_serialized_data_paths(
data: dict[str, Any], parent_path: str = ""
data: dict[str, dict[str, Any]], parent_path: str = ""
) -> list[str]:
"""
Generate a list of access paths for all attributes in a dictionary representing
data serialized with `pydase.utils.serializer.Serializer`, excluding those that are
methods.
methods. This function handles nested structures, including lists, by generating
paths for each element in the nested lists.
Args:
data: The dictionary representing serialized data, typically produced by
`pydase.utils.serializer.Serializer`.
parent_path: The base path to prepend to the keys in the `data` dictionary to
form the access paths. Defaults to an empty string.
data (dict[str, Any]): The dictionary representing serialized data, typically
produced by `pydase.utils.serializer.Serializer`.
parent_path (str, optional): The base path to prepend to the keys in the `data`
dictionary to form the access paths. Defaults to an empty string.
Returns:
A list of strings where each string is a dot-notation access path to an
attribute in the serialized data.
list[str]: A list of strings where each string is a dot-notation access path
to an attribute in the serialized data. For list elements, the path includes
the index in square brackets.
"""
paths: list[str] = []
for key, value in data.items():
if value["type"] == "method":
# ignoring methods
continue
new_path = f"{parent_path}.{key}" if parent_path else key
if isinstance(value["value"], dict) and value["type"] != "Quantity":
paths.extend(generate_serialized_data_paths(value["value"], new_path))
elif isinstance(value["value"], list):
for index, item in enumerate(value["value"]):
indexed_key_path = f"{new_path}[{index}]"
if isinstance(item["value"], dict):
paths.extend(
generate_serialized_data_paths(item["value"], indexed_key_path)
)
else:
paths.append(new_path)
if serialized_dict_is_nested_object(value):
if isinstance(value["value"], list):
for index, item in enumerate(value["value"]):
indexed_key_path = f"{new_path}[{index}]"
paths.append(indexed_key_path)
else:
paths.append(new_path)
if serialized_dict_is_nested_object(item):
paths.extend(
generate_serialized_data_paths(
item["value"], indexed_key_path
)
)
continue
paths.extend(generate_serialized_data_paths(value["value"], new_path))
return paths
def serialized_dict_is_nested_object(serialized_dict: dict[str, Any]) -> bool:
return (
serialized_dict["type"] != "Quantity"
and isinstance(serialized_dict["value"], dict)
) or isinstance(serialized_dict["value"], list)

4
tests/data_service/test_data_service_cache.py
View File

@ -1,6 +1,7 @@
import logging
import pydase
import pytest
from pydase.data_service.data_service_observer import DataServiceObserver
from pydase.data_service.state_manager import StateManager
@ -34,7 +35,8 @@ def test_nested_attributes_cache_callback() -> None:
)
def test_task_status_update() -> None:
@pytest.mark.asyncio
async def test_task_status_update() -> None:
class ServiceClass(pydase.DataService):
name = "World"

85
tests/data_service/test_state_manager.py
View File

@ -3,9 +3,9 @@ from pathlib import Path
from typing import Any
import pydase
import pydase.components
import pydase.units as u
import pytest
from pydase.components.coloured_enum import ColouredEnum
from pydase.data_service.data_service_observer import DataServiceObserver
from pydase.data_service.state_manager import (
StateManager,
@ -19,12 +19,53 @@ class SubService(pydase.DataService):
name = "SubService"
class State(ColouredEnum):
class State(pydase.components.ColouredEnum):
RUNNING = "#0000FF80"
COMPLETED = "hsl(120, 100%, 50%)"
FAILED = "hsla(0, 100%, 50%, 0.7)"
class MySlider(pydase.components.NumberSlider):
@property
def min(self) -> float:
return self._min
@min.setter
@load_state
def min(self, value: float) -> None:
self._min = value
@property
def max(self) -> float:
return self._max
@max.setter
@load_state
def max(self, value: float) -> None:
self._max = value
@property
def step_size(self) -> float:
return self._step_size
@step_size.setter
@load_state
def step_size(self, value: float) -> None:
self._step_size = value
@property
def value(self) -> float:
return self._value
@value.setter
@load_state
def value(self, value: float) -> None:
if value < self._min or value > self._max:
raise ValueError("Value is either below allowed min or above max value.")
self._value = value
class Service(pydase.DataService):
def __init__(self, **kwargs: Any) -> None:
super().__init__(**kwargs)
@ -35,6 +76,7 @@ class Service(pydase.DataService):
self._property_attr = 1337.0
self._name = "Service"
self.state = State.RUNNING
self.my_slider = MySlider()
@property
def name(self) -> str:
@ -61,6 +103,37 @@ LOAD_STATE = {
"readonly": False,
"doc": None,
},
"my_slider": {
"type": "NumberSlider",
"value": {
"max": {
"type": "float",
"value": 101.0,
"readonly": False,
"doc": "The min property.",
},
"min": {
"type": "float",
"value": 1.0,
"readonly": False,
"doc": "The min property.",
},
"step_size": {
"type": "float",
"value": 2.0,
"readonly": False,
"doc": "The min property.",
},
"value": {
"type": "float",
"value": 1.0,
"readonly": False,
"doc": "The value property.",
},
},
"readonly": False,
"doc": None,
},
"name": {
"type": "str",
"value": "Another name",
@ -153,6 +226,10 @@ def test_load_state(tmp_path: Path, caplog: LogCaptureFixture) -> None:
assert service.name == "Service" # has not changed as readonly
assert service.some_float == 1.0 # has not changed due to different type
assert service.subservice.name == "SubService" # didn't change
assert service.my_slider.value == 1.0 # changed
assert service.my_slider.min == 1.0 # changed
assert service.my_slider.max == 101.0 # changed
assert service.my_slider.step_size == 2.0 # changed
assert "'some_unit' changed to '12.0 A'" in caplog.text
assert (
@ -168,6 +245,10 @@ def test_load_state(tmp_path: Path, caplog: LogCaptureFixture) -> None:
"Ignoring value from JSON file..." in caplog.text
)
assert "Value of attribute 'subservice.name' has not changed..." in caplog.text
assert "'my_slider.value' changed to '1.0'" in caplog.text
assert "'my_slider.min' changed to '1.0'" in caplog.text
assert "'my_slider.max' changed to '101.0'" in caplog.text
assert "'my_slider.step_size' changed to '2.0'" in caplog.text
def test_filename_warning(tmp_path: Path, caplog: LogCaptureFixture) -> None:

43
tests/data_service/test_task_manager.py
View File

@ -1,6 +1,8 @@
import asyncio
import logging
import pydase
import pytest
from pydase.data_service.data_service_observer import DataServiceObserver
from pydase.data_service.state_manager import StateManager
from pytest import LogCaptureFixture
@ -8,13 +10,14 @@ from pytest import LogCaptureFixture
logger = logging.getLogger()
def test_autostart_task_callback(caplog: LogCaptureFixture) -> None:
@pytest.mark.asyncio
async def test_autostart_task_callback(caplog: LogCaptureFixture) -> None:
class MyService(pydase.DataService):
def __init__(self) -> None:
super().__init__()
self._autostart_tasks = { # type: ignore
"my_task": (),
"my_other_task": (),
"my_task": (), # type: ignore
"my_other_task": (), # type: ignore
}
async def my_task(self) -> None:
@ -23,6 +26,7 @@ def test_autostart_task_callback(caplog: LogCaptureFixture) -> None:
async def my_other_task(self) -> None:
logger.info("Triggered other task.")
# Your test code here
service_instance = MyService()
state_manager = StateManager(service_instance)
DataServiceObserver(state_manager)
@ -32,7 +36,8 @@ def test_autostart_task_callback(caplog: LogCaptureFixture) -> None:
assert "'my_other_task' changed to '{}'" in caplog.text
def test_DataService_subclass_autostart_task_callback(
@pytest.mark.asyncio
async def test_DataService_subclass_autostart_task_callback(
caplog: LogCaptureFixture,
) -> None:
class MySubService(pydase.DataService):
@ -61,7 +66,8 @@ def test_DataService_subclass_autostart_task_callback(
assert "'sub_service.my_other_task' changed to '{}'" in caplog.text
def test_DataService_subclass_list_autostart_task_callback(
@pytest.mark.asyncio
async def test_DataService_subclass_list_autostart_task_callback(
caplog: LogCaptureFixture,
) -> None:
class MySubService(pydase.DataService):
@ -90,3 +96,30 @@ def test_DataService_subclass_list_autostart_task_callback(
assert "'sub_services_list[0].my_other_task' changed to '{}'" in caplog.text
assert "'sub_services_list[1].my_task' changed to '{}'" in caplog.text
assert "'sub_services_list[1].my_other_task' changed to '{}'" in caplog.text
@pytest.mark.asyncio
async def test_start_and_stop_task_methods(caplog: LogCaptureFixture) -> None:
class MyService(pydase.DataService):
def __init__(self) -> None:
super().__init__()
async def my_task(self, param: str) -> None:
while True:
logger.debug("Logging param: %s", param)
await asyncio.sleep(0.1)
# Your test code here
service_instance = MyService()
state_manager = StateManager(service_instance)
DataServiceObserver(state_manager)
service_instance.start_my_task("Hello")
await asyncio.sleep(0.01)
assert "'my_task' changed to '{'param': 'Hello'}'" in caplog.text
assert "Logging param: Hello" in caplog.text
caplog.clear()
service_instance.stop_my_task()
await asyncio.sleep(0.01)
assert "Task 'my_task' was cancelled" in caplog.text

68
tests/server/web_server/test_sio_setup.py Normal file
View File

@ -0,0 +1,68 @@
import asyncio
import logging
import pydase
import pytest
from pydase.data_service.data_service_observer import DataServiceObserver
from pydase.data_service.state_manager import StateManager
from pydase.server.web_server.sio_setup import (
RunMethodDict,
UpdateDict,
setup_sio_server,
)
logger = logging.getLogger(__name__)
@pytest.mark.asyncio
async def test_set_attribute_event() -> None:
class SubClass(pydase.DataService):
name = "SubClass"
class ServiceClass(pydase.DataService):
def __init__(self) -> None:
super().__init__()
self.sub_class = SubClass()
def some_method(self) -> None:
logger.info("Triggered 'test_method'.")
service_instance = ServiceClass()
state_manager = StateManager(service_instance)
observer = DataServiceObserver(state_manager)
server = setup_sio_server(observer, False, asyncio.get_running_loop())
test_sid = 1234
test_data: UpdateDict = {
"parent_path": "sub_class",
"name": "name",
"value": "new name",
}
server.handlers["/"]["set_attribute"](test_sid, test_data)
assert service_instance.sub_class.name == "new name"
@pytest.mark.asyncio
async def test_run_method_event(caplog: pytest.LogCaptureFixture):
class ServiceClass(pydase.DataService):
def test_method(self) -> None:
logger.info("Triggered 'test_method'.")
state_manager = StateManager(ServiceClass())
observer = DataServiceObserver(state_manager)
server = setup_sio_server(observer, False, asyncio.get_running_loop())
test_sid = 1234
test_data: RunMethodDict = {
"parent_path": "",
"name": "test_method",
"kwargs": {},
}
server.handlers["/"]["run_method"](test_sid, test_data)
assert "Triggered 'test_method'." in caplog.text

51
tests/server/web_server/test_web_server.py Normal file
View File

@ -0,0 +1,51 @@
import json
import logging
import tempfile
from pathlib import Path
import pydase
from pydase.data_service.data_service_observer import DataServiceObserver
from pydase.data_service.state_manager import StateManager
from pydase.server.web_server.web_server import WebServer
logger = logging.getLogger(__name__)
def test_web_settings() -> None:
class SubClass(pydase.DataService):
name = "Hello"
class ServiceClass(pydase.DataService):
def __init__(self) -> None:
super().__init__()
self.attr_1 = SubClass()
self.added = "added"
service_instance = ServiceClass()
state_manager = StateManager(service_instance)
observer = DataServiceObserver(state_manager)
with tempfile.TemporaryDirectory() as tmp:
web_settings = {
"attr_1": {"displayName": "Attribute"},
"attr_1.name": {"displayName": "Attribute name"},
}
web_settings_file = Path(tmp) / "web_settings.json"
with web_settings_file.open("w") as file:
file.write(json.dumps(web_settings))
server = WebServer(
observer,
host="0.0.0.0",
port=8001,
generate_web_settings=True,
config_dir=Path(tmp),
)
new_web_settings = server.web_settings
# existing entries are not overwritten, new entries are appended
assert new_web_settings == {**web_settings, "added": {"displayName": "added"}}
assert json.loads(web_settings_file.read_text()) == {
**web_settings,
"added": {"displayName": "added"},
}

5
tests/utils/test_helpers.py
View File

@ -1,5 +1,8 @@
import pytest
from pydase.utils.helpers import is_property_attribute
from pydase.utils.helpers import (
is_property_attribute,
)
@pytest.mark.parametrize(

133
tests/utils/test_serializer.py
View File

@ -11,6 +11,7 @@ from pydase.utils.serializer import (
dump,
get_nested_dict_by_path,
get_next_level_dict_by_key,
serialized_dict_is_nested_object,
set_nested_value_by_path,
)
@ -124,7 +125,8 @@ def test_ColouredEnum_serialize() -> None:
}
def test_method_serialization() -> None:
@pytest.mark.asyncio
async def test_method_serialization() -> None:
class ClassWithMethod(pydase.DataService):
def some_method(self) -> str:
return "some method"
@ -360,7 +362,9 @@ def test_update_invalid_list_index(setup_dict, caplog: pytest.LogCaptureFixture)
)
def test_update_invalid_path(setup_dict, caplog: pytest.LogCaptureFixture):
def test_update_invalid_path(
setup_dict: dict[str, Any], caplog: pytest.LogCaptureFixture
) -> None:
set_nested_value_by_path(setup_dict, "invalid_path", 30)
assert (
"Error occured trying to access the key 'invalid_path': it is either "
@ -369,66 +373,165 @@ def test_update_invalid_path(setup_dict, caplog: pytest.LogCaptureFixture):
)
def test_update_list_inside_class(setup_dict):
def test_update_list_inside_class(setup_dict: dict[str, Any]) -> None:
set_nested_value_by_path(setup_dict, "attr2.list_attr[1]", 40)
assert setup_dict["attr2"]["value"]["list_attr"]["value"][1]["value"] == 40
def test_update_class_attribute_inside_list(setup_dict):
def test_update_class_attribute_inside_list(setup_dict: dict[str, Any]) -> None:
set_nested_value_by_path(setup_dict, "attr_list[2].attr3", 50)
assert setup_dict["attr_list"]["value"][2]["value"]["attr3"]["value"] == 50
def test_get_next_level_attribute_nested_dict(setup_dict):
def test_get_next_level_attribute_nested_dict(setup_dict: dict[str, Any]) -> None:
nested_dict = get_next_level_dict_by_key(setup_dict, "attr1")
assert nested_dict == setup_dict["attr1"]
def test_get_next_level_list_entry_nested_dict(setup_dict):
def test_get_next_level_list_entry_nested_dict(setup_dict: dict[str, Any]) -> None:
nested_dict = get_next_level_dict_by_key(setup_dict, "attr_list[0]")
assert nested_dict == setup_dict["attr_list"]["value"][0]
def test_get_next_level_invalid_path_nested_dict(setup_dict):
def test_get_next_level_invalid_path_nested_dict(setup_dict: dict[str, Any]) -> None:
with pytest.raises(SerializationPathError):
get_next_level_dict_by_key(setup_dict, "invalid_path")
def test_get_next_level_invalid_list_index(setup_dict):
def test_get_next_level_invalid_list_index(setup_dict: dict[str, Any]) -> None:
with pytest.raises(SerializationPathError):
get_next_level_dict_by_key(setup_dict, "attr_list[10]")
def test_get_attribute(setup_dict):
def test_get_attribute(setup_dict: dict[str, Any]) -> None:
nested_dict = get_nested_dict_by_path(setup_dict, "attr1")
assert nested_dict["value"] == 1.0
def test_get_nested_attribute(setup_dict):
def test_get_nested_attribute(setup_dict: dict[str, Any]) -> None:
nested_dict = get_nested_dict_by_path(setup_dict, "attr2.attr3")
assert nested_dict["value"] == 1.0
def test_get_list_entry(setup_dict):
def test_get_list_entry(setup_dict: dict[str, Any]) -> None:
nested_dict = get_nested_dict_by_path(setup_dict, "attr_list[1]")
assert nested_dict["value"] == 1
def test_get_list_inside_class(setup_dict):
def test_get_list_inside_class(setup_dict: dict[str, Any]) -> None:
nested_dict = get_nested_dict_by_path(setup_dict, "attr2.list_attr[1]")
assert nested_dict["value"] == 1.0
def test_get_class_attribute_inside_list(setup_dict):
def test_get_class_attribute_inside_list(setup_dict: dict[str, Any]) -> None:
nested_dict = get_nested_dict_by_path(setup_dict, "attr_list[2].attr3")
assert nested_dict["value"] == 1.0
def test_get_invalid_list_index(setup_dict, caplog: pytest.LogCaptureFixture):
def test_get_invalid_list_index(setup_dict: dict[str, Any]) -> None:
with pytest.raises(SerializationPathError):
get_nested_dict_by_path(setup_dict, "attr_list[10]")
def test_get_invalid_path(setup_dict, caplog: pytest.LogCaptureFixture):
def test_get_invalid_path(setup_dict: dict[str, Any]) -> None:
with pytest.raises(SerializationPathError):
get_nested_dict_by_path(setup_dict, "invalid_path")
def test_serialized_dict_is_nested_object() -> None:
serialized_dict = {
"list_attr": {
"type": "list",
"value": [
{"type": "float", "value": 1.4, "readonly": False, "doc": None},
{"type": "float", "value": 2.0, "readonly": False, "doc": None},
],
"readonly": False,
"doc": None,
},
"my_slider": {
"type": "NumberSlider",
"value": {
"max": {
"type": "float",
"value": 101.0,
"readonly": False,
"doc": "The min property.",
},
"min": {
"type": "float",
"value": 1.0,
"readonly": False,
"doc": "The min property.",
},
"step_size": {
"type": "float",
"value": 2.0,
"readonly": False,
"doc": "The min property.",
},
"value": {
"type": "float",
"value": 1.0,
"readonly": False,
"doc": "The value property.",
},
},
"readonly": False,
"doc": None,
},
"string": {
"type": "str",
"value": "Another name",
"readonly": True,
"doc": None,
},
"float": {
"type": "int",
"value": 10,
"readonly": False,
"doc": None,
},
"unit": {
"type": "Quantity",
"value": {"magnitude": 12.0, "unit": "A"},
"readonly": False,
"doc": None,
},
"state": {
"type": "ColouredEnum",
"value": "FAILED",
"readonly": True,
"doc": None,
"enum": {
"RUNNING": "#0000FF80",
"COMPLETED": "hsl(120, 100%, 50%)",
"FAILED": "hsla(0, 100%, 50%, 0.7)",
},
},
"subservice": {
"type": "DataService",
"value": {
"name": {
"type": "str",
"value": "SubService",
"readonly": False,
"doc": None,
}
},
"readonly": False,
"doc": None,
},
}
assert serialized_dict_is_nested_object(serialized_dict["list_attr"])
assert serialized_dict_is_nested_object(serialized_dict["my_slider"])
assert serialized_dict_is_nested_object(serialized_dict["subservice"])
assert not serialized_dict_is_nested_object(
serialized_dict["list_attr"]["value"][0] # type: ignore[index]
)
assert not serialized_dict_is_nested_object(serialized_dict["string"])
assert not serialized_dict_is_nested_object(serialized_dict["unit"])
assert not serialized_dict_is_nested_object(serialized_dict["float"])
assert not serialized_dict_is_nested_object(serialized_dict["state"])