# Module `kosmos`

## Classes

`class KosmosAuthInvalid`

  The provided credentials are not correct

  ### Ancestors

  - KosmosError
  - builtins.Exception
  - builtins.BaseException
`class KosmosClient (base: str, username: str, password: str, subs: dict = None)`

  The baseclass for our Kosmos Client

  creates a new instance of the Kosmos Client

  ## Parameters

  `base` :`str`
    the base for the server
  `username` :`str`
    the username to use
  `password` :`str`
    the password to use
  `subs` :`dict[KosmosEvent,Callable]`
    the subscriptions to start with

  ## Examples

  sync without async updates

      >>> kosmos = KosmosClient("http://localhost:18080", "user", "pass")

  async with listeners

      >>> with KosmosClient("http://localhost:18080", "user", "pass",
      >>>               subs={
      >>>                   KosmosEvent.auth_ok: on_auth_ok,
      >>>                   KosmosEvent.device_created: on_device_created,
      >>>                   KosmosEvent.device_attribute_updated: on_device_attribute_changed
      >>>               }) as kosmos:

  ### Ancestors

  - websocket._app.WebSocketApp
  - threading.Thread

  ### Methods

  `def add_device(self, uuid: str, state: dict = {}, schema: str = 'https://kosmos-lab.de/schema/NoSchema.json', scopes: KosmosScopeList = None) ‑> None`

    add a device to Kosmos

    ## Parameters

    `uuid` :`str`
      the UUID to use for the device
    `state` :`dict`
      the current state of the device - NEEDS to contain everything that is required by the schema!
    `schema` :`str`
      the $id of the Schema, defaults to https://kosmos-lab.de/schema/NoSchema.json
    `scopes` :`KosmosScopeList`
      the Scopes to use - defaults to no Scopes

    ## Raises

    `KosmosAuthInvalid`
      the auth was invalid
    `KosmosNotFoundError`
      the schema was not found
    `KosmosConflictError`
      the uuid is already in use or the validation against the schema failed
    `KosmosError`
      another error occurred

    ## Examples

        >>> kosmos.add_device(uuid="virt_hsv_lamp_3", state={"on": False, "hue": 120, "saturation": 60, "dimmingLevel": 50},
        >>>              schema="https://kosmos-lab.de/schema/HSVLamp.json")

  `def add_schema(self, schema: dict) ‑> None`

    add a schema to kosmos

    ## Parameters

    `schema` :`dict`
      the schema to add - needs to be a valid schema

    ## Raises

    `KosmosAuthInvalid`
      the auth was invalid
    `KosmosForbidden`
      the auth has not sufficient access to delete the schema
    `KosmosConflictError`
      something could not be added due to a conflict
    `KosmosError`
      anything else went wrong

    ## Examples

        >>> kosmos.add_schema({
        >>>         "failures": [
        >>>         ],
        >>>         "$schema": "http://json-schema.org/draft-07/schema#",
        >>>         "examples": [
        >>>         ],
        >>>         "additionalProperties": True,
        >>>         "title": "TestSchema1",
        >>>         "type": "object",
        >>>         "properties": {
        >>>         },
        >>>         "$id": "https://kosmos-lab.de/schema/TestSchema1.json"
        >>>     })

  `def add_scope(self, scope: str | KosmosScope) ‑> KosmosScope`

    add a scope to kosmos

    ## Parameters

    `name` :`str`
      the name of the new scope

    ## Raises

    `KosmosAuthInvalid`
      the auth was invalid
    `KosmosForbidden`
      the auth has not sufficient access to delete the schema
    `KosmosConflictError`
      could not add scope, name is in use
    `KosmosNotFoundError`
      the device could not be found!
    `KosmosError`
      anything else went wrong
  `def connect(self) ‑> None`

    connect to the kosmos backend

  `def delete_device(self, device: str | KosmosDevice) ‑> None`

    deletes a device from Kosmos

    ## Parameters

    `device` :`str | KosmosDevice`
      the device to update

    ## Raises

    `KosmosAuthInvalid`
      the auth was invalid
    `KosmosForbidden`
      the auth has not sufficient access to delete the device
    `KosmosNotFoundError`
      the device was not found
    `KosmosError`
      another error occurred

    ## Examples

        >>> kosmos.delete_device("virt_lamp_3")

  `def delete_schema(self, schemaid: str) ‑> None`

    deletes as schema from kosmos

    ## Parameters

    `schemaid` :`str`
      the id of the schema

    ## Raises

    `KosmosAuthInvalid`
      the auth was invalid
    `KosmosForbidden`
      the auth has not sufficient access to delete the schema
    `KosmosNotFoundError`
      the schema was not found
    `KosmosConflictError`
      the schema cannot be deleted - it is still in use
    `KosmosError`
      another error occurred

    ## Examples

        >>> kosmos.delete_schema("https://kosmos-lab.de/schema/TestSchema1.json")

  `def delete_scope(self, scope: str | int | KosmosScope) ‑> None`

    delete a scope from kosmos

    ## Parameters

    `scope` :`str | int`
      the name or id of the scope

    ## Raises

    `KosmosAuthInvalid`
      the auth was invalid
    `KosmosForbidden`
      the auth has not sufficient access to delete the scope
    `KosmosConflictError`
      could not add scope, name is in use
    `KosmosNotFoundError`
      the device could not be found!
    `KosmosError`
      anything else went wrong
  `def get_device(self, uuid: str) ‑> KosmosDevice`

    get the device from internal cache if possible - get from api otherwise

    ## Parameters

    `uuid` :`str`
      the uuid of the device

    ## Returns

    `KosmosDevice`
      the KosmosDevice we wanted to get

    ## Raises

    `KosmosAuthInvalid`
      the auth was invalid
    `KosmosForbidden`
      the auth has not sufficient access to delete the device
    `KosmosNotFoundError`
      the device was not found
    `KosmosError`
      another error occurred

    ## Examples

        >>> print(kosmos.get_device("virt_lamp_0"))
        >>> print(kosmos.get_device("virt_lamp_1").state["on"])

  `def get_location(self, device: str | KosmosDevice) ‑> KosmosLocation`

    gets the location of a given UUID back from kosmos

    ## Parameters

    `device` :`str`
      the unique ID of the device

    ## Returns

    `KosmosLocation`
      the Location of the device

    ## Raises

    `KosmosNotFoundError`
      the device could not be found!
    `KosmosAuthInvalid`
      the auth was invalid
    `KosmosForbidden`
      the auth has not sufficient access to delete the schema
    `KosmosConflictError`
      could not add scope, name is in use
    `KosmosError`
      anything else went wrong
  `def get_schema(self, schemaid: str) ‑> dict`

    get a schema back from kosmos

    ## Parameters

    `schemaid` :`str`
      the $id of the schema

    ## Returns

    `dict`
      the schema

    ## Raises

    `KosmosAuthInvalid`
      the auth was invalid
    `KosmosForbidden`
      the auth has not sufficient access to delete the schema
    `KosmosNotFoundError`
      the schema was not found
    `KosmosError`
      something else went wrong with the communication

    ## Examples

        >>> print(kosmos.get_schema("https://kosmos-lab.de/schema/TestSchema1.json"))

  `def get_scope(self, scope: str | int) ‑> KosmosScope`

    add a scope to kosmos

    ## Parameters

    `name` :`str`
      the name of the new scope

    ## Raises

    `KosmosAuthInvalid`
      the auth was invalid
    `KosmosForbidden`
      the auth has not sufficient access to delete the schema
    `KosmosConflictError`
      could not add scope, name is in use
    `KosmosNotFoundError`
      the device could not be found!
    `KosmosError`
      anything else went wrong
  `def is_connected(self) ‑> bool`

    checks if a connection is established

    ## Returns

    `True if connection is established and most likely alive`
  `def is_stopped(self) ‑> bool`

    check if the client got the command to stop itself

    ## Returns

    `bool:`
    `True if the client should be stopping`
  `def list_schemas(self) ‑> list[dict]`

    list all schemas in the kosmos

    ## Returns

    `List with currently known Schemas (List` of `Dicts)`

    ## Raises

    `KosmosAuthInvalid`
      the auth was invalid
    `KosmosForbidden`
      the auth has not sufficient access the list of schemas
    `KosmosError`
      another error occurred

    ## Examples

        >>> print(kosmos.list_schemas())

  `def reconnect(self) ‑> None`

    closes the current Connection, will be reestablished as long as __stopped is not True

  `def run(self)`

    the runner that automatically reconnects the websocket if needed

  `def set(self, device: str | KosmosDevice, value: dict[str, any]) ‑> None`

    Set the state of a device in the kosmos backend

    ## Parameters

    `device` :`str | KosmosDevice`
      the uuid or the actual KosmosDevice to update
    `value` :`dict`
      a dict of one or multiple attributes to change

    ## Raises

    `KosmosAuthInvalid`
      the auth was invalid
    `KosmosForbidden`
      the auth has not sufficient access to delete the device
    `KosmosNotFoundError`
      the device was not found
    `KosmosConflict`
      the set could not be executed
    `KosmosError`
      another error occurred

    ## Examples

        >>> kosmos.set("virt_heater_3", {"heatingTemperatureSetting": 21})

    sets the attribute "heatingTemperatureSetting" of "virt_heater_3" to 21

  `def set_attribute(self, device: str | KosmosDevice, attribute: str | int | float, value: str | int | bool | float | list | tuple | None | dict)`

    sets the given attribute on the kosmos backend to a specified value

    ## Parameters

    `device` :`str | KosmosDevice`
      the uuid or the actual KosmosDevice to update
    `attribute` :`str`
      the attribute to change
    `value`
      the new value to set (will be converted to JSON)

    ## Raises

    `KosmosAuthInvalid`
      the auth was invalid
    `KosmosForbidden`
      the auth has not sufficient access to delete the device
    `KosmosNotFoundError`
      the device was not found
    `KosmosConflict`
      the set could not be executed because it failed validation
    `KosmosError`
      another error occurred

    ## Examples

        >>> kosmos.set_attribute("virt_heater_3", "heatingTemperatureSetting", 24)

    sets the attribute "heatingTemperatureSetting" of "virt_heater_3" to 24

  `def set_location(self, device: str | KosmosDevice, location: KosmosLocation) ‑> None`

    set the location of a device

    ## Parameters

    `device` :`str | KosmosDevice`
      the device to update
    `location` :`KosmosLocation`
      the location to set it to
  `def stop(self) ‑> None`

    Signal the client that we would want to stop the client. Will close all connections

  `def subscribe(self, event_type: KosmosEvent, fn: Callable) ‑> None`

    subscribe to an event with the given Callable

    ## Parameters

    `event_type` :`KosmosEvent`
      the event type to listen to
    `fn` :`Callable`
      the function that fulfills the keyword parameters seen in the documentation for each KosmosEvent

    ## Examples

    Subscribe to auth events

        >>> def on_auth_ok(kosmos: KosmosClient):
        >>>     print(f"auth okay  on {kosmos}...")
        >>> kosmos.subscribe(KosmosEvent.auth_ok, on_auth_ok)

    Subscribe to changed attributes

        >>> def on_device_attribute_changed(kosmos: KosmosClient, device: KosmosDevice, attribute: str | int | float, value: str | int | bool | float | list | tuple | None | dict):
        >>>     print(f"change on {kosmos} {device} - set {attribute} to {value}")
        >>> kosmos.subscribe(KosmosEvent.device_attribute_updated, on_device_attribute_changed)

