Create JsonModelLoader

[oliwenmandiamond]

Create a json_model_loader factory function. Wraps load_json_file_to_class but allows for use of configuring defaults. Will be used with plans and BlueapiClient.

from blueapi.client import BlueapiClient
bc = BlueapiClient.from_config_file("/path/to/config.yaml")
...

from dodal.common.data_util import json_model_loader, JsonModelConfig
load_data = json_model_loader(MyModel, JsonModelConfig.from_default_file("/my/default/file.json"))
...

# Plans can't use file paths, so need to be loaded on client side and data passed
# through via pydantic model.
bc.plan.my_plan(det1, load_data()) # Uses configured default file.
bc.plan.my_plan(det1, load_data("/path/to/data.json")) # Use absolute path
bc.plan.my_plan(det1, load_data("data.json")) # Use relative path

Instructions to reviewer on how to test:

1. Check tests pass.
2. Implementation makes sense.

Checks for reviewer

• Would the PR title make sense to a scientist on a set of release notes
• If a new device has been added does it follow the standards
• If changing the API for a pre-existing device, ensure that any beamlines using this device have updated their Bluesky plans accordingly
• Have the connection tests for the relevant beamline(s) been run via dodal connect ${BEAMLINE}

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 99.08%. Comparing base (178425f) to head (0dc67c1).

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #1934      +/-   ##
==========================================
+ Coverage   99.05%   99.08%   +0.02%     
==========================================
  Files         314      314              
  Lines       11779    11802      +23     
==========================================
+ Hits        11668    11694      +26     
+ Misses        111      108       -3     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[tpoliaw]

There isn't really anything wrong with the implementation but couldn't the same thing be done with a local function?

def load_data(file: str = DEFAULT) -> MyModel:
    return load_json_file_to_class(MyModel, file)

bc.plans.my_plan(det1, load_data())
bc.plans.my_plan(det1, load_data("path/to/data.json"))

As models are going to be serialized back to JSON and then reconstructed and validated on the server, it might be easier to read straight to JSON and pass it directly. At the moment the data is going file -> str -> json -> model -> json -> model where the two models aren't necessarily the same version.

def load_data(file: str = DEFAULT):
    with open(file) as src:
        return json.load(src)

I might have missed something though, is there a use case within plans where this would offer something more?

[oliwenmandiamond]

There isn't really anything wrong with the implementation but couldn't the same thing be done with a local function?

def load_data(file: str = DEFAULT) -> MyModel:
    return load_json_file_to_class(MyModel, file)

bc.plans.my_plan(det1, load_data())
bc.plans.my_plan(det1, load_data("path/to/data.json"))

Yes, I could do this

def p60_load_sequence(file: str) -> P60Sequence: 
     return load_json_file_to_class(P60Sequence, file) 

...
def i09_load_sequence(file: str) -> I09Sequence: 
    return load_json_file_to_class(I09Sequence, file) 

...
def i09_1_load_sequence(file: str) -> I091Sequence: 
    return load_json_file_to_class(I091Sequence, file) 

...
def b07_load_sequence(file: str) -> B07BSequence: 
    return load_json_file_to_class(B07BSequence, file)

...
def b07_1_load_sequence(file: str) -> B07CSequence: 
    return load_json_file_to_class(B07CSequence, file)

for each specific beamline. However, I wanted a standard way to do this via configuration rather defining a new function in the environment every time. This also means it is fully tested, has documentation, and can also reused with existing tests within dodal and plan repos rather than redefining a new loader function for each test.

As models are going to be serialized back to JSON and then reconstructed and validated on the server, it might be easier to read straight to JSON and pass it directly. At the moment the data is going file -> str -> json -> model -> json -> model where the two models aren't necessarily the same version.

def load_data(file: str = DEFAULT):
    with open(file) as src:
        return json.load(src)

I might have missed something though, is there a use case within plans where this would offer something more?

So I know the type checking isn't added yet, but when it is added surely this will provide the type checking for the model? E.g the model on client side and server side have to match?

def my_plan(model: MyModel1, ...):
     ...

load_model1 = JsonModelLoader[MyModel1](MyModel1)
load_model2 = JsonModelLoader[MyModel2](MyModel2)
...
bc.plans.my_plan(load_model1("/my/file")) # Okay
bc.plans.my_plan(load_model2("/my/file")) # Static type checker should complain as it isn't using the correct model?

[tpoliaw]

So I know the type checking isn't added yet, but when it is added surely this will provide the type checking for the model? E.g the model on client side and server side have to match?

