""" nappSyncToOpenbis.py Sync napp* beamtime metadata rows into openBIS objects: update an existing object if one already exists for a given file name, otherwise create a new one. One object per table row. Fields already filled in openBIS (e.g. by hand during the beamtime) are never overwritten by automated values. Called from MATLAB via: py.importlib.import_module('nappSyncToOpenbis') resultJson = py.nappSyncToOpenbis.sync_rows(rowsJson, url, space, ...) ============================================================================ HOW THIS FILE IS ORGANIZED ============================================================================ Two sections: CONFIGURATION -- everything you're likely to want to change: which FileList column goes to which openBIS property, fixed values, how text fields get combined, etc. All plain data (dicts/lists/strings) -- no logic. This is what you edit for: - a new column -> property mapping - changing which column(s) feed a text field (e.g. Notes, Sample Notes) - a new fixed/constant value - adapting this file for a DIFFERENT openBIS object type entirely: replace this whole section, leave the engine below alone. ENGINE -- generic code that reads the configuration above and does the actual work (connecting, matching existing objects, uploading files, embedding images, etc.). You shouldn't need to touch this for ordinary changes -- only if you need a genuinely new KIND of derivation that doesn't fit the existing patterns (direct copy, numeric conversion, timestamp, text-join, category-dependent source). Several things below are workarounds for confirmed pybis (v1.37.5) quirks, found by reading pybis's own source directly rather than guessing -- see the comments at each one for what was actually wrong and how it was confirmed. These live in the engine, not the config, since they're not something a schema change would ever need to touch. """ import json import re from datetime import datetime, timedelta from zoneinfo import ZoneInfo from pandas import DataFrame from pybis import Openbis LOCAL_TZ = ZoneInfo("Europe/Zurich") # <-- change if a future beamtime runs elsewhere # ============================================================================ # CONFIGURATION -- edit this section for ordinary changes. See file docstring. # ============================================================================ OBJECT_TYPE = "LAC_EXPERIMENTAL_STEP_BEAMTIME" MEASUREMENT_TYPE_VOCABULARY = "LAC_EXPERIMENTALSTEPTYPE_BEALINE" RAW_DATASET_TYPE = "RAW_DATA" RESULTS_PROPERTY = "default_experiment.experimental_results" FILENUMBER_PROPERTY = "filenumber" CATEGORY_PROPERTY = "lac_steptype" KE_RANGE_PROPERTY = "ke_range_center" KE_RANGE_SOURCE_COLUMN = "scientaEkinRange_eV" # --- File category, derived from filename extension -- drives lac_steptype # and which photon-energy source column is used below. Add new extensions # / categories here if a future beamtime uses different file types. CATEGORY_BY_EXTENSION = { "ibw": "XPS", "h5": "NEXAFS", "bmp": "PHOTO", "jpg": "PHOTO", "jpeg": "PHOTO", "png": "PHOTO", "wmv": "PHOTO", "mov": "PHOTO", } # --- Direct column -> property mappings ------------------------------------ # Simple 1:1 copies, no conversion needed (text properties, or numeric # properties where the JSON value's native type already matches -- e.g. # ke_step is REAL and a plain float is exactly right as-is). # Property codes are lowercase -- confirmed openBIS wants that at the wire # level regardless of what the admin UI/xlsx export displays. DIRECT_MAP = { "sample": "sample_name", "regionName": "region", "scientaLensMode": "lens_mode", "scientaAcquisitionMode": "acq_mode", "scientaEkinStep_eV": "ke_step", } # --- Numeric columns needing a type conversion ----------------------------- # value = (openBIS property code, "int" or "float"). # "int" -> rounded to a genuine Python int (for INTEGER-typed properties; # MATLAB/JSON numbers arrive as floats, e.g. 57.0, even for # whole-number fields). # "float" -> passed through as-is (REAL-typed properties -- JSON already # gives a plain float, nothing to convert). # Note: sequence_iterations_real / region_iterations_real are misleadingly # named -- their CODEs say "_real" but their actual openBIS data type is # INTEGER, confirmed from the schema export. Named per the actual type # needed here, not per the (inconsistent) code name. NUMERIC_MAP = { "sampleTemp_dC": ("temp_reading", "float"), "iceTemp_dC": ("temp_calibrated", "float"), "cellPressure_mbar": ("cell_pressure_real", "float"), "scientaDwellTime_ms": ("dwell_time_integer", "int"), "scientaPassEnergy_eV": ("passenergy_integer", "int"), "scientaSequenceIterations": ("sequence_iterations_real", "int"), "scientaRegionIterations": ("region_iterations_real", "int"), "smplX_mm": ("position_x_real", "float"), "smplY_mm": ("position_y_real", "float"), "smplZ_mm": ("position_z_real", "float"), } # --- Timestamp columns ------------------------------------------------------ # value = (openBIS property code, MATLAB source format). # "datevec" = a 6-element [Y M D H MI S] list (e.g. creationDate) # "datenum" = a MATLAB serial datenum (e.g. lastModifiedDatenum) TIMESTAMP_MAP = { "creationDate": ("start_date", "datevec"), "lastModifiedDatenum": ("end_date", "datenum"), } # --- Text fields built by joining one or more columns with newlines -------- # key = target property code, value = list of source columns, in order. # A column that's empty/missing for a given row is simply skipped; if NONE # of a field's source columns have anything, the property is left unset # entirely (never written as a blank string). # This is exactly where to make a change like "comment should go to # Notes, not Sample Notes" -- edit the list(s) below, nothing else. TEXT_JOIN_MAP = { "notes": ["comment"], } # --- Photon energy: which column to use depends on the file category ------- # XPS uses a single fixed excitation energy; NEXAFS genuinely scans a # range. "scalar" reads one number (unwrapping a 1-element list if needed); # "range" expects a 2-element [min, max] list. PHOTON_ENERGY_PROPERTY = "photon_energy" PHOTON_ENERGY_SOURCE_BY_CATEGORY = { "XPS": ("xRayEkin_eV", "scalar"), "NEXAFS": ("xRayEkinRange_eV", "range"), } # --- Fixed/constant values, written once per sync (not derived from any # row's own data) ------------------------------------------------------ # Note: these are VOCABULARY TERM values, not property codes -- a # different namespace, genuinely uppercase per the LAC_ENDSTATIONS # vocabulary definition (confirmed from the schema export). CONSTANT_MAP = { "endstation_name": "SOLID-GAS", # <-- change if a different campaign/instrument uses this script } # --- Campaign-level values, from FileList.Properties.CustomProperties, # same value applied to every row this run ------------------------------ CAMPAIGN_MAP = { "beamline": "campaignLocation", "team": "team", "researchproject": "projectName", } # Deliberately not represented anywhere in this openBIS object type # (confirmed OK to leave unmapped): contact, contactORCID, funding, # publicationDoi, publicationDataDoi, rawDataDoi, dataLevel, sourceDirectory, # campaignName. # ============================================================================ # ENGINE -- generic logic reading the configuration above. See file docstring. # ============================================================================ def _connect(url, token=None, username=None, password=None, verify_certificates=True): o = Openbis(url, verify_certificates=verify_certificates) if token: # NOT o.set_token(token) -- that calls is_token_valid(), which hits # openBIS's isSessionActive RPC. Confirmed (by reading pybis # 1.37.5's source) that set_token()/the token property setter/ # is_session_active() all route through that one same check, and # confirmed separately that PAT authentication is disabled at the # server level on this instance (community.openbis.ch/t/134) -- # tokens are creatable/listable but rejected on real use. Writing # directly to __dict__ bypasses the property setter (and that # check) without affecting anything else -- every other pybis # method reads self.token via the plain getter, no validation on # read. Kept for whenever PAT auth gets enabled server-side. o.__dict__["token"] = token elif username and password: o.login(username, password) else: raise ValueError("Provide either a token, or both username and password.") return o def _clean_str_or_none(value): """Coerce a value that may have crossed the MATLAB<->Python boundary (via pyargs, not JSON) into a clean native str, or None if it represents 'nothing'. Deliberately avoids `if value:` -- MATLAB's [] can arrive as a NumPy array, and NumPy raises its own error ("ambiguous truth value") for exactly that kind of implicit boolean check, rather than just being falsy the way None/'' are.""" if value is None: return None if hasattr(value, "size"): # NumPy array or similar return None if value.size == 0 else str(value) text = str(value).strip() return text if text else None def _to_clean_int(value): """Round a MATLAB-crossed numeric value (arrives as a JSON float, e.g. 57.0) to a genuine Python int for an INTEGER-typed property. Returns None if it can't be cleanly interpreted as a number.""" if value is None: return None try: return int(round(float(value))) except (ValueError, TypeError): return None def _matlab_datenum_to_str(datenum): """MATLAB datenum (days since 0-Jan-0000) -> 'YYYY-MM-DD HH:MM:SS +ZZZZ', matching the format openBIS itself displays for these TIMESTAMP properties (confirmed from a real exported entry). The -366 offset is the standard MATLAB<->Python datenum conversion constant (Python's ordinal day 1 = 1-Jan-0001). Localized to LOCAL_TZ so the UTC offset is correct across the CET/CEST boundary.""" if datenum is None: return None try: days = int(datenum) frac = datenum - days dt = datetime.fromordinal(days - 366) + timedelta(days=frac) return dt.replace(tzinfo=LOCAL_TZ).strftime("%Y-%m-%d %H:%M:%S %z") except (ValueError, TypeError, OverflowError): return None def _matlab_datevec_to_str(datevec): """[Y, M, D, H, MI, S] -> 'YYYY-MM-DD HH:MM:SS +ZZZZ' (see above).""" if not isinstance(datevec, list) or len(datevec) < 6: return None try: y, mo, d, h, mi, s = datevec[:6] dt = datetime(int(y), int(mo), int(d), int(h), int(mi), int(s)) return dt.replace(tzinfo=LOCAL_TZ).strftime("%Y-%m-%d %H:%M:%S %z") except (ValueError, TypeError): return None def _file_number(name): """Doubled-digit convention: 'PHOENIX0183183_1.ibw' -> 183 (the leading digit run '0183183' splits into '0183' + '183', the same number with and without a leading zero). For names that don't match that pattern but have a later, short digit run (e.g. 'PHOENIX04042016_001.h5', where the leading run is a date, not a real index), that trailing run is used instead. Returns None if no clean number can be extracted.""" if not name: return None digit_runs = re.findall(r"\d+", name) if not digit_runs: return None first = digit_runs[0] n = len(first) for split_at in range(-(-n // 2), n): # ceil(n/2) .. n-1 a, b = first[:split_at], first[split_at:] if a and b and int(a) == int(b): try: return int(a) except ValueError: return None if len(digit_runs) > 1 and len(digit_runs[-1]) <= 4: try: return int(digit_runs[-1]) except ValueError: pass try: return int(first) except ValueError: return None # e.g. a date-based name too long/ambiguous to use as-is def _category_for_name(name): if not name or "." not in name: return None ext = name.lower().rsplit(".", 1)[-1] return CATEGORY_BY_EXTENSION.get(ext) def _row_derived_props(row, category): """Everything that isn't a straight column copy: timestamps, file number, category, photon/KE energy (which may depend on category), and the text-join fields. Driven entirely by the CONFIGURATION section above -- add a new derivation *kind* here only if none of the existing patterns (direct/numeric/timestamp/text-join/category-dependent) fit.""" props = {} for col, (prop, fmt) in TIMESTAMP_MAP.items(): val = row.get(col) formatted = _matlab_datevec_to_str(val) if fmt == "datevec" else _matlab_datenum_to_str(val) if formatted: props[prop] = formatted fnum = _file_number(row.get("name")) if fnum is not None: props[FILENUMBER_PROPERTY] = fnum if category is not None: props[CATEGORY_PROPERTY] = category if category in PHOTON_ENERGY_SOURCE_BY_CATEGORY: col, kind = PHOTON_ENERGY_SOURCE_BY_CATEGORY[category] val = row.get(col) if kind == "scalar": if isinstance(val, list): val = val[0] if len(val) > 0 else None if isinstance(val, (int, float)): props[PHOTON_ENERGY_PROPERTY] = f"{val:.4g} eV" elif kind == "range": if isinstance(val, list) and len(val) == 2 and all(v is not None for v in val): props[PHOTON_ENERGY_PROPERTY] = f"{val[0]:.4g} - {val[1]:.4g} eV" ke_range = row.get(KE_RANGE_SOURCE_COLUMN) if isinstance(ke_range, list) and len(ke_range) == 2 and all(v is not None for v in ke_range): props[KE_RANGE_PROPERTY] = f"{ke_range[0]:.4g} - {ke_range[1]:.4g} eV" for prop, cols in TEXT_JOIN_MAP.items(): parts = [str(row[c]) for c in cols if row.get(c)] if parts: props[prop] = "\n".join(parts) return props def attach_result_image(o, obj, image_path, url, verify_certificates=True): """Uploads image_path to openBIS's ELN file service and appends it to RESULTS_PROPERTY, alongside any existing content there rather than replacing it. Confirmed working pattern from openBIS support: community.openbis.ch/t/391 """ import requests upload_url = f"{url}/openbis/openbis/file-service/eln-lims?type=Files&sessionID={o.token}" with open(image_path, "rb") as f: r = requests.post(upload_url, files={"file": f}, verify=verify_certificates) r.raise_for_status() file_url = json.loads(r.text)["url"] existing = _current_prop_safe(obj, RESULTS_PROPERTY) figure_html = f'
' obj.props[RESULTS_PROPERTY] = (existing or "") + figure_html def _current_prop_safe(obj, prop_code): try: return obj.props[prop_code] except Exception: return None def add_photo_vocabulary_term(url, token=None, username=None, password=None, verify_certificates=True): """One-time (or idempotent) setup: adds a PHOTO term to MEASUREMENT_TYPE_VOCABULARY, so photo rows can set CATEGORY_PROPERTY. Safe to call more than once -- checks for the term first. Run this deliberately, not as part of routine syncing. """ o = _connect(url, token=token, username=username, password=password, verify_certificates=verify_certificates) try: existing_terms = o.get_terms(vocabulary=MEASUREMENT_TYPE_VOCABULARY) # Read the underlying DataFrame directly rather than iterating # existing_terms -- iterating a Things collection internally calls # a single-item fetch (get_term) that's missing a required arg in # this pybis version (confirmed via a real traceback: TypeError: # get_term() missing 'vocabularyCode'), the same class of bug as # get_parents() below. .df sidesteps that entirely. if existing_terms is not None and len(existing_terms) > 0: df = existing_terms.df code_col = "code" if "code" in df.columns else df.columns[0] existing_codes = list(df[code_col]) else: existing_codes = [] if "PHOTO" in existing_codes: return json.dumps({"action": "already_exists"}) term = o.new_term( code="PHOTO", vocabularyCode=MEASUREMENT_TYPE_VOCABULARY, label="Photo", description="Camera photo of the sample or setup", ) term.save() return json.dumps({"action": "created"}) finally: o.logout() def get_property_assignments(url, object_type=None, token=None, username=None, password=None, verify_certificates=True): """Read-only: dumps the complete current property structure of an object type -- code, data type, section, order, mandatory -- exactly as openBIS has it right now. Use this as ground truth before designing a new/modified object type, rather than working from memory of an old export. Makes no changes.""" o = _connect(url, token=token, username=username, password=password, verify_certificates=verify_certificates) object_type = object_type or OBJECT_TYPE try: sample_type = o.get_object_type(object_type) pas = sample_type.get_property_assignments() df = pas.df if hasattr(pas, "df") else DataFrame(pas) cols = [c for c in ["code", "dataType", "section", "ordinal", "mandatory"] if c in df.columns] df = df[cols].sort_values("ordinal") if "ordinal" in df.columns else df[cols] return df.to_json(orient="records") finally: o.logout() def property_usage_report(url, object_type=None, token=None, username=None, password=None, verify_certificates=True): """Read-only report: for every property on the given object type, counts how many EXISTING entries have a non-empty value for it. Run this BEFORE removing or changing a property in openBIS's schema, so you know how much real data would actually be affected rather than guessing. Makes no changes -- purely a search + count. The property list is read live via get_property_assignments() rather than hardcoded, so this works unmodified for any object type -- part of what makes this file reusable as a template. Returns a JSON string: {"total_entries": N, "property_fill_counts": {"temp_reading": 12, "cell_pressure_real": 12, ...}} -- a property absent from the dict (or at 0) currently has no data anywhere, so removing it is safe; anything with a high count is exactly what you'd lose.""" o = _connect(url, token=token, username=username, password=password, verify_certificates=verify_certificates) object_type = object_type or OBJECT_TYPE try: sample_type = o.get_object_type(object_type) pas = sample_type.get_property_assignments() pa_df = pas.df if hasattr(pas, "df") else DataFrame(pas) all_props = list(pa_df["code"]) if "code" in pa_df.columns else [] df = o.get_objects(type=object_type, props=all_props) total = len(df) counts = {} for col in all_props: if col not in df.columns: continue non_empty = df[col].apply( lambda v: v is not None and v == v and v != "" # v==v excludes NaN ).sum() if non_empty > 0: counts[col] = int(non_empty) return json.dumps({"total_entries": int(total), "property_fill_counts": counts}) finally: o.logout() def sync_rows(rows_json, url, space, collection=None, object_type=None, campaign_props_json=None, parent_identifier=None, upload_files=True, token=None, username=None, password=None, verify_certificates=True): """ rows_json -- JSON string: array of objects, one per table row (built in MATLAB via jsonencode(table2struct(T))). Each row must include 'name' and 'folder' if upload_files=True, and 'resultImages' (a list of local PNG paths, or an empty list) if you want plots embedded in the Results field. Columns absent from a given table (e.g. a beamtime without smplX_mm/Y/Z) are simply skipped -- nothing needs to exist in every row. url -- openBIS instance URL space -- openBIS space code to create/find objects in collection -- openBIS experiment/collection identifier, e.g. '/SPACE/PROJECT/COLLECTION' (optional) object_type -- openBIS object type code (falls back to OBJECT_TYPE) campaign_props_json -- JSON string: {"campaignLocation": "...", "team": "...", "projectName": "..."} -- same value applied to every row this run (falls back to skipping campaign properties if not provided) parent_identifier -- identifier (e.g. '/SPACE/BEAMTIME_CODE') of a parent object representing the beamtime as a whole. Applied to every row's object via add_parents() (see the comment at that call for why not set_parents()/ get_parents()). upload_files -- if True, attaches the original raw file (from fullfile(folder,name)) as a RAW_DATASET_TYPE dataset on each object token / username+password -- authentication; provide one or the other Update behaviour: an existing object is matched by its $name property (not Code, which openBIS auto-generates for this type). Any property that already has a non-empty value on an existing object is left untouched -- automated sync only fills in blanks, never overwrites something entered by hand (0/0.0 counts as a real value, not empty). A property that can't be set (e.g. a vocabulary term that doesn't exist yet) is skipped and reported in that row's message, without preventing the rest of the row from saving. Returns a JSON string: one summary entry per row -- [{"name": "...", "action": "created"|"updated"|"error", "message": "..."}] """ rows = json.loads(rows_json) object_type = _clean_str_or_none(object_type) or OBJECT_TYPE campaign_props = json.loads(campaign_props_json) if campaign_props_json else {} url = _clean_str_or_none(url) space = _clean_str_or_none(space) collection = _clean_str_or_none(collection) parent_identifier = _clean_str_or_none(parent_identifier) o = _connect(url, token=token, username=username, password=password, verify_certificates=verify_certificates) results = [] # Resolve the parent identifier to an actual object once, up front, # rather than handing pybis a bare string to resolve internally on # every row. Confirmed via a real traceback that passing the string # directly works fine at object CREATION time but breaks UPDATE # (pybis/attribute.py's internal string->object resolution ends up # leaving a raw string where save() later expects a fully-resolved # identifier dict, causing "string indices must be integers, not # 'str'"). Passing an already-resolved object sidesteps that path. parent_obj = None parent_warning = None if parent_identifier: try: parent_obj = o.get_object(parent_identifier) except Exception as e: parent_warning = f"could not resolve parent_identifier '{parent_identifier}': {e}" def _is_empty(value): """Deliberately not `not value` -- that would treat a genuine 0 or 0.0 (a perfectly valid REAL/INTEGER reading, e.g. a position or pressure) as 'empty' and overwrite it. Only None and '' count.""" return value is None or value == "" def _current_value(obj, prop_code): try: return obj.props[prop_code] except Exception: return None def _set_if_empty(obj, prop_code, value, warnings): """Sets a property only if currently empty. Any failure (e.g. an unrecognized property code, or a vocabulary term that doesn't exist yet) is recorded as a warning rather than raised, so it can't prevent the rest of the row's properties from saving. prop_code is force-lowercased -- confirmed (via a real openBIS error listing every valid code) that property codes are lowercase at the wire level even though the admin UI/xlsx export displays them uppercase.""" prop_code = prop_code.lower() if _is_empty(value): return try: if _is_empty(_current_value(obj, prop_code)): obj.props[prop_code] = value except Exception as e: warnings.append(f"{prop_code}: {e}") try: for row in rows: name = row.get("name", "") warnings = [] if parent_warning: warnings.append(parent_warning) try: category = _category_for_name(name) existing = o.get_objects(type=object_type, space=space, **{"$name": name}) if len(existing) > 0: obj = existing[0] action = "updated" else: new_kwargs = dict(type=object_type, space=space, props={"$name": name}) if collection: # Only included when truthy: explicitly passing # collection=None causes pybis to emit a malformed # empty "experimentId": {} in the request (missing # its required @type discriminator), which the # server then rejects outright -- confirmed via a # real server error. Omitting the key entirely # avoids that path. new_kwargs["collection"] = collection if parent_obj: new_kwargs["parents"] = [parent_obj] obj = o.new_object(**new_kwargs) action = "created" if parent_obj and action == "updated": try: # add_parents() (not set_parents(), and NOT # preceded by a get_parents() check): get_parents() # has a confirmed bug where it populates the # internal _parents cache as a flat list of plain # identifier strings (via a DataFrame # .values.flatten()) -- a different shape than # add_parents()/set_parents() produce. Merely # calling it corrupts state that obj.save() later # can't handle ("string indices must be integers, # not 'str'"), even on rows that never touch # parents at all this run. add_parents() sidesteps # that path entirely, and unlike set_parents() it # only adds our parent rather than replacing the # whole list -- so it can't remove a different # parent someone added by hand. obj.add_parents([parent_obj]) except Exception as e: warnings.append(f"parent link: {e}") for col, prop in DIRECT_MAP.items(): if col in row: _set_if_empty(obj, prop, row[col], warnings) for col, (prop, kind) in NUMERIC_MAP.items(): if col in row: value = row[col] if kind == "int": value = _to_clean_int(value) _set_if_empty(obj, prop, value, warnings) for prop, value in _row_derived_props(row, category).items(): _set_if_empty(obj, prop, value, warnings) for prop, campaign_key in CAMPAIGN_MAP.items(): _set_if_empty(obj, prop, campaign_props.get(campaign_key), warnings) for prop, value in CONSTANT_MAP.items(): _set_if_empty(obj, prop, value, warnings) for img_path in (row.get("resultImages") or []): try: attach_result_image(o, obj, img_path, url, verify_certificates=verify_certificates) except Exception as e: warnings.append(f"result image ({img_path}): {e}") obj.save() if upload_files and row.get("folder"): src_path = f"{row['folder']}/{name}" try: ds = o.new_dataset(type=RAW_DATASET_TYPE, sample=obj, files=[src_path]) ds.save() except Exception as e: warnings.append(f"file upload: {e}") results.append({"name": name, "action": action, "message": "; ".join(warnings)}) except Exception as e: results.append({"name": name, "action": "error", "message": str(e)}) finally: o.logout() return json.dumps(results) def check_connection(url, token=None, username=None, password=None, verify_certificates=True): """Quick standalone connectivity/auth check -- call this before sync_rows on a full table, to confirm login/token works. Deliberately does NOT use is_session_active()/isSessionActive: that RPC has been observed rejecting valid PATs on this openBIS instance even though the same token works fine for real operations (get_spaces here, and get_objects/new_object/etc. in sync_rows). Space listing is used instead as a cheap, harmless, real call that actually proves the token/login works end to end.""" o = _connect(url, token=token, username=username, password=password, verify_certificates=verify_certificates) try: spaces = o.get_spaces() working = spaces is not None except Exception as e: return json.dumps({"connection_works": False, "message": str(e)}) finally: try: o.logout() except Exception: pass # harmless either way; not the check we care about here return json.dumps({"connection_works": working, "message": ""})