`class KosmosConflictError (*args, **kwargs)`

  there was a conflict (for example if the resource already existed, or the validation failed)

  ### Ancestors

  - KosmosError
  - builtins.Exception
  - builtins.BaseException
`class KosmosDevice (uuid: str, name: str = None, schema: str = None, state: dict = <factory>)`

  ### Class variables

  `var dataclass_json_config`
  `var last_change`
  `var last_update`
  `var name : str`
  `var state : dict`
  `var uuid : str`

  ### Static methods

  `def from_dict(kvs: Union[dict, list, str, int, float, bool, None], *, infer_missing=False) ‑> ~A`
  `def from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) ‑> ~A`
  `def schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]`

  ### Methods

  `def to_dict(self, encode_json=False) ‑> Dict[str, Union[dict, list, str, int, float, bool, None]]`
  `def to_json(self, *, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Union[int, str, None] = None, separators: Tuple[str, str] = None, default: Callable = None, sort_keys: bool = False, **kw) ‑> str`
`class KosmosError (*args, **kwargs)`

  a generic Kosmos Error

  ### Ancestors

  - builtins.Exception
  - builtins.BaseException

  ### Subclasses

  - KosmosAuthInvalid
  - KosmosConflictError
  - KosmosForbidden
  - KosmosNotFoundError
  - KosmosRequestFailed