Unfortunately not. The client is working entirely from the API schema provided by the server. For a plan that accepts MyModel, the schema only contains a list of fields that the model should have, it has no information about the source of the model. The PR for the offline client typing currently only uses Any for complex models as everything gets validated on the server.

You can see the spec the client has available by running blueapi controller -o json plans.

[tpoliaw]

However, I wanted a standard way to do this via configuration

It's not really any different but if you want to avoid writing a function you could do

def json_model_loader(model: type[TBaseModel], default_file: str|None=None):
    def load_json(file: str| None = default_file) -> TBaseModel:
        if file is None:
            raise ValueError("No filename given")
        return load_json_file_to_class(bm, file)
    return load_json

and then your example becomes

p60_load_sequence = json_model_loader(P60Sequence, file) 

i09_load_sequence = json_model_loader(I09Sequence, file) 

which isn't really that different to (and if anything is clearer than)

p60_load_sequence = JsonModelLoader[P60Sequence](P60Sequence, "my/default/file.json")

i09_load_sequence = JsonModelLoader[I09Sequence](I09Sequence, "my/default/file.json")

[oliwenmandiamond]

However, I wanted a standard way to do this via configuration

It's not really any different but if you want to avoid writing a function you could do

def json_model_loader(model: type[TBaseModel], file: str|None=None):
    def load_json(file: str| None) -> TBaseModel:
        if file is None:
            raise ValueError("No filename given")
        return load_json_file_to_class(bm, file)
    return load_json

and then your example becomes

p60_load_sequence = json_model_loader(P60Sequence, file) 

i09_load_sequence = json_model_loader(I09Sequence, file) 

which isn't really that different to (and if anything is clearer than)

p60_load_sequence = JsonModelLoader[P60Sequence](P60Sequence, "my/default/file.json")

i09_load_sequence = JsonModelLoader[I09Sequence](I09Sequence, "my/default/file.json")

With a small tweak, I do like how simple your function is

def json_model_loader(model: type[TBaseModel], default_file: str | None = None):
    def load_json(file: str | None = None) -> TBaseModel:
        if file is None and default_file is not None:
            return load_json_file_to_class(model, default_file)
        elif file is not None:
            return load_json_file_to_class(model, file)
        raise RuntimeError(
            f"{model.__name__} loader has no default file configured and no file was"
            "provided when trying to load in a model. "
        )
   return load_json

However, I am not sure it is correct approach because it is quite limiting. For example, I cannot view or update the default file at a later date, I have to rerun the factory function again and reassign to the variable for my function which I'm not sure I like

[tpoliaw]

When would you need to check or modify the default value? Isn't the point that you can override it if you want a different value?

Also, I posted my example function without checking it properly, I think if you use

def json_model_loader(model: type[TBaseModel], default_file: str|None=None):
    def load_json(file: str| None = default_file) -> TBaseModel:
        if file is None:
            raise ValueError("No filename given")
        return load_json_file_to_class(bm, file)
    return load_json

you get the default file behaviour you're after.

[oliwenmandiamond]

When would you need to check or modify the default value? Isn't the point that you can override it if you want a different value?

I was thinking of just trying to make it flexible enough. However, if we need it we can cross that bridge when we get there. I have update it to use your implementation so ready for review

[tpoliaw]

I think I stand by my original comment

There isn't really anything wrong with the implementation but couldn't the same thing be done with a local function?

I'm happy to let others chime in but until there's a clear need for updating defaults, this seems like a lot of complexity to avoid writing a one line function.

[oliwenmandiamond]

So I have added more optional configuration. You can optionally provide configuration providing a default file and default path, or provide just a default file and construct the path from the file without overloading the method with too many combinations but using a helper dataclass instead.

The use case is for electron analysers, they ususally do the following command

analyserscan detector "default_file.py" <scan_args>
analyserscan detector "new_file.py" <scan_args> #Using relative path
analyserscan detector "/path/to/new_file.py" <scan_args> #Using absolute path

The default path is based on the comissioning directoy, so if this ever changes we now can update it dynamically.

So now in BlueapiClient, we can do this instead

plans.analyserscan(detector, load_data(), <scan_args>) # Load default
plans.analyserscan(detector, load_data("new_file"), <scan_args>) # Load using relative path
plans.analyserscan(detector, load_data("/path/to/new_file"), <scan_args>) # Load using abs path

So I am trying to replicating this behaviour while giving flexibility so others may not need any default path or default file.

[Villtord]

I agree that it does look quite complex but also it makes util more flexible and adds good tests for functions, so would approve it.