admin/Odoo-Modules
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(jobs): add fp.job._generate_steps_from_recipe (Task 2.4)

Browse Source 
Native port of fusion_plating_bridge_mrp's
_generate_workorders_from_recipe method. Walks the recipe tree,
creates one fp.job.step per 'operation' node, formats child 'step'
nodes as step instructions on chatter, respects opt-in/out
overrides from fp.job.node.override.

Adaptations from the original:
- Creates fp.job.step (not mrp.workorder)
- Maps fusion.plating.work.center to fp.work.centre via forward
  link (x_fc_fp_work_centre_id) or code fallback
- Uses native field names (job_id, work_centre_id, etc.)
- Drops work_role_id (not on fp.job.step yet — Task 2.6+)
- Drops _fp_autofill_default_equipment (not yet on step)

5 new tests cover: basic generation, idempotency, no-recipe skip,
opt-in override behaviour, recipe_node_id link.

Manifest 19.0.1.2.0 → 19.0.1.3.0.

Part of: native job model migration (spec 2026-04-25)
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
gsinghpal
2026-04-24 23:17:47 -04:00
parent 4c68327b9c
commit 3b7eae9b78
3 changed files with 306 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
fusion_plating/fusion_plating_jobs/__manifest__.py
View File

@@ -3,7 +3,7 @@
# License OPL-1 (Odoo Proprietary License v1.0)
{
'name': 'Fusion Plating — Native Jobs',
'version': '19.0.1.2.0',
'version': '19.0.1.3.0',
'category': 'Manufacturing/Plating',
'summary': 'Native plating job model — replaces mrp.production / mrp.workorder bridge.',
'description': """

199
fusion_plating/fusion_plating_jobs/models/fp_job.py
View File

@@ -10,8 +10,14 @@
# qc_check_id is deferred to Task 2.7 (the underlying QC model still
# lives in fusion_plating_bridge_mrp; we'll address its sourcing then).
import logging
from markupsafe import Markup
from odoo import fields, models
_logger = logging.getLogger(__name__)
class FpJob(models.Model):
_inherit = 'fp.job'
@@ -41,3 +47,196 @@ class FpJob(models.Model):
'job_id',
string='Recipe Overrides',
)
# ------------------------------------------------------------------
# Recipe → fp.job.step generation (Task 2.4)
#
# Native port of fusion_plating_bridge_mrp's
# _generate_workorders_from_recipe. Walks the recipe tree, creates
# one fp.job.step per 'operation' node, formats child 'step' nodes
# as step instructions on chatter, respects opt-in/out overrides
# from fp.job.node.override.
#
# Adaptations from the original:
# - Creates fp.job.step (not mrp.workorder)
# - Maps fusion.plating.work.center → fp.work.centre via code
# fallback (no forward link exists yet)
# - Uses native field names (job_id, work_centre_id, etc.)
# - Drops work_role_id (not on fp.job.step yet — Task 2.6+)
# - Drops _fp_autofill_default_equipment (not yet on step)
# ------------------------------------------------------------------
def _generate_steps_from_recipe(self):
"""Generate fp.job.step records from the assigned recipe.
Walks the recipe tree, creates one step per 'operation' node,
and formats child 'step' nodes as step instructions on the
chatter. Respects opt-in/out overrides from override_ids.
"""
Step = self.env['fp.job.step']
Node = self.env['fusion.plating.process.node']
for job in self:
if not job.recipe_id:
continue # No recipe assigned
if job.step_ids:
continue # Steps already exist — don't duplicate
# Build lookup of overrides keyed by node ID
override_map = {ov.node_id.id: ov.included for ov in job.override_ids}
# Start-at-node: if set, the allowed set is the union of:
# 1. start_node and all its descendants
# 2. each ancestor of start_node
# 3. at each ancestor level, any LATER-sequence sibling and
# all of its descendants
start_node = job.start_at_node_id
allowed_ids = None # None = include everything
if start_node:
descendants = Node.search([('id', 'child_of', start_node.id)])
allowed_ids = set(descendants.ids)
cur = start_node
while cur.parent_id:
parent = cur.parent_id
allowed_ids.add(parent.id)
later_sibs = parent.child_ids.filtered(
lambda n: n.sequence > cur.sequence
)
for sib in later_sibs:
sib_descendants = Node.search([
('id', 'child_of', sib.id),
])
allowed_ids |= set(sib_descendants.ids)
cur = parent
step_vals_list = []
wo_steps = {} # {sequence: instruction text}
seq_counter = [10]
def _is_node_included(node):
"""Determine if a node should be included based on
opt-in/out logic, per-job overrides, and start-at-node
filter.
"""
nid = node.id
if allowed_ids is not None and nid not in allowed_ids:
return False
opt = node.opt_in_out or 'disabled'
if opt == 'disabled':
return True
if nid in override_map:
return override_map[nid]
if opt == 'opt_in':
return False # Default excluded
return True # opt_out → default included
def _resolve_work_centre(legacy_wc):
"""Map fusion.plating.work.center → fp.work.centre.
The legacy work-centre model does not (yet) have a forward
link to the new fp.work.centre. Try a forward link
(x_fc_fp_work_centre_id) if some bridge module added one;
otherwise fall back to a code lookup.
"""
if not legacy_wc:
return self.env['fp.work.centre']
# Forward link, if any
if (
'x_fc_fp_work_centre_id' in legacy_wc._fields
and legacy_wc.x_fc_fp_work_centre_id
):
return legacy_wc.x_fc_fp_work_centre_id
# Code fallback (legacy code is unique-per-facility,
# native code is globally unique — first match wins)
if legacy_wc.code:
found = self.env['fp.work.centre'].search(
[('code', '=', legacy_wc.code)], limit=1,
)
if found:
return found
return self.env['fp.work.centre']
def walk_node(node):
if not _is_node_included(node):
return
if node.node_type == 'operation':
work_centre = _resolve_work_centre(node.work_center_id)
if not work_centre:
_logger.warning(
'Job %s: operation "%s" has no mapped fp.work.centre — '
'creating step without work centre.',
job.name, node.name,
)
# Collect step instructions from child 'step' nodes
instructions = []
step_num = 1
for child in node.child_ids.sorted('sequence'):
if child.node_type == 'step' and _is_node_included(child):
line = '%d. %s' % (step_num, child.name)
if child.estimated_duration:
line += ' (%.0f min)' % child.estimated_duration
instructions.append(line)
step_num += 1
vals = {
'job_id': job.id,
'name': node.name,
'work_centre_id': work_centre.id if work_centre else False,
'duration_expected': node.estimated_duration or 0.0,
'sequence': seq_counter[0],
'recipe_node_id': node.id,
}
if node.estimated_duration:
vals['dwell_time_minutes'] = node.estimated_duration
# Pull thickness target from the coating config when
# this is a plating step (matched by node name keyword).
coating = job.coating_config_id
name_l = (node.name or '').lower()
is_plating_node = (
'plat' in name_l or 'nickel' in name_l
or 'chrome' in name_l or 'anodiz' in name_l
)
if coating and is_plating_node:
if (
'thickness_max' in coating._fields
and coating.thickness_max
):
vals['thickness_target'] = coating.thickness_max
if (
'thickness_uom' in coating._fields
and coating.thickness_uom
):
vals['thickness_uom'] = coating.thickness_uom
step_vals_list.append(vals)
if instructions:
wo_steps[seq_counter[0]] = '\n'.join(instructions)
seq_counter[0] += 10
elif node.node_type in ('recipe', 'sub_process'):
for child in node.child_ids.sorted('sequence'):
walk_node(child)
# 'step' nodes at top level are handled by their parent operation
# Walk from recipe root
walk_node(job.recipe_id)
# Bulk create
if step_vals_list:
created = Step.create(step_vals_list)
for step in created:
instr_text = wo_steps.get(step.sequence)
if instr_text:
step.message_post(
body=Markup(
'<b>Recipe steps:</b><br/><pre>%s</pre>'
) % instr_text,
subtype_xmlid='mail.mt_note',
)
job.message_post(
body=('%d steps generated from recipe "%s".') % (
len(step_vals_list), job.recipe_id.name,
),
)
return True

106
fusion_plating/fusion_plating_jobs/tests/test_fp_job_extensions.py
View File

@@ -154,3 +154,109 @@ class TestFpJobNodeOverride(TransactionCase):
})
self.job.invalidate_recordset(['override_ids'])
self.assertIn(ovr, self.job.override_ids)
class TestFpJobStepsGenerator(TransactionCase):
def setUp(self):
super().setUp()
self.partner = self.env['res.partner'].create({'name': 'C'})
self.product = self.env['product.product'].create({'name': 'W'})
self.wc = self.env['fp.work.centre'].create({
'name': 'Bath', 'code': 'BATH', 'kind': 'wet_line',
})
# Build a simple recipe: recipe → 2 operations + 1 opt-in op
self.recipe = self.env['fusion.plating.process.node'].create({
'name': 'TestRecipe',
'node_type': 'recipe',
})
# Legacy work centre (recipe nodes still point at the legacy model).
# Match the new fp.work.centre.code so the resolver picks it up.
facility = self.env['fusion.plating.facility'].search([], limit=1)
if not facility:
facility = self.env['fusion.plating.facility'].create({
'name': 'TestFacility',
'code': 'TF',
})
legacy_wc = self.env['fusion.plating.work.center'].search(
[('code', '=', 'BATH')], limit=1)
if not legacy_wc:
legacy_wc = self.env['fusion.plating.work.center'].create({
'name': 'Bath',
'code': 'BATH',
'facility_id': facility.id,
})
self.legacy_wc = legacy_wc
self.op1 = self.env['fusion.plating.process.node'].create({
'name': 'Plating Bath',
'node_type': 'operation',
'parent_id': self.recipe.id,
'sequence': 10,
'estimated_duration': 30.0,
'work_center_id': self.legacy_wc.id,
})
self.op2 = self.env['fusion.plating.process.node'].create({
'name': 'Bake',
'node_type': 'operation',
'parent_id': self.recipe.id,
'sequence': 20,
'estimated_duration': 60.0,
})
self.opt_in = self.env['fusion.plating.process.node'].create({
'name': 'Optional Inspect',
'node_type': 'operation',
'parent_id': self.recipe.id,
'sequence': 30,
'opt_in_out': 'opt_in',
})
def _make_job(self, **kw):
vals = {
'partner_id': self.partner.id,
'product_id': self.product.id,
'qty': 1.0,
'recipe_id': self.recipe.id,
}
vals.update(kw)
return self.env['fp.job'].create(vals)
def test_generator_creates_steps(self):
job = self._make_job()
job._generate_steps_from_recipe()
# 2 ops by default; opt_in skipped without an override
self.assertEqual(len(job.step_ids), 2)
def test_generator_idempotent(self):
job = self._make_job()
job._generate_steps_from_recipe()
first_count = len(job.step_ids)
job._generate_steps_from_recipe()
self.assertEqual(len(job.step_ids), first_count)
def test_generator_skips_no_recipe(self):
job = self.env['fp.job'].create({
'partner_id': self.partner.id,
'product_id': self.product.id,
'qty': 1.0,
})
job._generate_steps_from_recipe()
self.assertFalse(job.step_ids)
def test_generator_respects_opt_in_override(self):
job = self._make_job()
self.env['fp.job.node.override'].create({
'job_id': job.id,
'node_id': self.opt_in.id,
'included': True,
})
job._generate_steps_from_recipe()
# 3 steps: 2 default + 1 opted-in
self.assertEqual(len(job.step_ids), 3)
def test_generator_recipe_node_link(self):
job = self._make_job()
job._generate_steps_from_recipe()
first_step = job.step_ids.sorted('sequence')[0]
self.assertEqual(first_step.recipe_node_id, self.op1)
self.assertEqual(first_step.duration_expected, 30.0)
# Work centre resolved by code from legacy model
self.assertEqual(first_step.work_centre_id, self.wc)