`class KosmosEvent (value, names=None, *, module=None, qualname=None, type=None, start=1)`

  Event Enums for Kosmos

  ## Notes

  if you use subscribe to these events please make sure the function has the correct keyword parameters!

  ### Ancestors

  - enum.Enum

  ### Class variables

  `var auth_ok`

    will be triggered once the auth to kosmos was successful

    ## Parameter

        kosmos: KosmosClient
            the kosmos client

  `var connection_established`

    will be triggered everytime we have a new connection established with kosmos

    ## Parameter

        kosmos: KosmosClient
            the kosmos client

  `var connection_lost`

    will be triggered everytime we lose the connection to kosmos

    ## Parameter

        kosmos: KosmosClient
            the kosmos client

  `var device_attribute_updated`

    will be triggered once for each change to an attribute of a device

    ## Parameter

        kosmos: KosmosClient
            the kosmos client
        device: KosmosDevice
            the device that was updated
        attribute: str | int | float
            the attribute that changed
        value: str | int | bool | float | list | tuple | None | dict
            the new value

  `var device_created`

    will be triggered once for a new device, will also be triggered while doing the initial sync

    ## Parameter

        kosmos : KosmosClient
            the kosmos client
        device : KosmosDevice
            the new device that was found

  `var device_removed`

    will be triggered once for each device that was removed from the kosmos

    ## Parameter

        kosmos : KosmosClient
            the kosmos client
        device : KosmosDevice
            the device that was removed

  `var device_updated`

    will be triggered once for each update to a device's state

    ## Parameter

        kosmos: KosmosClient
            the kosmos client
        device: KosmosDevice
            the device that was updated

  `var stopped`

    will be triggered once if the stop method was called

    ## Parameter

        kosmos: KosmosClient
            the kosmos client

`class KosmosForbidden`

  The credentials are not valid for this request

  ### Ancestors

  - KosmosError
  - builtins.Exception
  - builtins.BaseException
`class KosmosGroup (id: int, name: str)`

  KosmosGroup(id: 'int', name: 'str')

  ### Class variables

  `var id : int`
  `var name : str`

  ### Static methods

  `def from_dict(kvs: Union[dict, list, str, int, float, bool, None], *, infer_missing=False) ‑> ~A`
  `def from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) ‑> ~A`
  `def schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]`

  ### Methods

  `def to_dict(self, encode_json=False) ‑> Dict[str, Union[dict, list, str, int, float, bool, None]]`
  `def to_json(self, *, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Union[int, str, None] = None, separators: Tuple[str, str] = None, default: Callable = None, sort_keys: bool = False, **kw) ‑> str`
`class KosmosLocation (x: Optional[int] = None, y: Optional[int] = None, z: Optional[int] = None, w: Optional[int] = None, d: Optional[int] = None, h: Optional[int] = None, yaw: Optional[int] = None, roll: Optional[int] = None, pitch: Optional[int] = None, area: Optional[str] = None)`

  KosmosLocation(x: 'Optional[int]' = None, y: 'Optional[int]' = None, z: 'Optional[int]' = None, w: 'Optional[int]' = None, d: 'Optional[int]' = None, h: 'Optional[int]' = None, yaw: 'Optional[int]' = None, roll: 'Optional[int]' = None, pitch: 'Optional[int]' = None, area: 'Optional[str]' = None)

  ### Class variables

  `var area : Optional[str]`
  `var d : Optional[int]`
  `var h : Optional[int]`
  `var pitch : Optional[int]`
  `var roll : Optional[int]`
  `var w : Optional[int]`
  `var x : Optional[int]`
  `var y : Optional[int]`
  `var yaw : Optional[int]`
  `var z : Optional[int]`

  ### Static methods

  `def from_dict(kvs: Union[dict, list, str, int, float, bool, None], *, infer_missing=False) ‑> ~A`
  `def from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) ‑> ~A`
  `def schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]`

  ### Methods

  `def to_dict(self, encode_json=False) ‑> Dict[str, Union[dict, list, str, int, float, bool, None]]`
  `def to_json(self, *, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Union[int, str, None] = None, separators: Tuple[str, str] = None, default: Callable = None, sort_keys: bool = False, **kw) ‑> str`
`class KosmosNotFoundError (*args, **kwargs)`

  the resource was not found

  ### Ancestors

  - KosmosError
  - builtins.Exception
  - builtins.BaseException
`class KosmosRequestFailed (response: Response)`

  The provided credentials are not correct

  ### Ancestors

  - KosmosError
  - builtins.Exception
  - builtins.BaseException
`class KosmosScope (users: list[KosmosUser], admins: list[KosmosUser], adminGroups: list[KosmosGroup], userGroups: list[KosmosGroup], name: str, id: int)`

  KosmosScope(users: 'list[KosmosUser]', admins: 'list[KosmosUser]', adminGroups: 'list[KosmosGroup]', userGroups: 'list[KosmosGroup]', name: 'str', id: 'int')

  ### Class variables

  `var adminGroups : list[KosmosGroup]`
  `var admins : list[KosmosUser]`
  `var id : int`
  `var name : str`
  `var userGroups : list[KosmosGroup]`
  `var users : list[KosmosUser]`

  ### Static methods

  `def from_dict(kvs: Union[dict, list, str, int, float, bool, None], *, infer_missing=False) ‑> ~A`
  `def from_dict_(s) ‑> KosmosScope`
  `def from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) ‑> ~A`
  `def schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]`

  ### Methods

  `def to_dict(self, encode_json=False) ‑> Dict[str, Union[dict, list, str, int, float, bool, None]]`
  `def to_json(self, *, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Union[int, str, None] = None, separators: Tuple[str, str] = None, default: Callable = None, sort_keys: bool = False, **kw) ‑> str`
`class KosmosScopeList (read: Optional[KosmosScope], write: Optional[KosmosScope], delete: Optional[KosmosScope])`

  KosmosScopeList(read: 'Optional[KosmosScope]', write: 'Optional[KosmosScope]', delete: 'Optional[KosmosScope]')

  ### Class variables

  `var delete : Optional[KosmosScope]`
  `var read : Optional[KosmosScope]`
  `var write : Optional[KosmosScope]`

  ### Static methods

  `def from_dict(kvs: Union[dict, list, str, int, float, bool, None], *, infer_missing=False) ‑> ~A`
  `def from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) ‑> ~A`
  `def schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]`

  ### Methods

  `def to_dict(self, encode_json=False) ‑> Dict[str, Union[dict, list, str, int, float, bool, None]]`
  `def to_json(self, *, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Union[int, str, None] = None, separators: Tuple[str, str] = None, default: Callable = None, sort_keys: bool = False, **kw) ‑> str`
`class KosmosUser (id: int, name: str)`

  KosmosUser(id: 'int', name: 'str')

  ### Class variables

  `var id : int`
  `var name : str`

  ### Static methods

  `def from_dict(kvs: Union[dict, list, str, int, float, bool, None], *, infer_missing=False) ‑> ~A`
  `def from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) ‑> ~A`
  `def schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]`

  ### Methods

  `def to_dict(self, encode_json=False) ‑> Dict[str, Union[dict, list, str, int, float, bool, None]]`
  `def to_json(self, *, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Union[int, str, None] = None, separators: Tuple[str, str] = None, default: Callable = None, sort_keys: bool = False, **kw) ‑> str`